Aegis/docs/superpowers/plans/2026-07-06-red-blue-review-workflow.md

Red/Blue Review Workflow Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking. The user has explicitly asked for autonomous execution — do NOT pause to ask clarifying questions between tasks; use the documented assumptions below and keep moving. Only escalate (BLOCKED status) for a genuine implementation blocker, not a design ambiguity — all design ambiguities are resolved below.

Goal: Insert a lead-review gate on each side of the test workflow (red_review after red execution, blue_review after blue evaluation), with load-balanced auto-assignment of the reviewing lead (Aegis + Jira in sync), and make each team blind to the other team's work until both reviews are approved.

Architecture: Two new TestState values inserted into the existing state machine, owned by the authoritative TestEntity domain class (not a shadow/unused entity — this one is live). Reviewer selection is a small load-balancing query (fewest tests currently in that review state). Blind visibility is enforced by masking fields on the GET /tests/{id} response based on viewer role + test state, not by hiding UI alone.

Tech Stack: FastAPI + SQLAlchemy + Alembic (backend), React 19 + TypeScript + TanStack Query (frontend), pytest.

---

Context for the implementing engineer

• TestEntity (backend/app/domain/test_entity.py) is the REAL, live state machine — unlike a similar-looking entity in a previous feature, this one is actually called by every workflow function in backend/app/services/test_workflow_service.py. Get transitions right here or the whole test lifecycle breaks. Run pytest tests/test_test_entity.py tests/test_workflow.py -v after every change to this file.
• Read backend/app/services/test_workflow_service.py fully before starting — it already has _create_phase_worklog, Jira push, notification patterns you'll extend, not replace.
• Read backend/app/routers/tests.py fully before starting — new endpoints go right after submit-blue/start-blue-work and follow the exact same pattern (crud_get_test_or_raise → assignee-lock check if applicable → UnitOfWork → workflow function → uow.commit() → db.refresh → return).
• Backend tests use fixtures from backend/tests/conftest.py (client, db, per-role user/headers fixtures including red_lead_headers, blue_lead_headers; a shared api(client) fixture already exists there that clears cookies before each request — use it for ANY test that mixes more than one role's *_headers fixture in a single test, per a hard-learned lesson from a previous feature: TestClient's cookie jar persists across requests within a test, and login sets an aegis_token cookie that get_current_user prefers over the Authorization header, so mixing roles without api(...) silently authenticates the wrong role).
• Frontend: Node.js/npm is NOT installed in this environment. Verify frontend changes by careful manual reading (import resolution, JSX tag balance, type correctness) — do not attempt npm run build.
• Run cd backend && ruff check app/ tests/ after backend changes.

Resolved design decisions (do not re-ask the user about these)

1. New states: red_review (after red submits, before blue queue) and blue_review (after blue submits, before cross-validation). Full sequence: draft → red_executing → red_review → blue_evaluating → blue_review → in_review → validated/rejected/disputed.
2. Reopen destinations: red_review reopens to red_executing. blue_review reopens to blue_evaluating.
3. Self-review ban: if the person submitting (the actual API caller, not necessarily the assignee) holds the reviewing lead role themselves (e.g. a red_lead executed the test personally and calls submit-red), they are excluded from the reviewer candidate pool — someone else must review. If no other candidate exists, the submit call fails with a clear BusinessRuleViolation (no silent self-review, no crash).
4. Load balancing metric: count of tests currently sitting in red_review (or blue_review) with red_reviewer_assignee (or blue_reviewer_assignee) equal to the candidate — Aegis-internal count only, not Jira ticket count. Tie-break by username ascending (deterministic, no randomness).
5. Jira sync: the Jira ticket is reassigned to the SAME lead Aegis selected as reviewer (not a generic reassignment).
6. Blue "gap" outcome: when the blue lead flags a system_gaps issue, the test proceeds forward to in_review (same as approve) — it does NOT return to the blue operator, since a missing capability isn't something a retry fixes. system_gaps is a plain Text field on Test, independent of the existing detection_lifecycle/decay_policy subsystem (confirmed unrelated after investigation — linking them would be overengineering).
7. Blind visibility scope: end-to-end. From draft through blue_review (inclusive), a red-side viewer (red_tech/red_lead) never sees blue fields and a blue-side viewer (blue_tech/blue_lead) never sees red fields, via the API response itself (not just hidden UI — the data must not leave the server). admin and viewer roles are never blinded (they aren't executing either side, so blinding serves no purpose for them). Once test.state is in_review, validated, rejected, or disputed, everyone sees everything (today's existing cross-validation UX, unchanged).
8. Reviewer-only action: only the specific assigned reviewer (test.red_reviewer_assignee / test.blue_reviewer_assignee) or admin may call the review-decision endpoint — not just "any red_lead".
9. "Queued Blue Team" label: this is a DISPLAY-ONLY label, not a new backend state. blue_evaluating + blue_work_started_at IS NULL = "Queued Blue Team" in the UI; blue_evaluating + blue_work_started_at IS NOT NULL = "Blue Evaluating". No backend change needed for this beyond what already exists — confirmed start-blue-work already sets blue_tech_assignee on pickup (a prior research pass on this feature incorrectly reported this as missing; the actual router code at tests.py around the start_blue_work endpoint already does if test.blue_tech_assignee is None: test.blue_tech_assignee = current_user.id — verify this is still true when you read the file, but do not "fix" something that already works).
10. Timer/worklog boundaries: the red-phase worklog is created at submit-red time (red_executing → red_review) — the review wait time is the LEAD's concern, not counted against the operator. blue_started_at (queue-entry timestamp, used for the old "how long did this wait" metric) is set only once red_review is APPROVED (red_review → blue_evaluating), not at submit time — this is when the test actually becomes visible/available to blue team. Symmetric logic for blue: the blue-phase worklog is created at submit-blue time (blue_evaluating → blue_review); blue_review approval does not set any new "queue" timestamp since the next stop (in_review) doesn't have one today.
11. Reopen timer reset: on reopen_red_review, reset test.red_started_at = now() (fresh timer for attempt #2) — the first attempt's worklog was already recorded at submit time, so this doesn't lose data. On reopen_blue_review, reset test.blue_work_started_at = None (so the operator must click "Start Evaluation" again, consistent with the existing pickup mechanic) and test.blue_started_at = now() (fresh queue-entry timestamp for attempt #2). Keep blue_tech_assignee/red_tech_assignee unchanged on reopen (same operator continues, no need to re-claim).

---

Task 1: Domain — TestState enum, TestEntity transitions, VALID_TRANSITIONS

Files:

• Modify: backend/app/domain/enums.py

• Modify: backend/app/domain/test_entity.py

• Modify: backend/app/models/enums.py (re-export check — likely no change needed, just verify)

• Test: backend/tests/test_test_entity.py

• Step 1: Read the existing test file to match its style

Read backend/tests/test_test_entity.py in full first — it already has helper fixtures/patterns for constructing a TestEntity and asserting transitions/exceptions. Match that exact style for your new tests (do not invent a different helper pattern).

• Step 2: Write the failing tests

Add tests (using whatever entity-construction helper the existing file already provides) covering:

• submit_red_evidence() now transitions to red_review, not blue_evaluating.

• New method approve_red_review(): red_review → blue_evaluating; raises InvalidStateTransition if called from any other state.

• New method reopen_red_review(notes=...): red_review → red_executing; stores nothing on the entity itself (notes are a service-layer/ORM concern per Task 5, not stored as an entity field) but does reset red_started_at to a fresh timestamp and clears red_paused_seconds back to 0 (mirroring how submit_red_evidence already resets blue_paused_seconds to 0 on its own transition — same "fresh start for the next phase" pattern). Raises InvalidStateTransition if called from any other state.

• submit_blue_evidence() now transitions to blue_review, not in_review.

• New method approve_blue_review(): blue_review → in_review; raises InvalidStateTransition if called from any other state.

• New method reopen_blue_review(): blue_review → blue_evaluating; resets blue_paused_seconds to 0. Raises InvalidStateTransition if called from any other state.

• New method flag_blue_review_gap(): blue_review → in_review (same target as approve — this is a second way to reach the same state, both valid from blue_review). Raises InvalidStateTransition if called from any other state.

• _assert_in_review (used by validate_red/validate_blue) still only accepts in_review/disputed — confirm validate_red/validate_blue still raise InvalidOperationError if called while the test is sitting in the NEW red_review/blue_review states (they must not be reachable from there — only from in_review/disputed).

• Step 3: Run tests to verify they fail

Run: cd backend && pytest tests/test_test_entity.py -v -k "review" Expected: FAIL — AttributeError for the new methods/states not existing yet.

• Step 4: Implement

In backend/app/domain/enums.py, add two members to TestState, inserted in lifecycle order:

class TestState(str, enum.Enum):
    """Lifecycle states in the security test state machine."""

    draft = "draft"
    red_executing = "red_executing"
    red_review = "red_review"
    blue_evaluating = "blue_evaluating"
    blue_review = "blue_review"
    in_review = "in_review"
    validated = "validated"
    rejected = "rejected"
    disputed = "disputed"   # one lead approved, the other rejected

In backend/app/domain/test_entity.py:

1. Add the same two members to the LOCAL TestState enum in this file (it's a duplicate mirror of the domain enum, not an import — keep that existing pattern, don't refactor it to import from app.domain.enums as that's out of scope for this task).

2. Update VALID_TRANSITIONS:

VALID_TRANSITIONS: dict[TestState, list[TestState]] = {
    TestState.draft:           [TestState.red_executing],
    TestState.red_executing:   [TestState.red_review],
    TestState.red_review:      [TestState.blue_evaluating, TestState.red_executing],
    TestState.blue_evaluating: [TestState.blue_review],
    TestState.blue_review:     [TestState.in_review, TestState.blue_evaluating],
    TestState.in_review:       [TestState.validated, TestState.rejected, TestState.disputed],
    TestState.disputed:        [TestState.validated, TestState.rejected],
    TestState.rejected:        [TestState.draft],
    TestState.validated:       [],
}

3. Update _PAUSABLE_STATES — leave unchanged (red_executing, blue_evaluating only; red_review/blue_review are lead-review states, not operator-working states, so no timer runs there).

4. Change submit_red_evidence()'s transition target from TestState.blue_evaluating to TestState.red_review, and REMOVE the lines that set self.blue_started_at/self.blue_paused_seconds from this method (that now happens in approve_red_review, not here — see Task 11's design note #10 above). The event name and paused-seconds-return behavior stay the same:

    def submit_red_evidence(self) -> int:
        """Transition the test from ``red_executing`` to ``red_review``.

        Auto-resumes if paused.  Returns paused seconds accumulated
        during this phase (for worklog calculation).
        """
        paused_extra = self._auto_resume()
        self._transition(TestState.red_review)
        total_paused = self.red_paused_seconds + paused_extra
        self._events.append(DomainEvent(
            "red_evidence_submitted",
            {"red_paused_seconds": total_paused},
        ))
        return total_paused

5. Add approve_red_review() and reopen_red_review() right after submit_red_evidence():

    def approve_red_review(self) -> None:
        """Transition the test from ``red_review`` to ``blue_evaluating``.

        Called when the assigned Red Lead approves the operator's work.
        Starts the Blue Team queue timer.
        """
        self._transition(TestState.blue_evaluating)
        self.blue_started_at = datetime.utcnow()
        self.blue_paused_seconds = 0
        self._events.append(DomainEvent("red_review_approved"))

    def reopen_red_review(self) -> None:
        """Transition the test from ``red_review`` back to ``red_executing``.

        Called when the assigned Red Lead sends the work back for rework.
        Resets the red-phase timer for a fresh attempt.
        """
        self._transition(TestState.red_executing)
        self.red_started_at = datetime.utcnow()
        self.red_paused_seconds = 0
        self._events.append(DomainEvent("red_review_reopened"))

6. Change submit_blue_evidence()'s transition target from TestState.in_review to TestState.blue_review (no other change to this method's body — it doesn't touch blue_started_at/blue_paused_seconds today, leave as-is):

    def submit_blue_evidence(self) -> int:
        """Transition the test from ``blue_evaluating`` to ``blue_review``.

        Auto-resumes if paused.  Returns paused seconds accumulated
        during this phase (for worklog calculation).
        """
        paused_extra = self._auto_resume()
        self._transition(TestState.blue_review)
        total_paused = self.blue_paused_seconds + paused_extra
        self._events.append(DomainEvent(
            "blue_evidence_submitted",
            {"blue_paused_seconds": total_paused},
        ))
        return total_paused

7. Add approve_blue_review(), reopen_blue_review(), flag_blue_review_gap() right after submit_blue_evidence():

    def approve_blue_review(self) -> None:
        """Transition the test from ``blue_review`` to ``in_review``.

        Called when the assigned Blue Lead approves the operator's work.
        """
        self._transition(TestState.in_review)
        self._events.append(DomainEvent("blue_review_approved"))

    def reopen_blue_review(self) -> None:
        """Transition the test from ``blue_review`` back to ``blue_evaluating``.

        Called when the assigned Blue Lead sends the work back for rework.
        Resets the blue-phase pickup state so the operator must claim it
        again (mirrors the existing "Start Evaluation" pickup mechanic).
        """
        self._transition(TestState.blue_evaluating)
        self.blue_started_at = datetime.utcnow()
        self.blue_paused_seconds = 0
        self._events.append(DomainEvent("blue_review_reopened"))

    def flag_blue_review_gap(self) -> None:
        """Transition the test from ``blue_review`` to ``in_review``.

        Called when the assigned Blue Lead determines the shortfall is a
        missing capability (tooling/visibility), not operator error — the
        test still proceeds to cross-validation since a retry can't fix a
        capability gap. The actual gap description is stored separately
        on the ORM model (``Test.system_gaps``), not on this entity.
        """
        self._transition(TestState.in_review)
        self._events.append(DomainEvent("blue_review_gap_flagged"))

Note: reopen_blue_review resets blue_work_started_at too, per design decision #11 — but blue_work_started_at is NOT a field on TestEntity today (it lives only on the ORM Test model, set directly by the router in start_blue_work, never touched by the entity). Do NOT add blue_work_started_at to the entity's dataclass fields or from_orm/apply_to methods — that's out of scope and would be a bigger refactor than this task needs. Instead, Task 5's service-layer reopen_blue_review function will reset test.blue_work_started_at = None directly on the ORM object, alongside calling entity.reopen_blue_review() for the state fields the entity DOES own. Document this split responsibility with a one-line comment in Task 5's implementation.

• Step 5: Run tests to verify they pass

Run: pytest tests/test_test_entity.py -v Expected: PASS — all new and pre-existing tests green.

• Step 6: Run the broader entity-dependent test suite

Run: pytest tests/test_workflow.py -v Expected: LIKELY FAILURES — test_workflow.py tests the OLD two-state-shorter pipeline (submit_red going straight to blue_evaluating, submit_blue going straight to in_review). This is expected and will be fixed in Task 5 (service layer) and Task 6 (router), which touch the same call chain these tests exercise. Do NOT try to fix test_workflow.py in this task — just confirm the failures are exactly the ones you'd expect from the state-machine change (tests asserting test.state == TestState.blue_evaluating right after a red submit, etc.), not some unrelated breakage. Note the exact failing test names in your report so Task 5's implementer knows what to expect and fix.

• Step 7: Commit

git add backend/app/domain/enums.py backend/app/domain/test_entity.py backend/tests/test_test_entity.py
git commit -m "feat(tests): add red_review/blue_review states to test state machine"

---

Task 2: Model — new Test columns

Files:

• Modify: backend/app/models/test.py

• Step 1: Add the new columns

In backend/app/models/test.py, add right after the existing red_tech_assignee/blue_tech_assignee block (the "Assignment fields" section):

    # ── Review assignment fields ────────────────────────────────────
    red_reviewer_assignee = Column(UUID(as_uuid=True), ForeignKey("users.id"), nullable=True)
    red_review_by = Column(UUID(as_uuid=True), ForeignKey("users.id"), nullable=True)
    red_review_at = Column(DateTime, nullable=True)
    red_review_notes = Column(Text, nullable=True)

    blue_reviewer_assignee = Column(UUID(as_uuid=True), ForeignKey("users.id"), nullable=True)
    blue_review_by = Column(UUID(as_uuid=True), ForeignKey("users.id"), nullable=True)
    blue_review_at = Column(DateTime, nullable=True)
    blue_review_notes = Column(Text, nullable=True)
    system_gaps = Column(Text, nullable=True)

Add the matching relationships right after the existing red_tech_assigned_user/blue_tech_assigned_user relationships:

    red_reviewer = relationship("User", foreign_keys=[red_reviewer_assignee])
    red_review_actor = relationship("User", foreign_keys=[red_review_by])
    blue_reviewer = relationship("User", foreign_keys=[blue_reviewer_assignee])
    blue_review_actor = relationship("User", foreign_keys=[blue_review_by])

• Step 2: Verify the app imports cleanly

Run: cd backend && python -c "import app.models; print('ok')" Expected: prints ok.

• Step 3: Commit

git add backend/app/models/test.py
git commit -m "feat(tests): add review assignment columns to Test model"

---

Task 3: Migration — new enum values + columns

Files:

• Create: backend/alembic/versions/b057_add_test_review_states.py

• Step 1: Find the current alembic head

Run: cd backend && alembic heads Note the current head revision ID — use it as down_revision (should be b056 per the last feature merged in this codebase, but VERIFY by running the command rather than assuming).

• Step 2: Write the migration

Postgres requires ALTER TYPE ... ADD VALUE to run outside a transaction block in older PG versions; the project's env.py may already run migrations non-transactionally per-migration or you may need op.get_bind().execute(sa.text(...)) inside an autocommit block. Check backend/alembic/versions/b046_add_disputed_test_state.py FIRST — it already added a new TestState enum value (disputed) to this exact Postgres enum (teststate), so it has the exact proven pattern for this database/Alembic setup. Copy its ALTER TYPE approach exactly, don't invent a new one.

"""Add red_review, blue_review states and review-assignment columns to tests.

Revision ID: b057
Revises: <VERIFY WITH `alembic heads` — fill in actual value>
Create Date: 2026-07-06
"""
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql

revision = "b057"
down_revision = "<FILL IN — verified head from Step 1>"
branch_labels = None
depends_on = None


def upgrade() -> None:
    # Match the exact enum-value-add pattern from b046_add_disputed_test_state.py
    op.execute("ALTER TYPE teststate ADD VALUE IF NOT EXISTS 'red_review'")
    op.execute("ALTER TYPE teststate ADD VALUE IF NOT EXISTS 'blue_review'")

    op.add_column("tests", sa.Column("red_reviewer_assignee", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id"), nullable=True))
    op.add_column("tests", sa.Column("red_review_by", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id"), nullable=True))
    op.add_column("tests", sa.Column("red_review_at", sa.DateTime(), nullable=True))
    op.add_column("tests", sa.Column("red_review_notes", sa.Text(), nullable=True))
    op.add_column("tests", sa.Column("blue_reviewer_assignee", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id"), nullable=True))
    op.add_column("tests", sa.Column("blue_review_by", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id"), nullable=True))
    op.add_column("tests", sa.Column("blue_review_at", sa.DateTime(), nullable=True))
    op.add_column("tests", sa.Column("blue_review_notes", sa.Text(), nullable=True))
    op.add_column("tests", sa.Column("system_gaps", sa.Text(), nullable=True))


def downgrade() -> None:
    op.drop_column("tests", "system_gaps")
    op.drop_column("tests", "blue_review_notes")
    op.drop_column("tests", "blue_review_at")
    op.drop_column("tests", "blue_review_by")
    op.drop_column("tests", "blue_reviewer_assignee")
    op.drop_column("tests", "red_review_notes")
    op.drop_column("tests", "red_review_at")
    op.drop_column("tests", "red_review_by")
    op.drop_column("tests", "red_reviewer_assignee")
    # Note: Postgres does not support removing an enum value in a downgrade
    # without recreating the type; this is consistent with how
    # b046_add_disputed_test_state.py's downgrade handles the same limitation
    # (check that file — likely a no-op or documented manual step for the enum
    # part specifically; mirror whatever it does).

Check b046_add_disputed_test_state.py's downgrade() for exactly how it handles (or explicitly punts on) removing an enum value, and mirror that same approach/comment style rather than guessing.

• Step 3: Verify migration chain

Run: cd backend && alembic heads Expected: b057 (head).

• Step 4: Commit

git add backend/alembic/versions/b057_add_test_review_states.py
git commit -m "feat(tests): migration for red_review/blue_review states and review columns"

---

Task 4: Service — reviewer load-balancing + Jira assignee override

Files:

• Modify: backend/app/services/test_workflow_service.py

• Modify: backend/app/services/jira_service.py

• Test: backend/tests/test_workflow.py (new test class, don't touch existing tests in this task)

• Step 1: Write the failing tests

Add to backend/tests/test_workflow.py (check the top of the file for its exact fixture/mocking style — it appears to construct Test/User objects directly and call workflow functions, possibly with @patch for log_action/notifications; match that style exactly rather than the db-fixture integration style used in test_campaign_approval.py from a different feature):

class TestReviewerSelection:
    def test_picks_lead_with_fewest_active_reviews(self, db):
        # Two red_leads; give one an existing red_review test so they have load 1, other has 0
        lead_a = User(username="reda", role="red_lead", hashed_password="x")
        lead_b = User(username="redb", role="red_lead", hashed_password="x")
        db.add_all([lead_a, lead_b])
        db.flush()

        tech = Technique(mitre_id="T1059", name="Cmd", tactic="execution", platforms=["windows"])
        db.add(tech)
        db.flush()

        busy_test = Test(technique_id=tech.id, name="Busy", state=TestState.red_review, red_reviewer_assignee=lead_a.id)
        db.add(busy_test)
        db.commit()

        from app.services.test_workflow_service import select_reviewer
        chosen = select_reviewer(db, role="red_lead")
        assert chosen.id == lead_b.id

    def test_excludes_the_submitter_if_they_are_a_lead(self, db):
        lead_a = User(username="reda2", role="red_lead", hashed_password="x")
        db.add(lead_a)
        db.commit()

        from app.services.test_workflow_service import select_reviewer
        from app.domain.errors import BusinessRuleViolation
        with pytest.raises(BusinessRuleViolation, match="No available"):
            select_reviewer(db, role="red_lead", exclude_user_id=lead_a.id)

    def test_no_candidates_raises_clear_error(self, db):
        from app.services.test_workflow_service import select_reviewer
        from app.domain.errors import BusinessRuleViolation
        with pytest.raises(BusinessRuleViolation, match="No available"):
            select_reviewer(db, role="red_lead")

Adjust imports/fixtures at the top of the test class to match whatever db/model-construction pattern the rest of test_workflow.py actually uses — READ THE FILE FIRST, do not assume the snippet above is copy-paste-exact; it's illustrative of intent (three scenarios: fewest-load wins, self-excluded-with-no-alternative raises, zero-candidates raises), adapt syntax to match the file's real conventions.

• Step 2: Run tests to verify they fail

Run: cd backend && pytest tests/test_workflow.py -v -k ReviewerSelection Expected: FAIL — ImportError for select_reviewer.

• Step 3: Implement select_reviewer

Add to backend/app/services/test_workflow_service.py (near the top, after the imports, before VALID_TRANSITIONS):

from app.domain.errors import BusinessRuleViolation
from app.models.user import User

(Add BusinessRuleViolation to the existing import if app.domain.errors isn't already imported in this file — check first, it may only import InvalidOperationError from app.domain.exceptions, a DIFFERENT module. Verify which module BusinessRuleViolation actually lives in — earlier features in this codebase used app.domain.errors.BusinessRuleViolation; confirm this is still correct by checking that module's contents before importing.)

Add the function near _create_phase_worklog (logically grouped with other cross-cutting helpers):

def select_reviewer(
    db: Session,
    *,
    role: str,
    exclude_user_id: uuid.UUID | None = None,
) -> User:
    """Pick the least-loaded active user with *role* to review a test.

    Load is measured as the count of tests currently sitting in the
    matching review state (``red_review`` for role ``red_lead``,
    ``blue_review`` for role ``blue_lead``) with that user set as the
    reviewer. Ties broken by username for determinism.

    Raises BusinessRuleViolation if no eligible reviewer exists (e.g. the
    only lead is the person who executed the test, or there are no leads
    with this role at all).
    """
    review_state = TestState.red_review if role == "red_lead" else TestState.blue_review
    reviewer_field = Test.red_reviewer_assignee if role == "red_lead" else Test.blue_reviewer_assignee

    candidates_query = db.query(User).filter(User.role == role, User.is_active == True)  # noqa: E712
    if exclude_user_id is not None:
        candidates_query = candidates_query.filter(User.id != exclude_user_id)
    candidates = candidates_query.order_by(User.username).all()

    if not candidates:
        raise BusinessRuleViolation(
            f"No available {role} to review this test (cannot self-review, "
            f"and no other {role} is active)"
        )

    best_user = None
    best_count = None
    for candidate in candidates:
        count = (
            db.query(Test)
            .filter(Test.state == review_state, reviewer_field == candidate.id)
            .count()
        )
        if best_count is None or count < best_count:
            best_user = candidate
            best_count = count

    return best_user

• Step 4: Run tests to verify they pass

Run: pytest tests/test_workflow.py -v -k ReviewerSelection Expected: PASS.

• Step 5: Add the Jira assignee override

In backend/app/services/jira_service.py, modify push_test_event's signature to accept an optional assignee and add a branch for the two new states:

def push_test_event(
    db: Session,
    test: Test,
    actor: User,
    new_state: str,
    *,
    extra: dict | None = None,
    assignee: User | None = None,
) -> None:

Add this block right after the existing if new_state == "red_executing": block (as a sibling elif/new if, do not nest inside it):

        if new_state in ("red_review", "blue_review") and assignee:
            jira_account_id = getattr(assignee, "jira_account_id", None)
            if jira_account_id:
                try:
                    jira.assign_issue(link.jira_issue_key, account_id=jira_account_id)
                    logger.info(
                        "Assigned Jira ticket %s to reviewer account %s",
                        link.jira_issue_key, jira_account_id,
                    )
                except Exception as exc_a:
                    logger.warning(
                        "Could not assign %s to reviewer %s: %s",
                        link.jira_issue_key, jira_account_id, exc_a,
                    )

• Step 6: Run the full backend suite to check for regressions

Run: cd backend && pytest tests/ -q 2>&1 | tail -15 Expected: the pre-existing test_workflow.py failures from Task 1 Step 6 STILL present (not fixed until Task 5) — no NEW failures beyond those plus your new passing ReviewerSelection tests.

• Step 7: Commit

git add backend/app/services/test_workflow_service.py backend/app/services/jira_service.py backend/tests/test_workflow.py
git commit -m "feat(tests): load-balanced reviewer selection and Jira reviewer sync"

---

Task 5: Service — retarget submit-red/submit-blue, add review-decision functions

Files:

• Modify: backend/app/services/test_workflow_service.py

• Test: backend/tests/test_workflow.py

• Step 1: Update the failing tests inherited from Task 1

Find every test in backend/tests/test_workflow.py that currently asserts submit_red_evidence/wf_submit_red results in TestState.blue_evaluating, or submit_blue_evidence/wf_submit_blue results in TestState.in_review. Update those assertions to expect TestState.red_review and TestState.blue_review respectively. Do NOT delete or weaken these tests — they still verify real behavior (evidence-required checks, worklog creation, paused-time handling), just against the new intermediate state.

Add new tests for the four new service functions:

class TestReviewDecisions:
    def test_approve_red_review_moves_to_blue_evaluating(self, db, ...):
        # set up a test in red_review with a red_reviewer_assignee
        # call approve_red_review(db, test, reviewer_user, notes="looks good")
        # assert test.state == TestState.blue_evaluating
        # assert test.red_review_by == reviewer_user.id
        # assert test.red_review_notes == "looks good"
        # assert test.blue_started_at is not None
        ...

    def test_reopen_red_review_requires_notes(self, db, ...):
        # call reopen_red_review(db, test, reviewer_user, notes="") or notes=None
        # assert raises InvalidOperationError (or BusinessRuleViolation — pick
        #   whichever this codebase's existing "required field" checks use;
        #   check test_workflow_service.py's submit_red_evidence for the
        #   precedent — it uses InvalidOperationError for "evidence required")
        ...

    def test_reopen_red_review_moves_to_red_executing_with_notes(self, db, ...):
        # call reopen_red_review(db, test, reviewer_user, notes="add more detail")
        # assert test.state == TestState.red_executing
        # assert test.red_review_notes == "add more detail"
        ...

    def test_approve_blue_review_moves_to_in_review(self, db, ...): ...
    def test_reopen_blue_review_requires_notes(self, db, ...): ...
    def test_reopen_blue_review_moves_to_blue_evaluating(self, db, ...): ...
    def test_flag_blue_review_gap_requires_system_gaps_text(self, db, ...):
        # call flag_blue_review_gap(db, test, reviewer_user, system_gaps="")
        # assert raises (empty system_gaps not allowed)
        ...
    def test_flag_blue_review_gap_moves_to_in_review(self, db, ...):
        # call flag_blue_review_gap(db, test, reviewer_user, system_gaps="Missing EDR agent on host X")
        # assert test.state == TestState.in_review
        # assert test.system_gaps == "Missing EDR agent on host X"
        ...

Fill in the ... with real setup code matching this file's existing conventions (read a nearby existing test in the same file for the exact Test/User/mock pattern before writing these — do not guess at fixture names).

• Step 2: Run tests to verify they fail

Run: pytest tests/test_workflow.py -v Expected: FAIL — both the inherited-from-Task-1 assertion mismatches AND ImportError/AttributeError for the new functions.

• Step 3: Retarget submit_red_evidence and submit_blue_evidence

In backend/app/services/test_workflow_service.py:

In submit_red_evidence(db, test, user): change the transition_state(db, test, TestState.blue_evaluating, user, ...) call to TestState.red_review. REMOVE the lines test.blue_started_at = now and test.blue_paused_seconds = 0 from this function (moved to approve_red_review, see below). After the transition and worklog creation, add reviewer selection and assignment:

    test = transition_state(
        db, test, TestState.red_review, user,
        action_name="submit_red_evidence",
    )

    _create_phase_worklog(
        db, test=test, user=user,
        phase_started_at=test.red_started_at,
        phase_ended_at=now,
        paused_seconds=(test.red_paused_seconds or 0) + paused_extra,
        activity_type="red_team_execution",
        description=f"Red Team execution: {test.name}",
    )

    reviewer = select_reviewer(db, role="red_lead", exclude_user_id=user.id if user.role == "red_lead" else None)
    test.red_reviewer_assignee = reviewer.id
    db.flush()

    try:
        create_notification(
            db, user_id=reviewer.id, type="review_assigned",
            title="Test awaiting your review",
            message=f'Test "{test.name}" is waiting for your red-team review.',
            entity_type="test", entity_id=test.id,
        )
    except Exception as e:
        logger.warning("Notification failed for test %s: %s", test.id, e, exc_info=True)

    try:
        from app.services.jira_service import push_test_event
        push_test_event(db, test, user, "red_review", assignee=reviewer)
    except Exception as e:
        logger.warning("Jira push failed for test %s: %s", test.id, e, exc_info=True)

    return test

Remove the OLD push_test_event(db, test, user, "blue_evaluating") call from the end of this function (it's replaced by the "red_review" push above — the blue_evaluating Jira push now happens in approve_red_review instead, added below).

In submit_blue_evidence(db, test, user): change the transition_state(db, test, TestState.in_review, user, ...) call to TestState.blue_review. Do NOT remove any blue_started_at/blue_paused_seconds logic here (this function never touched those). After the worklog creation, add the mirrored reviewer selection:

    test = transition_state(
        db, test, TestState.blue_review, user,
        action_name="submit_blue_evidence",
    )

    _create_phase_worklog(
        db, test=test, user=user,
        phase_started_at=test.blue_work_started_at or test.blue_started_at,
        phase_ended_at=now,
        paused_seconds=(test.blue_paused_seconds or 0) + paused_extra,
        activity_type="blue_team_evaluation",
        description=f"Blue Team evaluation: {test.name}",
    )

    reviewer = select_reviewer(db, role="blue_lead", exclude_user_id=user.id if user.role == "blue_lead" else None)
    test.blue_reviewer_assignee = reviewer.id
    db.flush()

    try:
        create_notification(
            db, user_id=reviewer.id, type="review_assigned",
            title="Test awaiting your review",
            message=f'Test "{test.name}" is waiting for your blue-team review.',
            entity_type="test", entity_id=test.id,
        )
    except Exception as e:
        logger.warning("Notification failed for test %s: %s", test.id, e, exc_info=True)

    try:
        from app.services.jira_service import push_test_event
        push_test_event(db, test, user, "blue_review", assignee=reviewer)
    except Exception as e:
        logger.warning("Jira push failed for test %s: %s", test.id, e, exc_info=True)

    return test

Remove the OLD push_test_event(db, test, user, "in_review") call from the end of this function (replaced by "blue_review" above — the "in_review" push now happens in approve_blue_review/flag_blue_review_gap instead).

• Step 4: Implement the four review-decision functions

Add these new functions right after submit_blue_evidence (before pause_timer):

def approve_red_review(db: Session, test: Test, user: User, notes: str | None = None) -> Test:
    """Red Lead approves the operator's work — moves red_review to blue_evaluating."""
    entity = TestEntity.from_orm(test)
    entity.approve_red_review()
    entity.apply_to(test)
    test.red_review_by = user.id
    test.red_review_at = datetime.utcnow()
    test.red_review_notes = notes
    db.flush()

    log_action(
        db, user_id=user.id, action="approve_red_review",
        entity_type="test", entity_id=test.id,
        details={"notes": notes, "test_name": test.name},
    )

    try:
        notify_test_state_change(db, test, "blue_evaluating")
    except Exception as e:
        logger.warning("Notification failed for test %s: %s", test.id, e, exc_info=True)

    try:
        from app.services.jira_service import push_test_event
        push_test_event(db, test, user, "blue_evaluating")
    except Exception as e:
        logger.warning("Jira push failed for test %s: %s", test.id, e, exc_info=True)

    return test


def reopen_red_review(db: Session, test: Test, user: User, notes: str) -> Test:
    """Red Lead sends the operator's work back for rework — moves red_review to red_executing."""
    if not notes or not notes.strip():
        raise InvalidOperationError("A comment is required when reopening a test for rework")

    entity = TestEntity.from_orm(test)
    entity.reopen_red_review()
    entity.apply_to(test)
    test.red_review_by = user.id
    test.red_review_at = datetime.utcnow()
    test.red_review_notes = notes.strip()
    db.flush()

    log_action(
        db, user_id=user.id, action="reopen_red_review",
        entity_type="test", entity_id=test.id,
        details={"notes": notes, "test_name": test.name},
    )

    if test.red_tech_assignee:
        try:
            create_notification(
                db, user_id=test.red_tech_assignee, type="test_reopened",
                title="Test sent back for rework",
                message=f'Test "{test.name}" was sent back by your Red Lead: {notes.strip()[:200]}',
                entity_type="test", entity_id=test.id,
            )
        except Exception as e:
            logger.warning("Notification failed for test %s: %s", test.id, e, exc_info=True)

    return test


def approve_blue_review(db: Session, test: Test, user: User, notes: str | None = None) -> Test:
    """Blue Lead approves the operator's work — moves blue_review to in_review."""
    entity = TestEntity.from_orm(test)
    entity.approve_blue_review()
    entity.apply_to(test)
    test.blue_review_by = user.id
    test.blue_review_at = datetime.utcnow()
    test.blue_review_notes = notes
    db.flush()

    log_action(
        db, user_id=user.id, action="approve_blue_review",
        entity_type="test", entity_id=test.id,
        details={"notes": notes, "test_name": test.name},
    )

    try:
        notify_test_state_change(db, test, "in_review")
    except Exception as e:
        logger.warning("Notification failed for test %s: %s", test.id, e, exc_info=True)

    try:
        from app.services.jira_service import push_test_event
        push_test_event(db, test, user, "in_review")
    except Exception as e:
        logger.warning("Jira push failed for test %s: %s", test.id, e, exc_info=True)

    return test


def reopen_blue_review(db: Session, test: Test, user: User, notes: str) -> Test:
    """Blue Lead sends the operator's work back for rework — moves blue_review to blue_evaluating."""
    if not notes or not notes.strip():
        raise InvalidOperationError("A comment is required when reopening a test for rework")

    entity = TestEntity.from_orm(test)
    entity.reopen_blue_review()
    entity.apply_to(test)
    test.blue_work_started_at = None  # split responsibility: entity doesn't own this field
    test.blue_review_by = user.id
    test.blue_review_at = datetime.utcnow()
    test.blue_review_notes = notes.strip()
    db.flush()

    log_action(
        db, user_id=user.id, action="reopen_blue_review",
        entity_type="test", entity_id=test.id,
        details={"notes": notes, "test_name": test.name},
    )

    if test.blue_tech_assignee:
        try:
            create_notification(
                db, user_id=test.blue_tech_assignee, type="test_reopened",
                title="Test sent back for rework",
                message=f'Test "{test.name}" was sent back by your Blue Lead: {notes.strip()[:200]}',
                entity_type="test", entity_id=test.id,
            )
        except Exception as e:
            logger.warning("Notification failed for test %s: %s", test.id, e, exc_info=True)

    return test


def flag_blue_review_gap(db: Session, test: Test, user: User, system_gaps: str, notes: str | None = None) -> Test:
    """Blue Lead flags a capability gap — proceeds to in_review anyway (a retry can't fix a missing tool)."""
    if not system_gaps or not system_gaps.strip():
        raise InvalidOperationError("system_gaps description is required when flagging a capability gap")

    entity = TestEntity.from_orm(test)
    entity.flag_blue_review_gap()
    entity.apply_to(test)
    test.blue_review_by = user.id
    test.blue_review_at = datetime.utcnow()
    test.blue_review_notes = notes
    test.system_gaps = system_gaps.strip()
    db.flush()

    log_action(
        db, user_id=user.id, action="flag_blue_review_gap",
        entity_type="test", entity_id=test.id,
        details={"system_gaps": system_gaps, "notes": notes, "test_name": test.name},
    )

    try:
        notify_test_state_change(db, test, "in_review")
    except Exception as e:
        logger.warning("Notification failed for test %s: %s", test.id, e, exc_info=True)

    try:
        from app.services.jira_service import push_test_event
        push_test_event(db, test, user, "in_review")
    except Exception as e:
        logger.warning("Jira push failed for test %s: %s", test.id, e, exc_info=True)

    return test

• Step 5: Run tests to verify they pass

Run: pytest tests/test_workflow.py -v Expected: PASS — all tests, including the ones updated in Step 1.

• Step 6: Run the full backend suite

Run: cd backend && pytest tests/ -q 2>&1 | tail -15 Expected: PASS except the one known pre-existing unrelated Tempo test failure documented in this codebase's history (if you see exactly one unrelated failure in test_tempo_service.py, that's expected and not caused by this change — verify it's the same failure by checking it's unrelated to test state/review logic, don't just assume).

• Step 7: Commit

git add backend/app/services/test_workflow_service.py backend/tests/test_workflow.py
git commit -m "feat(tests): retarget submit-red/submit-blue through review gates, add review decisions"

---

Task 6: Router — review-red/review-blue endpoints, reviewer-only guard

Files:

• Modify: backend/app/routers/tests.py

• Modify: backend/app/schemas/test.py

• Test: backend/tests/test_t109_tests_router.py or a new backend/tests/test_review_gates_router.py (check which existing file already does router-level HTTP tests for /tests/* with the client/api fixtures — use that one; if none does real HTTP-level testing for this router, create a new file following the db+client+role-header-fixture pattern already established for campaign router tests in backend/tests/test_campaign_approval_router.py, including using the shared api(...) fixture from conftest.py for any test mixing roles)

• Step 1: Add the Pydantic payloads

In backend/app/schemas/test.py, add near TestRedValidate/TestBlueValidate:

class TestRedReview(BaseModel):
    """Payload sent by the assigned Red Lead reviewer."""

    decision: str  # "approve" | "reopen"
    notes: str | None = None


class TestBlueReview(BaseModel):
    """Payload sent by the assigned Blue Lead reviewer."""

    decision: str  # "approve" | "reopen" | "gap"
    notes: str | None = None
    system_gaps: str | None = None

• Step 2: Write the failing tests

Create tests exercising, at minimum:

• A red_lead who is NOT the assigned red_reviewer_assignee gets 403 when calling POST /tests/{id}/review-red.
• The assigned reviewer approving moves the test to blue_evaluating (200, response reflects new state).
• The assigned reviewer reopening with empty notes gets a 400.
• The assigned reviewer reopening with notes moves the test back to red_executing (200).
• Same four shapes mirrored for review-blue, PLUS: the assigned blue reviewer flagging a gap with system_gaps text moves the test to in_review and the response includes the system_gaps value.
• admin can always act as reviewer regardless of assignment.

Use the existing _make_draft_campaign-style test-setup helper pattern from test_campaign_approval_router.py as a MODEL for how to construct a draft Test + drive it through start-execution → submit-red via real HTTP calls to get it into red_review state for these tests, rather than hand-constructing ORM objects mid-state (safer — proves the whole chain works, not just the isolated endpoint). Use the api(...) fixture for every multi-role test.

• Step 3: Run tests to verify they fail

Run: cd backend && pytest tests/test_review_gates_router.py -v (or whichever file you created/extended) Expected: FAIL — 404 for the new routes.

• Step 4: Implement the endpoints

In backend/app/routers/tests.py, add the service imports near the existing workflow imports (check the top of the file for the exact from app.services.test_workflow_service import (... as wf_...) pattern already used and match it):

from app.services.test_workflow_service import (
    approve_red_review as wf_approve_red_review,
)
from app.services.test_workflow_service import (
    reopen_red_review as wf_reopen_red_review,
)
from app.services.test_workflow_service import (
    approve_blue_review as wf_approve_blue_review,
)
from app.services.test_workflow_service import (
    reopen_blue_review as wf_reopen_blue_review,
)
from app.services.test_workflow_service import (
    flag_blue_review_gap as wf_flag_blue_review_gap,
)

Add the endpoints right after start-blue-work (before pause-timer):

# ---------------------------------------------------------------------------
# POST /tests/{id}/review-red  —  Red Lead reviews the operator's submission
# ---------------------------------------------------------------------------


@router.post("/{test_id}/review-red", response_model=TestOut)
def review_red(
    test_id: uuid.UUID,
    payload: TestRedReview,
    db: Session = Depends(get_db),
    current_user: User = Depends(require_any_role("red_lead", "admin")),
) -> TestOut:
    """Assigned Red Lead approves or reopens a test sitting in red_review."""
    test = crud_get_test_or_raise(db, test_id)

    if current_user.role != "admin" and test.red_reviewer_assignee != current_user.id:
        raise HTTPException(status_code=403, detail="You are not the assigned reviewer for this test")

    with UnitOfWork(db) as uow:
        if payload.decision == "approve":
            test = wf_approve_red_review(db, test, current_user, notes=payload.notes)
        elif payload.decision == "reopen":
            test = wf_reopen_red_review(db, test, current_user, notes=payload.notes or "")
        else:
            raise HTTPException(status_code=400, detail="decision must be 'approve' or 'reopen'")
        uow.commit()
    db.refresh(test)
    return test


# ---------------------------------------------------------------------------
# POST /tests/{id}/review-blue  —  Blue Lead reviews the operator's submission
# ---------------------------------------------------------------------------


@router.post("/{test_id}/review-blue", response_model=TestOut)
def review_blue(
    test_id: uuid.UUID,
    payload: TestBlueReview,
    db: Session = Depends(get_db),
    current_user: User = Depends(require_any_role("blue_lead", "admin")),
) -> TestOut:
    """Assigned Blue Lead approves, reopens, or flags a capability gap on a test in blue_review."""
    test = crud_get_test_or_raise(db, test_id)

    if current_user.role != "admin" and test.blue_reviewer_assignee != current_user.id:
        raise HTTPException(status_code=403, detail="You are not the assigned reviewer for this test")

    with UnitOfWork(db) as uow:
        if payload.decision == "approve":
            test = wf_approve_blue_review(db, test, current_user, notes=payload.notes)
        elif payload.decision == "reopen":
            test = wf_reopen_blue_review(db, test, current_user, notes=payload.notes or "")
        elif payload.decision == "gap":
            test = wf_flag_blue_review_gap(db, test, current_user, system_gaps=payload.system_gaps or "", notes=payload.notes)
        else:
            raise HTTPException(status_code=400, detail="decision must be 'approve', 'reopen', or 'gap'")
        uow.commit()
    db.refresh(test)
    return test

Note: InvalidOperationError (raised by wf_reopen_red_review/wf_reopen_blue_review/wf_flag_blue_review_gap for missing notes/system_gaps) — check how this exception is mapped to an HTTP status elsewhere in this router (it should already be registered as a domain exception handler somewhere, likely mapping to 400, same as this codebase's other domain errors). Do NOT add a manual try/except around these calls — let the existing global exception handler convert it, matching the pattern every other endpoint in this file already relies on.

Also update TestState-adjacent OpenAPI docstring comment block at the top of tests.py (the """...""" module docstring listing all endpoints) to add the two new routes, matching its existing format.

• Step 5: Run tests to verify they pass

Run: pytest tests/test_review_gates_router.py -v (or wherever you put them) Expected: PASS.

• Step 6: Run the full backend suite

Run: cd backend && pytest tests/ -q 2>&1 | tail -15 Expected: PASS except the one known pre-existing unrelated Tempo failure.

• Step 7: Commit

git add backend/app/routers/tests.py backend/app/schemas/test.py backend/tests/
git commit -m "feat(tests): review-red/review-blue router endpoints"

---

Task 7: Backend — blind visibility masking on GET /tests/{id}

Files:

• Modify: backend/app/services/test_crud_service.py (or wherever TestOut/the detail response is actually assembled — READ backend/app/routers/tests.py's GET /{test_id} endpoint first to find exactly where the response object comes from)

• Test: new or existing router test file

• Step 1: Locate the exact response-construction point

Read the GET /tests/{test_id} endpoint in backend/app/routers/tests.py in full. It almost certainly calls something like crud_get_test_detail(db, test_id) or directly returns an ORM Test object relying on response_model=TestOut's automatic ORM serialization. Determine precisely which of these it is — this changes where masking must be applied (a plain dict you can mutate vs. an ORM object FastAPI serializes automatically, which you cannot mutate to hide fields since Pydantic reads directly off the ORM attributes).

• Step 2: Write the failing tests

Add router-level tests (using the client/api fixtures) that:

• Create a test, drive it to blue_evaluating via real HTTP calls (start-execution → submit-red → approve-red-review-as-reviewer).

• As a blue_tech/blue_lead, GET /tests/{id} and assert red-only fields (procedure_text, tool_used, attack_success, red_summary, execution_start_time, execution_end_time, red_validation_status, red_validation_notes) are all None in the response, even though they have real values on the ORM record.

• As a red_tech/red_lead, GET /tests/{id} on the SAME test and assert blue-only fields (detection_result, containment_result, detection_time, containment_time, blue_summary, blue_validation_status, blue_validation_notes, system_gaps) are None.

• As admin, GET /tests/{id} and assert ALL fields are visible (both sides).

• As viewer, GET /tests/{id} and assert ALL fields are visible (viewer role is never blinded per design decision #7).

• Drive the SAME test all the way to in_review (submit-blue → approve-blue-review) and assert that NOW both a red_tech and a blue_tech viewer see BOTH sides fully (cross-validation visibility restored).

• Do NOT touch evidence file lists (red_evidences/blue_evidences) in this task's masking — check whether those are part of TestOut at all; if they're fetched via a SEPARATE endpoint (GET /tests/{id}/evidence or similar), that's out of scope for this task (flag it as a known gap in your report, don't silently expand scope to also patch the evidence endpoint unless it's part of the exact same TestOut payload you're already modifying).

• Step 3: Run tests to verify they fail

Run the new test file. Expected: FAIL — both teams currently see everything regardless of state.

• Step 4: Implement the masking

Based on what Step 1 found, implement one of:

If the endpoint returns a plain dict/service function output: add a helper function _mask_test_for_viewer(test_dict: dict, viewer_role: str, test_state: str) -> dict and call it right before returning.

If the endpoint relies on response_model=TestOut auto-serializing an ORM object directly: change the endpoint to explicitly build the TestOut first (TestOut.model_validate(test) or equivalent — check this codebase's Pydantic v2 idiom, likely already used elsewhere for similar patterns), apply masking to the resulting Pydantic model via .model_copy(update={...}), and return that constructed model instead of the raw ORM object.

Either way, the masking logic is:

_RED_ONLY_FIELDS = [
    "procedure_text", "tool_used", "attack_success",
    "execution_start_time", "execution_end_time", "red_summary",
    "red_validation_status", "red_validated_by", "red_validated_at", "red_validation_notes",
]
_BLUE_ONLY_FIELDS = [
    "detection_result", "containment_result", "detection_time", "containment_time",
    "blue_summary", "blue_validation_status", "blue_validated_by", "blue_validated_at",
    "blue_validation_notes", "system_gaps",
]
_BLIND_STATES = {"draft", "red_executing", "red_review", "blue_evaluating", "blue_review"}


def _mask_for_team_blindness(payload, *, viewer_role: str, test_state: str):
    """Hide the other team's fields until both individual reviews are done.

    admin and viewer are never blinded. Once the test reaches in_review or
    beyond, both sides see everything (existing cross-validation behavior).
    """
    if viewer_role in ("admin", "viewer"):
        return payload
    if test_state not in _BLIND_STATES:
        return payload

    hide_fields = []
    if viewer_role in ("blue_tech", "blue_lead"):
        hide_fields = _RED_ONLY_FIELDS
    elif viewer_role in ("red_tech", "red_lead"):
        hide_fields = _BLUE_ONLY_FIELDS

    if not hide_fields:
        return payload

    # adapt this line to whichever return type Step 1 determined:
    # dict:   return {**payload, **{f: None for f in hide_fields}}
    # Pydantic model: return payload.model_copy(update={f: None for f in hide_fields})
    ...

Fill in the exact final return line based on what you found in Step 1 — the plan can't specify this precisely without knowing which shape the codebase actually uses; determine it and write the correct one, don't leave both branches in the shipped code.

• Step 5: Run tests to verify they pass

Run the new test file. Expected: PASS.

• Step 6: Run the full backend suite

Run: cd backend && pytest tests/ -q 2>&1 | tail -15 Expected: PASS except the one known pre-existing unrelated Tempo failure. Pay close attention to any EXISTING test that asserts on GET /tests/{id} response content for a role/state combination this masking now touches — if any pre-existing test breaks because it expected to see the other team's data as admin or in in_review+, that's a sign the masking condition is wrong (should still show admin/in_review+ everything); if it breaks for a red_tech/blue_tech viewer during an earlier state expecting to see the other side, decide whether that test's expectation was simply wrong before this feature (likely) and needs updating to match the new, intentional behavior — update it, don't work around the masking to keep the old test passing.

• Step 7: Commit

git add backend/app/services/test_crud_service.py backend/app/routers/tests.py backend/tests/
git commit -m "feat(tests): blind red/blue visibility until both reviews are approved"

---

Task 8: Backend full-suite regression check

Files: none (verification only)

• Step 1: Run the complete backend test suite

Run: cd backend && pytest tests/ -q 2>&1 | tail -15 Expected: 100% pass except the one documented pre-existing unrelated Tempo failure (verify by name that it's the same one, not a new one).

• Step 2: Run Ruff

Run: cd backend && ruff check app/ tests/ Expected: clean, or only pre-existing issues in files this plan never touched (verify with git log -1 -- <file> on any flagged file before assuming it's pre-existing — don't wave away something you actually introduced).

• Step 3: Commit any fixes

git add -A
git commit -m "chore(tests): fix lint issues from review workflow feature"

(Skip if nothing to fix.)

---

Task 9: Frontend — types, API client, review action functions

Files:

• Modify: frontend/src/types/models.ts

• Modify: frontend/src/api/tests.ts

• Step 1: Update TestState and Test interface

In frontend/src/types/models.ts, update TestState:

export type TestState =
  | "draft"
  | "red_executing"
  | "red_review"
  | "blue_evaluating"
  | "blue_review"
  | "in_review"
  | "validated"
  | "rejected"
  | "disputed";

Add to the Test interface, in the Red/Blue field sections respectively:

  // Red Team review fields
  red_reviewer_assignee: string | null;
  red_review_by: string | null;
  red_review_at: string | null;
  red_review_notes: string | null;

  // Blue Team review fields
  blue_reviewer_assignee: string | null;
  blue_review_by: string | null;
  blue_review_at: string | null;
  blue_review_notes: string | null;
  system_gaps: string | null;

RED_EDITABLE_STATES/BLUE_EDITABLE_STATES stay unchanged — red_review/blue_review are NOT operator-editable states (only the assigned lead acts there via the review endpoints, not the field-editing endpoints).

• Step 2: Add API functions

In frontend/src/api/tests.ts, add near validateAsRedLead/validateAsBlueLead:

export interface RedReviewPayload {
  decision: "approve" | "reopen";
  notes?: string;
}

export interface BlueReviewPayload {
  decision: "approve" | "reopen" | "gap";
  notes?: string;
  system_gaps?: string;
}

/** Assigned Red Lead approves or reopens a test sitting in red_review. */
export async function reviewAsRedLead(testId: string, payload: RedReviewPayload): Promise<Test> {
  const { data } = await client.post<Test>(`/tests/${testId}/review-red`, payload);
  return data;
}

/** Assigned Blue Lead approves, reopens, or flags a gap on a test in blue_review. */
export async function reviewAsBlueLead(testId: string, payload: BlueReviewPayload): Promise<Test> {
  const { data } = await client.post<Test>(`/tests/${testId}/review-blue`, payload);
  return data;
}

• Step 3: Verify by careful manual reading (no build tool available)

Confirm Test import type is used consistently (check whether tests.ts imports Test from ../types/models already — it should, given TestOut-equivalent functions already exist there).

• Step 4: Commit

git add frontend/src/types/models.ts frontend/src/api/tests.ts
git commit -m "feat(tests): types and API client for red/blue review workflow"

---

Task 10: Frontend — review panel UI, blind visibility, state badges

Files:

• Modify: frontend/src/components/test-detail/TeamTabs.tsx

• Modify: frontend/src/pages/TestDetailPage.tsx

• Grep for and modify every other file with a TestState-keyed color/label map (e.g. a tests list page, CampaignDetailPage.tsx's embedded testStateColors, AddTestToCampaignModal.tsx's stateBadge, CampaignTimeline.tsx's stateColors)

• Step 1: Find every TestState color/label map

Run: grep -rln "red_executing" frontend/src to enumerate every file with a hardcoded state list. For EACH one found, add red_review and blue_review entries with a distinct color (suggest amber/purple family, distinct from the existing red/indigo/blue/green palette already used for the other states — check each file's existing palette before picking, to avoid a collision with an already-used color in that specific file).

• Step 2: Add "Queued Blue Team" display label

Wherever blue_evaluating is rendered as a human label (search for .replace(/_/g, " ") applied to test.state or a blue_evaluating label lookup), add a helper that shows "Queued Blue Team" when state === "blue_evaluating" && !blue_work_started_at, and "Blue Evaluating" otherwise. Do NOT invent a new backend state for this — it is purely test.state === "blue_evaluating" with a conditional label based on test.blue_work_started_at.

• Step 3: Add the review panel to TeamTabs.tsx

Add a new render function renderRedReviewPanel() shown ONLY when test.state === "red_review" AND the viewer is the assigned reviewer (user?.id === test.red_reviewer_assignee) or admin. Panel has: Approve button, Reopen button (opens a small inline textarea requiring notes before enabling submit — mirror the existing validate-red/blue notes UX already in this file's validation-status display blocks, or the campaign-feature's reject-modal pattern from a previous feature for inspiration on the "notes required before enabling submit" interaction). Wire both buttons to reviewAsRedLead via a useMutation in TestDetailPage.tsx (this component doesn't own its own mutations today — mutations live in the parent page and get passed down as props, matching the existing onRedFieldChange/onBlueFieldChange prop-drilling pattern already used in this file; follow that same convention for the new review actions rather than adding mutations inside TeamTabs.tsx itself).

Add the mirrored renderBlueReviewPanel() for blue_review state, with THREE actions: Approve, Reopen (notes required), Flag Gap (system_gaps text required, separate from the general notes field — two distinct textareas, only one of which is used depending on which button the reviewer intends to click; disable Flag Gap's submit until system_gaps is non-empty, disable Reopen's submit until notes is non-empty, Approve has no requirement).

• Step 4: Implement blind visibility in the tab bar and tab content

In TeamTabs.tsx, compute:

const BLIND_STATES: TestState[] = ["draft", "red_executing", "red_review", "blue_evaluating", "blue_review"];
const isBlind = BLIND_STATES.includes(test.state) && role !== "admin" && role !== "viewer";
const hideBlueFromMe = isBlind && (role === "red_tech" || role === "red_lead");
const hideRedFromMe = isBlind && (role === "blue_tech" || role === "blue_lead");

Since the BACKEND already nulls out the hidden fields (Task 7), the frontend doesn't need to invent its own hiding logic for the DATA — test.detection_result etc. will already arrive as null for a blinded viewer. What the frontend DOES need to change is the UX around it: when hideBlueFromMe is true, the "Blue Team" tab should show a clear placeholder ("Hidden until your team's review is complete — this is a blind test") instead of rendering the normal blue-tab content with confusingly-empty fields that could be mistaken for "not filled in yet" rather than "intentionally hidden". Same for hideRedFromMe and the Red tab. Do this by adding an early return at the top of renderRedTab()/renderBlueTab():

  const renderRedTab = () => {
    if (hideRedFromMe) {
      return (
        <div className="py-12 text-center">
          <Shield className="mx-auto h-10 w-10 text-gray-600" />
          <p className="mt-2 text-sm text-gray-400">
            Red Team work is hidden until both teams' reviews are complete — this keeps detection testing blind.
          </p>
        </div>
      );
    }
    return (
      <div className="space-y-6">
        {/* ...existing content unchanged... */}

Same pattern for renderBlueTab() with hideBlueFromMe.

• Step 5: Verify by careful manual reading (no build tool available)

Read the entire modified TeamTabs.tsx end-to-end. Confirm JSX balance, that BLIND_STATES/isBlind/hideBlueFromMe/hideRedFromMe are declared before use, and that the new review-panel render functions are correctly wired into TAB_CONTENT or rendered as an additional block above the existing tab content (your choice, whichever reads more naturally in context — but be explicit and deliberate about which, don't leave it ambiguous).

• Step 6: Commit

git add frontend/src/components/test-detail/TeamTabs.tsx frontend/src/pages/TestDetailPage.tsx frontend/src/**/*.tsx
git commit -m "feat(tests): review panels, blind visibility UX, red_review/blue_review badges"

(Adjust the git add file list to the actual files Step 1 found — don't blindly glob **/*.tsx, list them explicitly in the real commit.)

---

Task 11: Final end-to-end verification

Files: none (verification only)

• Step 1: Run the full backend suite one more time

Run: cd backend && pytest tests/ -q 2>&1 | tail -15 Expected: 100% pass except the one documented pre-existing Tempo failure.

• Step 2: Manual code trace of the full lifecycle

Trace through the code (not a live browser test, since Node/npm isn't available) for this exact sequence, confirming each step's state/field changes by reading the actual functions involved:

1. red_tech creates test, start-execution → red_executing.
2. Same red_tech calls submit-red → red_review, reviewer auto-assigned (a DIFFERENT red_lead, load-balanced), Jira ticket reassigned to that reviewer, that reviewer notified.
3. The assigned reviewer calls review-red with decision=reopen, notes="add more detail" → back to red_executing, original operator notified, red_started_at reset.
4. Operator adds more evidence, calls submit-red again → red_review again (reviewer re-selected, could be same or different lead depending on current load).
5. Assigned reviewer calls review-red with decision=approve → blue_evaluating, blue_started_at set, blue_tech users notified (existing broadcast).
6. A blue_tech calls start-blue-work → blue_work_started_at set, blue_tech_assignee claimed.
7. Same blue_tech calls submit-blue → blue_review, a blue_lead auto-assigned, notified.
8. Assigned blue reviewer calls review-blue with decision=gap, system_gaps="Missing EDR on host" → in_review, system_gaps persisted, both leads notified (existing in_review broadcast).
9. validate-red and validate-blue proceed exactly as before (unchanged) → validated/rejected/disputed.

Confirm via GET /tests/{id} at steps 1-7 that a blue_tech/blue_lead viewer sees red fields as null, and a red_tech/red_lead viewer sees blue fields as null — and that this flips to full visibility at step 8 onward.

• Step 3: Report completion

No commit for this task — it's verification only. If any step in the trace reveals a real gap, return to the relevant task above and fix it before considering the feature done.

---