W7: Possibilities matrix (work-already-done fast path) (#522)

* [W7] Add possibilities_matrix_enabled feature flag (default off)

* [W7] Add _work_appears_done predicate (status+commits+PR+ACs+no-open-findings)

* [W7] Add CI-green quality proxy for the fast path (local fallback on no-CI)

* [W7] Add work-already-done fast path in i_am_done (slimmed gates, no rich plan)

* [W7] Add WORK_ALREADY_DONE prompt state

* [W7] Make fast path mypy-clean (cast to helpers for _resolve_ci_status; typed mock locals)

* [W7] Extract _all_criteria_addressed to bring _work_appears_done under xenon B

---------

Co-authored-by: Renn F <rennf93@users.noreply.github.com>

panel/src/components/settings/feature-flags-card.tsx

@@ -51,6 +51,8 @@ const FLAG_DESCRIPTIONS: Record<string, string> = {
     "Provision each agent workspace with the target project's Python (not RoboCo's) and block delivery gates when its test suite can't be executed.",
   conventions_enabled:
     "Enforce a per-project architectural standard (.roboco/conventions.yml): inject the map, attach baseline constraints, and block i_am_done / pr_pass on misplaced definitions or lint suppressions.",
+  possibilities_matrix_enabled:
+    "When a task's work is already done (commits + open PR + all acceptance criteria addressed + no open findings), submit it for QA in one i_am_done call instead of 3-6 turns — skips the retroactive plan, journal tracing, and local quality (CI-green proxy) gates. Off by default: the standard path is unchanged until you arm this.",
   rag_auto_update_enabled:
     "Keep the knowledge base index refreshed automatically.",
   transcript_prune_enabled:

roboco/config.py

@@ -318,6 +318,16 @@ class Settings(BaseSettings):
             "fully inert."
         ),
     )
+    possibilities_matrix_enabled: bool = Field(
+        default=False,
+        description=(
+            "Possibilities matrix: when a task's work is already done (commits "
+            "+ PR open + all acceptance criteria addressed + no open findings), "
+            "let i_am_done submit it for QA in one call, skipping the retroactive "
+            "rich-plan, journal tracing, and local quality (CI-green proxy) "
+            "gates. Off => the standard i_am_done path is unchanged."
+        ),
+    )
 
     # ==========================================================================
     # Web Research (pluggable external search/fetch for Board + PM roles)

roboco/runtime/orchestrator.py

@@ -14169,6 +14169,18 @@ Run the project's quality checks against acceptance criteria:
    i_am_done(task_id="{task_id}", notes="<verification summary>")
    — chains submit_verification + push + create_pr + submit_qa.
 4. If issues found: commit() the fixes and retry.
+""",
+            "WORK_ALREADY_DONE": f"""## WORK ALREADY DONE
+
+Your task already has commits and an open PR — the work appears complete. The
+server's fast path runs the slimmed gate set (ownership / commits / PR / open
+findings / acceptance criteria / branch-pushed / not-behind-base / conventions /
+CI-green) for you, so skip re-verification and submit directly:
+
+  i_am_done(task_id="{task_id}", notes="<one-line summary of what was done>")
+
+If the fast path refuses (a gate it checks is not actually met), the
+`remediate` field names exactly what's missing — fix it and retry i_am_done.
 """,
         }
         return instructions.get(
@@ -14185,6 +14197,20 @@ Run the project's quality checks against acceptance criteria:
         # Determine workflow state based on task attributes
         has_plan = bool(task.get("plan"))
         workflow_state = self._get_workflow_state(status, has_plan)
+        # Possibilities-matrix prompt proxy: when the flag is armed and the
+        # task already has commits + an open PR, steer the dev to submit in one
+        # turn instead of re-deriving (re-running gates, re-reading the diff).
+        # This is a cheap sync proxy — the async DB gates (AC coverage, open
+        # findings) are NOT re-checked here; the server fast path
+        # (_i_am_done_fast_path) is the authority and runs them. The prompt
+        # just collapses the 3-5-turn re-derivation to a single i_am_done call.
+        if (
+            settings.possibilities_matrix_enabled
+            and status in ("claimed", "in_progress", "verifying")
+            and task.get("pr_created")
+            and task.get("commits")
+        ):
+            workflow_state = "WORK_ALREADY_DONE"
         open_findings_block = ""
         if workflow_state == "REVISION_REQUIRED":
             open_findings_block = await self._open_findings_prompt_block(str(task_id))

roboco/services/gateway/choreographer/_impl.py

@@ -2042,14 +2042,11 @@ class Choreographer:
             original_developer_slug=_extract_original_developer(t),
             notes=notes,
         )
-        # Recovery re-entry: task already in `verifying` owned by the caller
+        # Pre-spec-gate short-circuits (verifying-resume recovery + the
-        # (e.g. orchestrator restart between submit_verification and
+        # work-already-done fast path), folded into one helper to keep i_am_done
-        # submit_qa). The spec gate would reject because the first composed
+        # within its return-count budget (mirrors _fresh_dev_claim's extraction).
-        # action `submit_verification` requires source IN_PROGRESS. Run only
+        if dispatched := await self._i_am_done_pre_gate_dispatch(ctx, t, agent_id):
-        # submit_qa via the runner-equivalent path, then continue with the
+            return dispatched
-        # standard tracing/field gates beforehand.
-        if str(t.status) == "verifying" and t.assigned_to == agent_id:
-            return await self._i_am_done_resume_from_verifying(ctx)
         # Stale/superseded agent: the task was reassigned out from under this
         # agent (PM reassign / reaper unclaim / escalation redirect) while its
         # container kept running. Without this, the spec gate's
@@ -2180,6 +2177,55 @@ class Choreographer:
             context_briefing=ctx.briefing,
         )
 
+    async def _fast_path_quality_verdict(
+        self, ctx: _IAmDoneContext
+    ) -> tuple[Envelope | None, bool]:
+        """Quality posture for the work-already-done fast path.
+
+        Trusts the PR's CI-green signal — the same signal the downstream
+        ``pr_pass`` gate trusts (``_ci_status_guard``) — and skips the local
+        workspace gate when CI is green. A repo with no CI configured (or an
+        unresolvable lookup) has no CI signal, so the local gate is the only
+        signal and runs. A CI ``failure`` is a known-bad build: refuse the fast
+        path rather than shipping red, even before the local gate runs.
+
+        Returns ``(rejection|None, ran_local)``. ``ran_local`` lets the caller
+        distinguish "the local gate already decided" from "CI green, gate
+        skipped" so it never double-gates.
+        """
+        # ``_resolve_ci_status`` lives on ``PRGateMixin``; ``_LegacyChoreographer``
+        # reaches it via the typed-helpers cast (same idiom as the pr_pass path).
+        status = await cast("ChoreographerHelpers", self)._resolve_ci_status(
+            ctx.task_id, ctx.task
+        )
+        if status is not None:
+            state = status.get("state")
+            if state == "success":
+                return None, False
+            if state == "failure":
+                names = (
+                    ", ".join(status.get("failing_checks") or [])
+                    or "one or more checks"
+                )
+                return (
+                    Envelope.invalid_state(
+                        message=(
+                            f"fast path refused — PR CI is failing ({names}); "
+                            "QA reviews working code, not a red build"
+                        ),
+                        remediate=(
+                            "fix the failing CI checks, commit, and call i_am_done "
+                            "again — or run the standard path (leave "
+                            "possibilities_matrix off)"
+                        ),
+                        context_briefing=ctx.briefing,
+                    ),
+                    False,
+                )
+        # No CI signal (no_ci_configured / pending / pending_not_scheduled /
+        # error / unresolvable): the local gate is the only signal — run it.
+        return await self._check_quality_gate(ctx), True
+
     async def _toolchain_broken_guard(
         self, agent_id: UUID, task: Any, *, reviewer: bool = False
     ) -> Envelope | None:
@@ -2723,6 +2769,112 @@ class Choreographer:
         await self._touch(ctx.task_id)
         return await self._build_i_am_done_ok(ctx.agent_id, ctx.task_id, t)
 
+    async def _i_am_done_pre_gate_dispatch(
+        self, ctx: _IAmDoneContext, t: Any, agent_id: UUID
+    ) -> Envelope | None:
+        """Pre-spec-gate short-circuits, folded to keep i_am_done within its
+        return-count budget:
+
+        1. verifying-resume recovery (task already ``verifying`` owned by the
+           caller — orchestrator restart between submit_verification and
+           submit_qa; the spec gate would reject the first composed action).
+        2. the possibilities-matrix fast path (work already done).
+
+        Returns the dispatched envelope, or None to fall through to the
+        standard spec/soup/gate path.
+        """
+        if str(t.status) == "verifying" and t.assigned_to == agent_id:
+            return await self._i_am_done_resume_from_verifying(ctx)
+        return await self._maybe_i_am_done_fast_path(ctx, t, agent_id)
+
+    async def _maybe_i_am_done_fast_path(
+        self, ctx: _IAmDoneContext, t: Any, agent_id: UUID
+    ) -> Envelope | None:
+        """Possibilities-matrix fast-path gate: returns the fast-path envelope
+        when the flag is armed, the task is claimed/in_progress and owned by
+        the caller, and ``_work_appears_done`` holds; None to fall through to
+        the standard i_am_done path. Extracted from i_am_done to keep that
+        verb within its return-count budget. ``assigned_to`` is asserted here
+        so a reassigned agent cannot fast-path a task that is no longer theirs.
+        """
+        from roboco.config import settings as _settings
+
+        if not _settings.possibilities_matrix_enabled:
+            return None
+        if str(t.status) not in ("claimed", "in_progress"):
+            return None
+        if t.assigned_to != agent_id:
+            return None
+        if not await self._work_appears_done(t):
+            return None
+        return await self._i_am_done_fast_path(ctx)
+
+    async def _i_am_done_fast_path(self, ctx: _IAmDoneContext) -> Envelope:
+        """Work-already-done fast path: slimmed gates + direct transition chain.
+
+        Skips the turn-costly gates the standard ``_i_am_done_gate`` runs: the
+        retroactive rich-plan gate (``start`` without ``set_plan`` — a done task
+        has no plan to author), the journal tracing gates (``notes`` is the
+        reflect/handoff substitute, recorded as a progress entry by
+        ``submit_verification`` / ``submit_qa``), and the local quality gate
+        (the CI-green proxy in ``_fast_path_quality_verdict``). Keeps the
+        non-negotiable gates the predicate already asserts, plus conventions
+        (leaf dev tasks skip ``awaiting_pr_review`` so the conventions check
+        must not also be skipped here — it is tree-sitter, sub-second, 0 turns
+        when green).
+        """
+        await self._apply_resolved_findings(ctx)
+        guards = (
+            lambda: self._check_submit_qa_field_gates(
+                ctx.agent_id, ctx.task_id, ctx.task
+            ),
+            lambda: self._behind_base_gate(ctx),
+            lambda: self._ensure_branch_pushed(ctx),
+            lambda: self._conventions_gate(ctx),
+        )
+        for guard in guards:
+            if rejection := await guard():
+                return await self._reject_i_am_done(ctx, rejection)
+        # FINDINGS_ADDRESSED re-checked post-resolution: a resolved_findings entry
+        # in THIS call may have closed the open set; anything still open blocks.
+        if await self._open_finding_ids(ctx.task_id):
+            return await self._reject_i_am_done(
+                ctx,
+                Envelope.invalid_state(
+                    message="fast path blocked — open findings remain on the ledger",
+                    remediate=(
+                        "name every open finding in resolved_findings "
+                        "({finding_id, note}), or leave possibilities_matrix off "
+                        "and run the standard i_am_done path"
+                    ),
+                    context_briefing=ctx.briefing,
+                ),
+            )
+        rejection, _ran_local = await self._fast_path_quality_verdict(ctx)
+        if rejection is not None:
+            return await self._reject_i_am_done(ctx, rejection)
+        try:
+            if str(ctx.task.status) == "claimed":
+                await self.task.start(ctx.task_id, ctx.agent_id, "developer")
+            await self.task.submit_verification(ctx.agent_id, ctx.task_id, ctx.notes)
+            submitted = await self.task.submit_qa(ctx.agent_id, ctx.task_id, ctx.notes)
+        except Exception as exc:
+            return await self._reject_i_am_done(
+                ctx,
+                Envelope.invalid_state(
+                    message=f"fast-path transition failed: {exc}",
+                    remediate="check workspace + retry; if persistent, call i_am_idle",
+                    context_briefing=ctx.briefing,
+                ),
+            )
+        t = submitted if submitted is not None else ctx.task
+        await self._notify_qa(ctx.agent_id, ctx.task_id, t)
+        await self._touch(ctx.task_id)
+        await self._record_milestone_progress(
+            ctx.task_id, ctx.agent_id, "submitted for QA review", percentage=90
+        )
+        return await self._build_i_am_done_ok(ctx.agent_id, ctx.task_id, t)
+
     async def _open_finding_ids(self, task_id: UUID) -> tuple[str, ...]:
         """8-char ids of the task's still-OPEN revision-ledger findings.
 
@@ -2745,6 +2897,67 @@ class Choreographer:
             return ()
         return tuple(str(row.id)[:8] for row in open_rows)
 
+    async def _work_appears_done(self, t: Any) -> bool:
+        """True when a task's work is provably already done — the gate for the
+        possibilities-matrix fast path. status in {claimed, in_progress,
+        verifying} + >=1 commit + PR open + every acceptance criterion
+        addressed + no open findings. Ownership is NOT checked here (the
+        fast-path branch asserts ``assigned_to`` so a reassigned agent cannot
+        fast-path a task that is no longer theirs).
+
+        A criterion counts as addressed if its per-criterion row carries a
+        non-empty artifact reference OR an explicit ``addressed`` flag. The
+        writer (``_new_criterion_entry``) stores ``artifact_ref`` while the
+        canonical reader (``_already_addressed_criteria``) looks for
+        ``referencing_artifact_id`` — a latent key drift masked today because
+        the standard AC gate passes via the ``journal_reflect_present``
+        blanket. The fast path skips that blanket (it skips the reflect gate),
+        so this predicate reads BOTH keys plus ``addressed`` to see real data
+        rather than ship dead code.
+        """
+        if str(t.status) not in ("claimed", "in_progress", "verifying"):
+            return False
+        if not getattr(t, "commits", None):
+            return False
+        if not (getattr(t, "pr_created", False) or getattr(t, "pr_number", None)):
+            return False
+        if not self._all_criteria_addressed(t):
+            return False
+        return not await self._open_finding_ids(t.id)
+
+    @staticmethod
+    def _all_criteria_addressed(t: Any) -> bool:
+        """Every acceptance criterion has a per-criterion row marked addressed.
+
+        A criterion counts as addressed if its row carries a non-empty
+        artifact reference OR an explicit ``addressed`` flag. The writer
+        (``_new_criterion_entry``) stores ``artifact_ref`` while the canonical
+        reader (``_already_addressed_criteria``) looks for
+        ``referencing_artifact_id`` — a latent key drift masked today because
+        the standard AC gate passes via the ``journal_reflect_present``
+        blanket. The fast path skips that blanket (it skips the reflect gate),
+        so this reads BOTH keys plus ``addressed`` to see real data rather
+        than ship dead code. A task with no criteria is trivially covered.
+        """
+        criteria = list(getattr(t, "acceptance_criteria", []) or [])
+        if not criteria:
+            return True
+        rows = list(getattr(t, "acceptance_criteria_status", []) or [])
+        addressed: set[str] = set()
+        for row in rows:
+            if not isinstance(row, dict):
+                continue
+            if not (
+                row.get("referencing_artifact_id")
+                or row.get("artifact_ref")
+                or row.get("addressed")
+            ):
+                continue
+            crit = row.get("criterion")
+            if crit:
+                addressed.add(str(crit))
+        return addressed >= set(criteria)
+
     async def _check_tracing_gates(
         self, agent_id: UUID, task_id: UUID, t: Any
     ) -> Envelope | None:

roboco/services/gateway/choreographer/_protocol.py

@@ -141,6 +141,9 @@ class ChoreographerHelpers:
     ) -> Envelope:
         raise NotImplementedError
 
+    async def _resolve_ci_status(self, task_id: UUID, t: Any) -> dict[str, Any] | None:
+        raise NotImplementedError
+
 
 def actor_context_fields(agent: Any) -> tuple[str | None, str | None]:
     """``(actor_slug, agent_team)`` for a spec ``Context``, None-agent safe.

roboco/services/settings.py

@@ -53,6 +53,10 @@ FEATURE_FLAGS: tuple[tuple[str, str], ...] = (
     ("provisioning_enabled", "Pitch auto-provisioning"),
     ("toolchain_match_enabled", "Agent runtime toolchain matching"),
     ("conventions_enabled", "Architectural conventions standard"),
+    (
+        "possibilities_matrix_enabled",
+        "Possibilities matrix (work-already-done fast path)",
+    ),
     ("rag_auto_update_enabled", "RAG auto-update"),
     ("transcript_prune_enabled", "Transcript pruning"),
     ("gateway_health_enabled", "Gateway-health recovery"),

tests/unit/config/test_possibilities_matrix_flag.py

@@ -0,0 +1,27 @@
+"""Possibilities matrix (work-already-done fast path) is gated by a default-off
+config flag, registered as a panel-tunable feature flag (the #487 lesson)."""
+
+from __future__ import annotations
+
+import os
+from unittest import mock
+
+from roboco.config import Settings
+from roboco.services.settings import FEATURE_FLAGS, validate_setting
+
+
+def test_possibilities_matrix_disabled_by_default() -> None:
+    assert Settings().possibilities_matrix_enabled is False
+
+
+def test_possibilities_matrix_reads_env_var() -> None:
+    with mock.patch.dict(os.environ, {"ROBOCO_POSSIBILITIES_MATRIX_ENABLED": "true"}):
+        assert Settings().possibilities_matrix_enabled is True
+
+
+def test_possibilities_matrix_flag_registered_in_feature_flags() -> None:
+    assert "possibilities_matrix_enabled" in [key for key, _ in FEATURE_FLAGS]
+
+
+def test_possibilities_matrix_flag_validates_as_bool() -> None:
+    validate_setting("possibilities_matrix_enabled", "true")

tests/unit/gateway/test_possibilities_matrix.py

@@ -0,0 +1,367 @@
+"""W7 possibilities matrix: the ``_work_appears_done`` predicate + fast path.
+
+A task whose work is already done (commits + PR open + every acceptance
+criterion addressed + no open findings) qualifies for the fast path. These
+tests pin the predicate's truth table and the schema it actually reads (the
+per-criterion rows the writer persists use ``artifact_ref``; the predicate
+unions ``artifact_ref`` / ``referencing_artifact_id`` / ``addressed`` so it
+sees real data, not the latent reader/writer key drift).
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock
+from uuid import uuid4
+
+import pytest
+from roboco.services.gateway.choreographer import Choreographer, ChoreographerDeps
+from roboco.services.gateway.envelope import Envelope
+
+
+def _deps() -> ChoreographerDeps:
+    return ChoreographerDeps(
+        task=AsyncMock(),
+        work_session=AsyncMock(),
+        git=AsyncMock(),
+        a2a=AsyncMock(),
+        journal=AsyncMock(),
+        audit=AsyncMock(),
+        evidence_repo=AsyncMock(),
+    )
+
+
+async def _no_findings(_task_id: Any) -> tuple[()]:
+    return ()
+
+
+async def _one_open(_task_id: Any) -> tuple[str, ...]:
+    return ("f1abcd12",)
+
+
+def _t(
+    *,
+    status: str = "claimed",
+    commits: tuple[int, ...] = (1,),
+    pr_created: bool = True,
+    criteria: tuple[str, ...] = ("ac1", "ac2"),
+    ac_status: list[dict[str, Any]] | None = None,
+) -> MagicMock:
+    if ac_status is None:
+        ac_status = [
+            {"criterion": "ac1", "addressed": True, "artifact_ref": "sha1"},
+            {"criterion": "ac2", "addressed": True, "artifact_ref": "sha2"},
+        ]
+    t = MagicMock()
+    t.id = uuid4()
+    t.status = status
+    t.commits = commits
+    t.pr_created = pr_created
+    t.pr_number = 12345 if pr_created else None
+    t.acceptance_criteria = list(criteria)
+    t.acceptance_criteria_status = ac_status
+    return t
+
+
+@pytest.mark.asyncio
+async def test_work_appears_done_true_when_all_hold(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    c = Choreographer(_deps())
+    monkeypatch.setattr(c, "_open_finding_ids", _no_findings)
+    assert await c._work_appears_done(_t()) is True
+
+
+@pytest.mark.asyncio
+async def test_work_appears_done_false_when_no_pr(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    c = Choreographer(_deps())
+    monkeypatch.setattr(c, "_open_finding_ids", _no_findings)
+    assert await c._work_appears_done(_t(pr_created=False)) is False
+
+
+@pytest.mark.asyncio
+async def test_work_appears_done_false_when_no_commits(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    c = Choreographer(_deps())
+    monkeypatch.setattr(c, "_open_finding_ids", _no_findings)
+    assert await c._work_appears_done(_t(commits=())) is False
+
+
+@pytest.mark.asyncio
+async def test_work_appears_done_false_when_ac_unaddressed(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    c = Choreographer(_deps())
+    monkeypatch.setattr(c, "_open_finding_ids", _no_findings)
+    t = _t(
+        ac_status=[
+            {"criterion": "ac1", "addressed": True, "artifact_ref": "sha1"},
+            {"criterion": "ac2", "addressed": False, "artifact_ref": None},
+        ]
+    )
+    assert await c._work_appears_done(t) is False
+
+
+@pytest.mark.asyncio
+async def test_work_appears_done_true_with_no_criteria(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    c = Choreographer(_deps())
+    monkeypatch.setattr(c, "_open_finding_ids", _no_findings)
+    assert await c._work_appears_done(_t(criteria=(), ac_status=[])) is True
+
+
+@pytest.mark.asyncio
+async def test_work_appears_done_false_when_open_finding(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    c = Choreographer(_deps())
+    monkeypatch.setattr(c, "_open_finding_ids", _one_open)
+    assert await c._work_appears_done(_t()) is False
+
+
+@pytest.mark.asyncio
+async def test_work_appears_done_false_when_terminal_status(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    c = Choreographer(_deps())
+    monkeypatch.setattr(c, "_open_finding_ids", _no_findings)
+    assert await c._work_appears_done(_t(status="awaiting_qa")) is False
+    assert await c._work_appears_done(_t(status="completed")) is False
+    assert await c._work_appears_done(_t(status="needs_revision")) is False
+
+
+@pytest.mark.asyncio
+async def test_work_appears_done_reads_referencing_artifact_id_schema(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    c = Choreographer(_deps())
+    monkeypatch.setattr(c, "_open_finding_ids", _no_findings)
+    t = _t(
+        ac_status=[
+            {"criterion": "ac1", "referencing_artifact_id": "sha1"},
+            {"criterion": "ac2", "referencing_artifact_id": "sha2"},
+        ]
+    )
+    assert await c._work_appears_done(t) is True
+
+
+# ---------------------------------------------------------------------------
+# _fast_path_quality_verdict: CI-green proxy for the skipped local gate
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_quality_verdict_ci_success_skips_local_gate(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    c = Choreographer(_deps())
+    monkeypatch.setattr(
+        c, "_resolve_ci_status", AsyncMock(return_value={"state": "success"})
+    )
+    local = AsyncMock(return_value=None)
+    monkeypatch.setattr(c, "_check_quality_gate", local)
+    rejection, ran_local = await c._fast_path_quality_verdict(
+        MagicMock(task_id=uuid4(), task=MagicMock(), briefing={})
+    )
+    assert rejection is None
+    assert ran_local is False
+    local.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_quality_verdict_ci_failure_refuses(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    c = Choreographer(_deps())
+    monkeypatch.setattr(
+        c,
+        "_resolve_ci_status",
+        AsyncMock(return_value={"state": "failure", "failing_checks": ["lint"]}),
+    )
+    local = AsyncMock(return_value=None)
+    monkeypatch.setattr(c, "_check_quality_gate", local)
+    rejection, ran_local = await c._fast_path_quality_verdict(
+        MagicMock(task_id=uuid4(), task=MagicMock(), briefing={})
+    )
+    assert rejection is not None
+    assert ran_local is False
+    local.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_quality_verdict_no_ci_falls_back_to_local_gate(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    c = Choreographer(_deps())
+    monkeypatch.setattr(
+        c, "_resolve_ci_status", AsyncMock(return_value={"state": "no_ci_configured"})
+    )
+    local = AsyncMock(return_value=None)
+    monkeypatch.setattr(c, "_check_quality_gate", local)
+    rejection, ran_local = await c._fast_path_quality_verdict(
+        MagicMock(task_id=uuid4(), task=MagicMock(), briefing={})
+    )
+    assert rejection is None
+    assert ran_local is True
+    local.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_quality_verdict_unresolvable_falls_back_to_local_gate(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    c = Choreographer(_deps())
+    monkeypatch.setattr(c, "_resolve_ci_status", AsyncMock(return_value=None))
+    local = AsyncMock(return_value=None)
+    monkeypatch.setattr(c, "_check_quality_gate", local)
+    rejection, ran_local = await c._fast_path_quality_verdict(
+        MagicMock(task_id=uuid4(), task=MagicMock(), briefing={})
+    )
+    assert rejection is None
+    assert ran_local is True
+    local.assert_awaited_once()
+
+
+# ---------------------------------------------------------------------------
+# _i_am_done_fast_path: gate ordering + transition chain (skips rich plan)
+# ---------------------------------------------------------------------------
+
+
+def _ctx(status: str = "claimed") -> Any:
+    ctx = MagicMock()
+    ctx.agent_id = uuid4()
+    ctx.task_id = uuid4()
+    ctx.task = _t(status=status)
+    ctx.briefing = {}
+    ctx.notes = "done"
+    ctx.resolved_findings = None
+    ctx.role_str = "developer"
+    return ctx
+
+
+@dataclass
+class _FastPathMocks:
+    """Typed handles to the patched verb-boundary mocks. mypy keeps the
+    declared method types on ``c.<method>`` even after ``monkeypatch.setattr``
+    (the runtime override is invisible to static analysis), so assertions
+    must go through these locals typed as ``AsyncMock`` — not ``c.<method>``."""
+
+    ok: AsyncMock
+    reject: AsyncMock
+    conventions: AsyncMock
+    open_findings: AsyncMock
+    record_milestone: AsyncMock
+
+
+def _stub_fast_path(
+    c: Choreographer, monkeypatch: pytest.MonkeyPatch, quality_rejection: Any = None
+) -> _FastPathMocks:
+    conventions = AsyncMock(return_value=None)
+    open_findings = AsyncMock(return_value=())
+    ok = AsyncMock(return_value="OK")
+    reject = AsyncMock(return_value="REJECT")
+    record_milestone = AsyncMock(return_value=None)
+    monkeypatch.setattr(c, "_apply_resolved_findings", AsyncMock(return_value=None))
+    monkeypatch.setattr(c, "_check_submit_qa_field_gates", AsyncMock(return_value=None))
+    monkeypatch.setattr(c, "_behind_base_gate", AsyncMock(return_value=None))
+    monkeypatch.setattr(c, "_ensure_branch_pushed", AsyncMock(return_value=None))
+    monkeypatch.setattr(c, "_conventions_gate", conventions)
+    monkeypatch.setattr(c, "_open_finding_ids", open_findings)
+    monkeypatch.setattr(
+        c,
+        "_fast_path_quality_verdict",
+        AsyncMock(return_value=(quality_rejection, False)),
+    )
+    monkeypatch.setattr(c, "_notify_qa", AsyncMock(return_value=None))
+    monkeypatch.setattr(c, "_touch", AsyncMock(return_value=None))
+    monkeypatch.setattr(c, "_record_milestone_progress", record_milestone)
+    monkeypatch.setattr(c, "_build_i_am_done_ok", ok)
+    monkeypatch.setattr(c, "_reject_i_am_done", reject)
+    return _FastPathMocks(
+        ok=ok,
+        reject=reject,
+        conventions=conventions,
+        open_findings=open_findings,
+        record_milestone=record_milestone,
+    )
+
+
+@pytest.mark.asyncio
+async def test_fast_path_claimed_starts_without_set_plan(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    c = Choreographer(_deps())
+    stubs = _stub_fast_path(c, monkeypatch)
+    await c._i_am_done_fast_path(_ctx(status="claimed"))
+    stubs.ok.assert_awaited_once()  # OK path taken, no rejection
+    c.task.start.assert_awaited_once()  # claimed -> in_progress
+    c.task.set_plan.assert_not_awaited()  # rich plan SKIPPED (no set_plan)
+    c.task.submit_verification.assert_awaited_once()
+    c.task.submit_qa.assert_awaited_once()
+    stubs.conventions.assert_awaited_once()  # conventions KEPT
+    stubs.record_milestone.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_fast_path_in_progress_skips_start(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    c = Choreographer(_deps())
+    stubs = _stub_fast_path(c, monkeypatch)
+    await c._i_am_done_fast_path(_ctx(status="in_progress"))
+    stubs.ok.assert_awaited_once()
+    c.task.start.assert_not_awaited()  # already in_progress
+    c.task.submit_verification.assert_awaited_once()
+    c.task.submit_qa.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_fast_path_open_findings_blocks(monkeypatch: pytest.MonkeyPatch) -> None:
+    c = Choreographer(_deps())
+    stubs = _stub_fast_path(c, monkeypatch)
+    monkeypatch.setattr(c, "_open_finding_ids", AsyncMock(return_value=("f1abcd12",)))
+    await c._i_am_done_fast_path(_ctx())
+    stubs.reject.assert_awaited_once()
+    c.task.submit_qa.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_fast_path_conventions_block_rejects(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    c = Choreographer(_deps())
+    stubs = _stub_fast_path(c, monkeypatch)
+    monkeypatch.setattr(
+        c,
+        "_conventions_gate",
+        AsyncMock(return_value=Envelope.invalid_state(message="x", remediate="fix")),
+    )
+    await c._i_am_done_fast_path(_ctx())
+    stubs.reject.assert_awaited_once()
+    c.task.submit_qa.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_fast_path_ci_failure_rejects_before_transition(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    c = Choreographer(_deps())
+    stubs = _stub_fast_path(c, monkeypatch)
+    monkeypatch.setattr(
+        c,
+        "_fast_path_quality_verdict",
+        AsyncMock(
+            return_value=(
+                Envelope.invalid_state(message="ci red", remediate="fix"),
+                False,
+            )
+        ),
+    )
+    await c._i_am_done_fast_path(_ctx())
+    stubs.reject.assert_awaited_once()
+    c.task.submit_qa.assert_not_awaited()

tests/unit/runtime/test_possibilities_matrix_prompt.py

@@ -0,0 +1,149 @@
+"""The possibilities-matrix proxy rewrites a dev prompt to WORK_ALREADY_DONE when
+the flag is armed and the task already carries commits + an open PR.
+
+The proxy is a cheap sync read — it does NOT re-check the async DB gates
+(AC coverage, open findings); the server fast path ``_i_am_done_fast_path`` is
+the authority. The prompt just collapses a 3-5 turn re-derivation to one
+``i_am_done`` call. With the flag off (or the precondition unmet) the existing
+``WORKFLOW STATE`` mapping is byte-for-byte unchanged.
+"""
+
+from __future__ import annotations
+
+from contextlib import asynccontextmanager
+from typing import TYPE_CHECKING, Any
+from unittest.mock import AsyncMock, patch
+from uuid import uuid4
+
+import pytest
+from roboco.config import settings
+from roboco.runtime.orchestrator import AgentOrchestrator
+
+if TYPE_CHECKING:
+    from collections.abc import AsyncIterator
+
+
+def _orch() -> AgentOrchestrator:
+    orch = object.__new__(AgentOrchestrator)
+    orch._instances = {}
+    return orch
+
+
+def _task(**over: Any) -> dict[str, Any]:
+    base: dict[str, Any] = {
+        "id": str(uuid4()),
+        "title": "Ship it",
+        "status": "in_progress",
+        "plan": "did the thing",
+        "pr_created": True,
+        "commits": [{"sha": "abc"}],
+    }
+    base.update(over)
+    return base
+
+
+def _no_findings_db() -> tuple[Any, Any]:
+    """Patch the findings fetch out so a needs_revision prompt builds no DB."""
+
+    @asynccontextmanager
+    async def _fake_ctx() -> AsyncIterator[AsyncMock]:
+        yield AsyncMock()
+
+    repo = AsyncMock()
+    repo.list_for_task = AsyncMock(return_value=[])
+    return (
+        patch("roboco.db.base.get_db_context", _fake_ctx),
+        patch(
+            "roboco.services.repositories.review_findings.ReviewFindingsRepository",
+            return_value=repo,
+        ),
+    )
+
+
+@pytest.mark.asyncio
+async def test_armed_in_progress_with_pr_and_commits_is_work_already_done(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(settings, "possibilities_matrix_enabled", True)
+    prompt = await _orch()._build_dev_prompt(_task(status="in_progress"))
+    assert "WORK ALREADY DONE" in prompt
+    assert "i_am_done(task_id=" in prompt
+
+
+@pytest.mark.asyncio
+async def test_armed_claimed_with_pr_and_commits_is_work_already_done(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(settings, "possibilities_matrix_enabled", True)
+    prompt = await _orch()._build_dev_prompt(_task(status="claimed"))
+    assert "WORK ALREADY DONE" in prompt
+
+
+@pytest.mark.asyncio
+async def test_armed_verifying_with_pr_and_commits_is_work_already_done(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(settings, "possibilities_matrix_enabled", True)
+    prompt = await _orch()._build_dev_prompt(_task(status="verifying"))
+    assert "WORK ALREADY DONE" in prompt
+
+
+@pytest.mark.asyncio
+async def test_flag_off_leaves_in_progress_as_executing(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(settings, "possibilities_matrix_enabled", False)
+    prompt = await _orch()._build_dev_prompt(_task(status="in_progress"))
+    assert "WORK ALREADY DONE" not in prompt
+    assert "WORKFLOW STATE: EXECUTING" in prompt
+
+
+@pytest.mark.asyncio
+async def test_armed_without_pr_stays_on_standard_path(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(settings, "possibilities_matrix_enabled", True)
+    prompt = await _orch()._build_dev_prompt(
+        _task(status="in_progress", pr_created=False)
+    )
+    assert "WORK ALREADY DONE" not in prompt
+    assert "WORKFLOW STATE: EXECUTING" in prompt
+
+
+@pytest.mark.asyncio
+async def test_armed_without_commits_stays_on_standard_path(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(settings, "possibilities_matrix_enabled", True)
+    prompt = await _orch()._build_dev_prompt(_task(status="in_progress", commits=[]))
+    assert "WORK ALREADY DONE" not in prompt
+    assert "WORKFLOW STATE: EXECUTING" in prompt
+
+
+@pytest.mark.asyncio
+async def test_armed_needs_revision_is_unaffected(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # needs_revision is not in the proxy's status set — REVISION_REQUIRED wins.
+    monkeypatch.setattr(settings, "possibilities_matrix_enabled", True)
+    db_ctx, repo_ctx = _no_findings_db()
+    with db_ctx, repo_ctx:
+        prompt = await _orch()._build_dev_prompt(_task(status="needs_revision"))
+    assert "WORK ALREADY DONE" not in prompt
+    assert "REVISION REQUESTED" in prompt
+
+
+@pytest.mark.asyncio
+async def test_flag_off_claimed_no_plan_is_needs_plan(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(settings, "possibilities_matrix_enabled", False)
+    prompt = await _orch()._build_dev_prompt(
+        _task(status="claimed", plan=None, pr_created=True, commits=[{"sha": "x"}])
+    )
+    assert "WORK ALREADY DONE" not in prompt
+    assert "WORKFLOW STATE: NEEDS_PLAN" in prompt
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-q"])