diff --git a/tests/auto_agents/controller/test_master_ci_status_poll.py b/tests/auto_agents/controller/test_master_ci_status_poll.py
index 3cb78e323..c2b19598c 100644
--- a/tests/auto_agents/controller/test_master_ci_status_poll.py
+++ b/tests/auto_agents/controller/test_master_ci_status_poll.py
@@ -144,7 +144,10 @@ class TestEventRows:
                 text(
                     "SELECT event_type, from_state, to_state, payload "
                     "FROM controller_events "
-                    "WHERE event_type = 'ci-green'"
+                    # PD8: event_type now mirrors the state-machine
+                    # event names so consumers joining on event_type
+                    # don't see drift between sources.
+                    "WHERE event_type = 'ci_green'"
                 ),
             ).all()
         assert len(events) == 1
@@ -165,7 +168,7 @@ class TestEventRows:
             events = s.execute(
                 text(
                     "SELECT event_type, payload FROM controller_events "
-                    "WHERE event_type = 'ci-red'"
+                    "WHERE event_type = 'ci_red_retry_same_tier'"
                 ),
             ).all()
         assert len(events) == 1
@@ -174,16 +177,62 @@ class TestEventRows:
 
 
 class TestExtendedStateMapping:
-    def test_cancelled_treated_as_red(self, engine):
+    def test_cancelled_treated_as_wait(self, engine):
+        """CA8: ``cancelled`` is typically an operator action or CI-system
+        shutdown — not a real failure. Map to wait, not retry, so a
+        cancel doesn't burn a pickup_count slot on a healthy PR."""
+        wf_id = _seed_awaiting_ci_with_attempt(engine)
+        report = run_ci_status_poll_tick(
+            engine, owner="o", repo="r",
+            get_ci_status=lambda o, r, sha: {"state": "cancelled"},
+        )
+        assert report.workflows_waiting == 1
+        assert report.workflows_advanced_red == 0
+        with session_scope(engine) as s:
+            w = s.query(Workflow).filter_by(workflow_id=wf_id).one()
+            assert w.current_state == "AWAITING_CI"
+
+    def test_stale_treated_as_wait(self, engine):
+        """CA8: ``stale`` means a newer push superseded this run — wait
+        for the current run's status instead of retrying."""
+        wf_id = _seed_awaiting_ci_with_attempt(engine)
+        report = run_ci_status_poll_tick(
+            engine, owner="o", repo="r",
+            get_ci_status=lambda o, r, sha: {"state": "stale"},
+        )
+        assert report.workflows_waiting == 1
+        with session_scope(engine) as s:
+            w = s.query(Workflow).filter_by(workflow_id=wf_id).one()
+            assert w.current_state == "AWAITING_CI"
+
+    def test_timed_out_treated_as_red(self, engine):
+        """``timed_out`` IS a real failure (vs cancelled which is
+        operator-triggered) — keep as red retry."""
         wf_id = _seed_awaiting_ci_with_attempt(engine)
         run_ci_status_poll_tick(
             engine, owner="o", repo="r",
-            get_ci_status=lambda o, r, sha: {"state": "cancelled"},
+            get_ci_status=lambda o, r, sha: {"state": "timed_out"},
         )
         with session_scope(engine) as s:
             w = s.query(Workflow).filter_by(workflow_id=wf_id).one()
             assert w.current_state == "IMPLEMENTING"
 
+    def test_unknown_state_treated_as_wait_and_warns(self, engine, caplog):
+        """TE9: unknown Forgejo states log a WARNING so operators see
+        them, but treat as wait (don't transition on unrecognized data)."""
+        import logging
+        wf_id = _seed_awaiting_ci_with_attempt(engine)
+        with caplog.at_level(logging.WARNING):
+            report = run_ci_status_poll_tick(
+                engine, owner="o", repo="r",
+                get_ci_status=lambda o, r, sha: {"state": "asdf_unknown"},
+            )
+        assert report.workflows_waiting == 1
+        assert any(
+            "unknown CI state" in rec.message and "asdf_unknown" in rec.message
+            for rec in caplog.records
+        )
+
     def test_neutral_treated_as_green(self, engine):
         wf_id = _seed_awaiting_ci_with_attempt(engine)
         run_ci_status_poll_tick(
diff --git a/tests/auto_agents/controller/test_post_review_fixes.py b/tests/auto_agents/controller/test_post_review_fixes.py
new file mode 100644
index 000000000..544a73ff0
--- /dev/null
+++ b/tests/auto_agents/controller/test_post_review_fixes.py
@@ -0,0 +1,612 @@
+"""Tests for fixes from the post-Phase-1m adversarial review.
+
+Covers: MCP cross-session state reset, finalize_and_emit output_path
+precedence, inline-vs-MCP race protection, legacy_adapter quality
+upgrades, FORGEJO_URL suffix-strip, opencode.json registration parity,
+prompts cite the canonical poller path, agent_runner stale-file cleanup.
+"""
+from __future__ import annotations
+
+import io
+import json
+import logging
+import os
+import tempfile
+from pathlib import Path
+from unittest import mock
+
+import pytest
+
+
+# ─── helpers ─────────────────────────────────────────────────────────
+
+
+def _reset_builder(module):
+    module._STATE.started = False
+    module._STATE.finalized = False
+    module._STATE.fields.clear()
+    module._STATE.identity.clear()
+    module._STATE.audit.clear()
+    module._STATE.started_at = None
+    if hasattr(module, "_GATES_AUTO_ACK"):
+        module._GATES_AUTO_ACK.clear()
+    if hasattr(module, "_GATES_DISCUSSED"):
+        module._GATES_DISCUSSED.clear()
+
+
+# ─── CA2/PD5: cross-session _STATE reset ─────────────────────────────
+
+
+class TestCrossSessionReset:
+    def test_implementer_handles_two_attempts_in_one_subprocess(self):
+        """Phase 1m wires the MCPs as OpenCode local servers — the
+        subprocess persists across attempts. Without per-attempt reset
+        the second attempt's _start would raise 'already started'."""
+        from tools.controller.mcp import implementer_builder as ib
+        _reset_builder(ib)
+        try:
+            # Attempt 1 — full happy path.
+            assert ib.implementer_start(1, 100, 30, tier=0)["status"] == "ok"
+            ib.implementer_record_file_modified("a.py", 1, 0)
+            ib.implementer_record_commit("abc1234", "fix")
+            ib.implementer_set_outcome("resolved")
+            ib.implementer_set_confidence("high")
+            with tempfile.TemporaryDirectory() as td:
+                p1 = os.path.join(td, "a1.json")
+                r1 = ib.implementer_finalize(output_path=p1)
+                assert r1["status"] == "ok"
+                # Attempt 2 — different attempt_id. Should reset cleanly.
+                r2 = ib.implementer_start(1, 200, 30, tier=1)
+                assert r2["status"] == "ok", f"reset failed: {r2}"
+                # And the previous attempt's files_touched should be gone.
+                ib.implementer_record_file_modified("b.py", 2, 1)
+                ib.implementer_record_commit("def5678", "fix2")
+                ib.implementer_set_outcome("resolved")
+                ib.implementer_set_confidence("medium")
+                p2 = os.path.join(td, "a2.json")
+                r2f = ib.implementer_finalize(output_path=p2)
+                assert r2f["status"] == "ok", r2f
+                with open(p2) as f:
+                    out = json.loads(f.read())
+                assert out["files_touched"] == ["b.py"]
+                assert out["commit_shas"] == ["def5678"]
+                assert out["used_tier"] == 1
+        finally:
+            _reset_builder(ib)
+
+    def test_implementer_rejects_intra_session_double_start(self):
+        """Same attempt_id calling _start twice IS a real bug — the
+        reset should NOT swallow it."""
+        from tools.controller.mcp import implementer_builder as ib
+        _reset_builder(ib)
+        try:
+            assert ib.implementer_start(1, 100, 30, tier=0)["status"] == "ok"
+            r = ib.implementer_start(1, 100, 30, tier=0)  # same attempt_id
+            assert "error" in r
+            assert "once per attempt" in r["error"]
+        finally:
+            _reset_builder(ib)
+
+    def test_reviewer_clears_gate_dicts_on_cross_attempt_reset(self):
+        from tools.controller.mcp import reviewer_builder as rb
+        _reset_builder(rb)
+        try:
+            rb.reviewer_start(
+                workflow_id=1, attempt_id=100, pr_number=30,
+                head_sha="abc1234",
+                gates=[{"name": "g1", "status": "passed"}],
+            )
+            assert "g1" in rb._GATES_AUTO_ACK
+            # New attempt should wipe the gate cache.
+            rb.reviewer_start(
+                workflow_id=1, attempt_id=200, pr_number=30,
+                head_sha="def5678",
+                gates=[{"name": "g2", "status": "failed"}],  # different status ok
+            )
+            assert "g1" not in rb._GATES_AUTO_ACK
+            assert "g2" in rb._GATES_AUTO_ACK
+        finally:
+            _reset_builder(rb)
+
+    def test_estimator_has_explicit_start_tool(self):
+        """The prompt advertises estimator_start; this test pins it
+        exists + handles cross-session reset."""
+        from tools.controller.mcp import estimator_builder as eb
+        _reset_builder(eb)
+        try:
+            r = eb.estimator_start(workflow_id=1, attempt_id=100)
+            assert r["status"] == "ok"
+            # Second attempt different id resets cleanly.
+            r2 = eb.estimator_start(workflow_id=1, attempt_id=200)
+            assert r2["status"] == "ok"
+        finally:
+            _reset_builder(eb)
+
+    def test_summarizer_has_explicit_start_tool(self):
+        from tools.controller.mcp import summarizer_builder as sb
+        _reset_builder(sb)
+        try:
+            r = sb.summarizer_start(workflow_id=1, attempt_id=100)
+            assert r["status"] == "ok"
+            r2 = sb.summarizer_start(workflow_id=1, attempt_id=200)
+            assert r2["status"] == "ok"
+        finally:
+            _reset_builder(sb)
+
+
+# ─── TE1/TE2: finalize_and_emit output_path precedence + errors ─────
+
+
+class TestFinalizeEmitOutputPath:
+    def test_output_path_arg_wins_over_env(self, monkeypatch):
+        from tools.controller.mcp import implementer_builder as ib
+        _reset_builder(ib)
+        try:
+            with tempfile.TemporaryDirectory() as td:
+                arg_path = os.path.join(td, "arg.json")
+                env_path = os.path.join(td, "env.json")
+                monkeypatch.setenv(
+                    "CONTROLLER_CANONICAL_OUTPUT_PATH", env_path,
+                )
+                ib.implementer_start(1, 1, 30, tier=0)
+                ib.implementer_record_file_modified("x.py", 1, 0)
+                ib.implementer_record_commit("abc1234", "fix")
+                ib.implementer_set_outcome("resolved")
+                ib.implementer_set_confidence("high")
+                r = ib.implementer_finalize(output_path=arg_path)
+                assert r["status"] == "ok"
+                assert r["wrote_to"] == arg_path
+                assert os.path.exists(arg_path)
+                assert not os.path.exists(env_path)
+        finally:
+            _reset_builder(ib)
+            monkeypatch.delenv(
+                "CONTROLLER_CANONICAL_OUTPUT_PATH", raising=False,
+            )
+
+    def test_env_used_when_no_arg(self, monkeypatch):
+        from tools.controller.mcp import implementer_builder as ib
+        _reset_builder(ib)
+        try:
+            with tempfile.TemporaryDirectory() as td:
+                env_path = os.path.join(td, "env.json")
+                monkeypatch.setenv(
+                    "CONTROLLER_CANONICAL_OUTPUT_PATH", env_path,
+                )
+                ib.implementer_start(1, 1, 30, tier=0)
+                ib.implementer_record_file_modified("x.py", 1, 0)
+                ib.implementer_record_commit("abc1234", "fix")
+                ib.implementer_set_outcome("resolved")
+                ib.implementer_set_confidence("high")
+                r = ib.implementer_finalize()
+                assert r["wrote_to"] == env_path
+                assert os.path.exists(env_path)
+        finally:
+            _reset_builder(ib)
+            monkeypatch.delenv(
+                "CONTROLLER_CANONICAL_OUTPUT_PATH", raising=False,
+            )
+
+    def test_creates_parent_dir(self):
+        from tools.controller.mcp import implementer_builder as ib
+        _reset_builder(ib)
+        try:
+            with tempfile.TemporaryDirectory() as td:
+                nested = os.path.join(td, "a", "b", "c", "out.json")
+                ib.implementer_start(1, 1, 30, tier=0)
+                ib.implementer_record_file_modified("x.py", 1, 0)
+                ib.implementer_record_commit("abc1234", "fix")
+                ib.implementer_set_outcome("resolved")
+                ib.implementer_set_confidence("high")
+                r = ib.implementer_finalize(output_path=nested)
+                assert r["status"] == "ok"
+                assert os.path.exists(nested)
+        finally:
+            _reset_builder(ib)
+
+    def test_rejects_relative_output_path(self):
+        from tools.controller.mcp import implementer_builder as ib
+        _reset_builder(ib)
+        try:
+            ib.implementer_start(1, 1, 30, tier=0)
+            ib.implementer_record_file_modified("x.py", 1, 0)
+            ib.implementer_record_commit("abc1234", "fix")
+            ib.implementer_set_outcome("resolved")
+            ib.implementer_set_confidence("high")
+            r = ib.implementer_finalize(output_path="rel/out.json")
+            assert "error" in r
+            assert "absolute" in r["error"].lower()
+            # CA12: failed write must NOT mark state.finalized.
+            assert not ib._STATE.finalized
+        finally:
+            _reset_builder(ib)
+
+    def test_rejects_nul_byte_in_path(self):
+        from tools.controller.mcp import implementer_builder as ib
+        _reset_builder(ib)
+        try:
+            ib.implementer_start(1, 1, 30, tier=0)
+            ib.implementer_record_file_modified("x.py", 1, 0)
+            ib.implementer_record_commit("abc1234", "fix")
+            ib.implementer_set_outcome("resolved")
+            ib.implementer_set_confidence("high")
+            r = ib.implementer_finalize(output_path="/tmp/foo\x00bar.json")
+            assert "error" in r
+            assert "NUL" in r["error"]
+        finally:
+            _reset_builder(ib)
+
+    def test_failed_write_leaves_state_unfinalized_so_retry_possible(self):
+        """CA12: previously state.finalized was set BEFORE the write —
+        if disk-full / OSError fired, the agent couldn't retry within
+        the session. Verify the order is now write-first."""
+        from tools.controller.mcp import implementer_builder as ib
+        _reset_builder(ib)
+        try:
+            ib.implementer_start(1, 1, 30, tier=0)
+            ib.implementer_record_file_modified("x.py", 1, 0)
+            ib.implementer_record_commit("abc1234", "fix")
+            ib.implementer_set_outcome("resolved")
+            ib.implementer_set_confidence("high")
+            # Force a write failure by pointing at a path whose parent
+            # is a regular file.
+            with tempfile.TemporaryDirectory() as td:
+                fake_parent = os.path.join(td, "iamfile")
+                with open(fake_parent, "w") as f:
+                    f.write("not a dir")
+                r = ib.implementer_finalize(
+                    output_path=os.path.join(fake_parent, "out.json"),
+                )
+                assert "error" in r
+                assert not ib._STATE.finalized
+        finally:
+            _reset_builder(ib)
+
+
+# ─── TE5/TE6: legacy_adapter quality coverage ────────────────────────
+
+
+class TestLegacyAdapterExtensions:
+    """Coverage for the M3 adapter upgrades + roles previously untested."""
+
+    def test_estimator_tier_above_2_clamped(self):
+        from tools.controller.worker.legacy_adapter import adapt_to_v1
+        legacy = {"recommended_tier": 5, "is_confident": True,
+                  "reasoning": "deep refactor needed"}
+        v1 = adapt_to_v1(
+            "estimator", legacy, tier=None, wallclock_seconds=1.0,
+        )
+        assert v1["recommended_tier"] == 2
+
+    def test_estimator_tier_negative_clamped(self):
+        from tools.controller.worker.legacy_adapter import adapt_to_v1
+        legacy = {"recommended_tier": -1, "is_confident": True,
+                  "reasoning": "trivial"}
+        v1 = adapt_to_v1(
+            "estimator", legacy, tier=None, wallclock_seconds=1.0,
+        )
+        assert v1["recommended_tier"] == 0
+
+    def test_implementer_blocker_capped_at_4096_chars(self):
+        from tools.controller.worker.legacy_adapter import adapt_to_v1
+        huge = "X" * 100_000
+        legacy = {"outcome": "blocked", "blockers": [huge]}
+        v1 = adapt_to_v1(
+            "implementer", legacy, tier=0, wallclock_seconds=1.0,
+        )
+        assert len(v1["blockers"][0]) == 4096
+
+    def test_implementer_non_string_commits_dropped_with_warning(self, caplog):
+        from tools.controller.worker.legacy_adapter import adapt_to_v1
+        legacy = {
+            "outcome": "resolved",
+            "files_touched": ["a.py"],
+            "commit_shas": ["abc1234", 12345, None, "def5678"],
+        }
+        with caplog.at_level(logging.WARNING):
+            v1 = adapt_to_v1(
+                "implementer", legacy, tier=0, wallclock_seconds=1.0,
+            )
+        assert v1["commit_shas"] == ["abc1234", "def5678"]
+        assert any(
+            "non-string commit" in rec.message for rec in caplog.records
+        )
+
+    def test_reviewer_blocking_issues_strings_coerced_to_dicts(self):
+        from tools.controller.contracts.v1 import ReviewerOutputV1
+        from tools.controller.worker.legacy_adapter import adapt_to_v1
+        legacy = {
+            "verdict": "request-changes",
+            "blocking_issues": ["use a for loop", "tests missing"],
+        }
+        v1 = adapt_to_v1(
+            "reviewer", legacy, tier=None, wallclock_seconds=1.0,
+        )
+        parsed = ReviewerOutputV1.model_validate(v1)
+        assert len(parsed.blocking_issues) == 2
+        assert parsed.blocking_issues[0].description == "use a for loop"
+        assert parsed.blocking_issues[0].severity == "error"
+
+    def test_conflict_resolver_full_legacy_roundtrip(self):
+        from tools.controller.contracts.v1 import ConflictResolverOutputV1
+        from tools.controller.worker.legacy_adapter import adapt_to_v1
+        legacy = {
+            "outcome": "resolved",
+            "files_modified": ["a.py", "b.py"],
+            "commit_sha": "abc1234",  # singular fallback (PD22)
+            "new_head_sha": "def5678",
+            "reasoning": "kept ours for a, three-way for b",
+            "is_confident": True,
+        }
+        v1 = adapt_to_v1(
+            "conflict_resolver", legacy, tier=1, wallclock_seconds=2.0,
+        )
+        parsed = ConflictResolverOutputV1.model_validate(v1)
+        assert parsed.outcome == "resolved"
+        assert parsed.commit_shas == ["abc1234"]
+        assert parsed.new_head_sha == "def5678"
+        assert parsed.confidence == "high"
+
+    def test_conflict_resolver_clears_new_head_on_non_resolved(self):
+        from tools.controller.worker.legacy_adapter import adapt_to_v1
+        legacy = {
+            "outcome": "irreconcilable",
+            "new_head_sha": "stale_sha",
+            "reasoning": "could not resolve cleanly",
+        }
+        v1 = adapt_to_v1(
+            "conflict_resolver", legacy, tier=1, wallclock_seconds=1.0,
+        )
+        assert v1["new_head_sha"] is None
+
+    def test_summarizer_too_short_summary_padded(self):
+        from tools.controller.contracts.v1 import SummarizerOutputV1
+        from tools.controller.worker.legacy_adapter import adapt_to_v1
+        legacy = {"summary": "short", "covers_through_attempt": 1}
+        v1 = adapt_to_v1(
+            "summarizer", legacy, tier=None, wallclock_seconds=1.0,
+        )
+        parsed = SummarizerOutputV1.model_validate(v1)
+        assert len(parsed.summary) >= 50
+        assert "adapter-padded" in parsed.summary
+
+    def test_summarizer_normal_path(self):
+        from tools.controller.contracts.v1 import SummarizerOutputV1
+        from tools.controller.worker.legacy_adapter import adapt_to_v1
+        legacy = {
+            "summary": "A" * 200, "covers_through_attempt": 3,
+        }
+        v1 = adapt_to_v1(
+            "summarizer", legacy, tier=None, wallclock_seconds=1.0,
+        )
+        parsed = SummarizerOutputV1.model_validate(v1)
+        assert parsed.covers_through_attempt == 3
+
+    def test_invalid_confidence_logs_warning(self, caplog):
+        from tools.controller.worker.legacy_adapter import adapt_to_v1
+        legacy = {
+            "outcome": "resolved",
+            "files_touched": ["a.py"],
+            "commit_shas": ["abc1234"],
+            "confidence": "super-strong",  # not in {high,medium,low}
+        }
+        with caplog.at_level(logging.WARNING):
+            v1 = adapt_to_v1(
+                "implementer", legacy, tier=0, wallclock_seconds=1.0,
+            )
+        assert v1["confidence"] == "medium"
+        assert any(
+            "unrecognized confidence" in rec.message
+            for rec in caplog.records
+        )
+
+    def test_v1_passthrough_does_not_log_fallback_warning(self, caplog):
+        """Already-V1 payloads should NOT emit the migration nag log."""
+        from tools.controller.worker.legacy_adapter import adapt_to_v1
+        v1_in = {
+            "output_version": "V1", "recommended_tier": 0,
+            "is_metadata_only": False, "confidence": "high",
+            "reasoning": "ok", "wallclock_seconds": 1.0,
+        }
+        with caplog.at_level(logging.INFO):
+            out = adapt_to_v1(
+                "estimator", v1_in, tier=None, wallclock_seconds=1.0,
+            )
+        assert out is v1_in
+        assert not any(
+            "chat-JSON fallback" in rec.message for rec in caplog.records
+        )
+
+
+# ─── TE11: opencode.json schema parity ───────────────────────────────
+
+
+class TestOpenCodeJsonRegistration:
+    def test_all_5_role_builders_registered(self):
+        """The prompts.py output-contract sections name MCPs by string
+        ("implementer-response-builder" etc.). Verify each named MCP
+        is actually registered in opencode.json — typo here would
+        silently fall back to chat-JSON + adapter."""
+        oc_path = Path(__file__).resolve().parents[3] / ".opencode" / "opencode.json"
+        with open(oc_path) as f:
+            oc = json.load(f)
+        mcps = oc.get("mcp", {})
+        for role in [
+            "implementer", "reviewer", "estimator",
+            "conflict-resolver", "summarizer",
+        ]:
+            key = f"{role}-response-builder"
+            assert key in mcps, (
+                f"{key!r} missing from opencode.json — agents will not "
+                f"have access to the canonical-emission MCP"
+            )
+            entry = mcps[key]
+            assert entry.get("enabled") is True
+            assert "-m" in entry.get("command", [])
+            # Each command should reference the matching builder module.
+            cmdline = " ".join(entry["command"])
+            module_role = role.replace("-", "_")
+            assert f"{module_role}_builder" in cmdline
+
+
+# ─── TE12/TE13: prompts cite the canonical poller path + DO NOT emit ─
+
+
+class TestPromptOutputContractSections:
+    """The agent_runner's _wait_for_canonical_output polls
+    ``{workspace_dir}/{role}_output.json`` as a fallback. The prompt
+    MUST advertise the SAME path so MCP-written and inline-written
+    files land on the canonical poller.
+    """
+
+    @pytest.mark.parametrize("role,expected_filename", [
+        ("implementer", "implementer_output.json"),
+        ("reviewer", "reviewer_output.json"),
+        ("estimator", "estimator_output.json"),
+        ("conflict_resolver", "conflict_resolver_output.json"),
+        ("summarizer", "summarizer_output.json"),
+    ])
+    def test_prompt_mentions_canonical_output_filename(
+        self, role, expected_filename,
+    ):
+        from tools.controller.worker.prompts import build_prompt
+        tier = 0 if role in {"implementer", "conflict_resolver"} else None
+        ip = {
+            "workspace_dir": "/tmp/ws", "workflow_id": 1, "pr_number": 30,
+            "head_sha": "abc1234", "head_ref": "feat/x", "base_branch": "main",
+            "diff_summary": "...", "newly_aged_out_attempt": {},
+        }
+        p = build_prompt(role, tier, ip)
+        assert expected_filename in p, (
+            f"role={role}: prompt does not mention {expected_filename}; "
+            "fallback poller and prompt path are out of sync"
+        )
+
+    @pytest.mark.parametrize("role,tier", [
+        ("implementer", 0), ("reviewer", None), ("estimator", None),
+        ("conflict_resolver", 0), ("summarizer", None),
+    ])
+    def test_prompt_says_do_not_emit_chat_json(self, role, tier):
+        from tools.controller.worker.prompts import build_prompt
+        ip = {
+            "workspace_dir": "/tmp/ws", "workflow_id": 1, "pr_number": 30,
+            "head_sha": "abc1234", "head_ref": "feat/x", "base_branch": "main",
+            "diff_summary": "...", "newly_aged_out_attempt": {},
+        }
+        p = build_prompt(role, tier, ip)
+        assert "DO NOT emit a JSON object" in p, (
+            f"role={role}: prompt missing the 'do not emit chat-JSON' "
+            "directive — agents will continue emitting legacy shape and "
+            "the canonical channel won't win"
+        )
+
+
+# ─── PD4: FORGEJO_URL/API_BASE suffix-strip ──────────────────────────
+
+
+class TestForgejoUrlSuffixStrip:
+    """The previous ``.rstrip('/api/v1')`` did a character-set strip
+    that happened to work for typical hosts. The new explicit suffix
+    strip is correctness-equivalent for those AND safe for hosts where
+    the final char is in the set."""
+
+    @pytest.mark.parametrize("api_base,expected", [
+        ("https://forge.example.com/api/v1", "https://forge.example.com"),
+        ("https://api.example.com/api/v1", "https://api.example.com"),
+        ("https://x.com:8080/api/v1", "https://x.com:8080"),
+        ("https://forge.example.com/api/v1/", "https://forge.example.com"),
+        # The catastrophic case for rstrip: trailing chars in {a,p,i,v,1,/}
+        # AFTER the suffix gets correctly stripped. With suffix strip, no chars
+        # past /api/v1 are touched.
+        ("https://api.example.org/v1/api/v1", "https://api.example.org/v1"),
+        ("", ""),  # empty base
+    ])
+    def test_strip(self, api_base, expected):
+        # Replicate the closure inside worker/__main__.py without import-
+        # exposing it (it's a local def). Test the algorithm.
+        def _strip_api_suffix(s: str) -> str:
+            s = s.rstrip("/")
+            if s.endswith("/api/v1"):
+                s = s[: -len("/api/v1")]
+            return s
+        assert _strip_api_suffix(api_base) == expected
+
+
+# ─── CA1: stale-file cleanup between attempts ────────────────────────
+
+
+class TestStaleOutputFileCleanup:
+    def test_canonical_poller_does_not_read_prior_attempt_file(
+        self, monkeypatch, tmp_path,
+    ):
+        """Before the fix, a per-PR workspace dir is reused across
+        attempts and ``{workspace_dir}/{role}_output.json`` from
+        attempt N-1 was visible to attempt N's poller as a valid
+        first-non-empty read. Verify _wait_for_canonical_output reads
+        from the freshly-written file, not the pre-existing one — but
+        the real defense is the unlink-before-session that
+        production_agent_runner now does. Simulate by pre-creating
+        the file then asserting the runner's pre-session cleanup
+        unlinks it before passing to the session."""
+        from tools.controller.worker import agent_runner
+
+        workspace = tmp_path / "pr-o-r-30"
+        workspace.mkdir()
+        stale_path = workspace / "implementer_output.json"
+        stale_path.write_text(json.dumps({
+            "output_version": "V1", "outcome": "resolved",
+            "files_touched": ["stale.py"],
+            "commit_shas": ["staleSHA"], "confidence": "high",
+            "blockers": [], "used_tier": 0, "wallclock_seconds": 1.0,
+        }))
+        assert stale_path.exists()
+
+        # Mock everything below the unlink to verify the cleanup happens.
+        # We use a session runner that asserts the file is gone before
+        # it returns + writes a fresh V1 to the same path.
+        def fake_session(
+            *, role, tier, input_payload, mcp_process, attempt_id,
+            instance_id, lost_lock_check, inline_output_callback=None,
+        ):
+            assert not stale_path.exists(), (
+                "stale file should have been unlinked before session ran"
+            )
+            # Simulate the agent writing fresh canonical output.
+            fresh = {
+                "output_version": "V1", "outcome": "noop",
+                "files_touched": [], "commit_shas": [],
+                "confidence": "high", "blockers": [],
+                "used_tier": 0, "wallclock_seconds": 1.0,
+            }
+            stale_path.write_text(json.dumps(fresh))
+
+        # Patch _spawn_mcp_subprocess + _terminate_mcp_subprocess to no-ops.
+        class _FakeSpawn:
+            class process:
+                @staticmethod
+                def poll(): return None
+                pid = 12345
+            stdout_buffer: list = []
+            class reader_thread:
+                @staticmethod
+                def join(timeout=None): pass
+
+        monkeypatch.setattr(
+            agent_runner, "_spawn_mcp_subprocess",
+            lambda *a, **kw: _FakeSpawn(),
+        )
+        monkeypatch.setattr(
+            agent_runner, "_terminate_mcp_subprocess", lambda spawn: None,
+        )
+
+        out = agent_runner.production_agent_runner(
+            attempt_id=999,
+            role="implementer", tier=0,
+            input_payload={"workspace_dir": str(workspace)},
+            instance_id="test",
+            lost_lock_check=lambda: False,
+            run_opencode_session=fake_session,
+            workspace_dir=workspace,
+            finalize_timeout_s=5.0,
+        )
+        assert out["outcome"] == "noop"  # fresh write, not the stale "resolved"
diff --git a/tests/auto_agents/controller/test_worker_prompts.py b/tests/auto_agents/controller/test_worker_prompts.py
index 9ad681262..080b2cfd1 100644
--- a/tests/auto_agents/controller/test_worker_prompts.py
+++ b/tests/auto_agents/controller/test_worker_prompts.py
@@ -481,9 +481,19 @@ class TestBuildPromptDispatcher:
         with pytest.raises(ValueError, match="implementer prompt requires tier"):
             build_prompt("implementer", None, _impl_input())
 
-    def test_conflict_resolver_requires_tier(self):
-        with pytest.raises(ValueError, match="conflict_resolver prompt requires tier"):
-            build_prompt("conflict_resolver", None, _conflict_input())
+    def test_conflict_resolver_defaults_tier_when_missing(self, caplog):
+        """PD13: scheduler always sets tier for conflict_resolver but
+        defend against contract drift — defaulting to tier=1 with a
+        warning beats burning a pickup_count + stack trace inside
+        production_agent_runner."""
+        import logging
+        with caplog.at_level(logging.WARNING):
+            p = build_prompt("conflict_resolver", None, _conflict_input())
+        assert "# Conflict Resolver — Tier 1" in p
+        assert any(
+            "conflict_resolver prompt called without tier" in rec.message
+            for rec in caplog.records
+        )
 
     def test_unknown_role_raises(self):
         with pytest.raises(ValueError, match="unknown role"):
diff --git a/tools/controller/master/ci_status_poll.py b/tools/controller/master/ci_status_poll.py
index afd762f27..8101fa671 100644
--- a/tools/controller/master/ci_status_poll.py
+++ b/tools/controller/master/ci_status_poll.py
@@ -59,6 +59,16 @@ GetCIStatusCallback = Callable[[str, str, str], dict | None]
 
 # Map Forgejo combined-status state → state-machine event.
 # None = no-op (wait for next tick).
+#
+# CA8: ``cancelled`` and ``stale`` are NOT genuine CI failures —
+# ``cancelled`` is typically an operator hitting "cancel job" or CI
+# system shutting down; ``stale`` is when a new push superseded the
+# run on a different branch (common during force-pushes). Treating
+# these as ci_red burns a ``pickup_count`` slot on a healthy PR,
+# which combined with MAX_PICKUPS pushes the workflow to STUCK after
+# a few cancels. Map them to None (wait for the next tick) so the
+# poller picks up the eventual real status (or operator can re-run).
+# ``timed_out`` IS a real failure — keep that as red.
 _STATE_TO_EVENT: dict[str | None, str | None] = {
     "success": "ci_green",
     "failure": "ci_red_retry_same_tier",
@@ -69,10 +79,10 @@ _STATE_TO_EVENT: dict[str | None, str | None] = {
     "warning": "ci_green",  # advisory; treat as passed
     "neutral": "ci_green",
     "skipped": "ci_green",
-    "cancelled": "ci_red_retry_same_tier",
+    "cancelled": None,  # CA8 — operator/CI-system action, not a failure
     "timed_out": "ci_red_retry_same_tier",
     "action_required": None,  # human intervention needed; wait
-    "stale": "ci_red_retry_same_tier",
+    "stale": None,  # CA8 — superseded run, wait for current to land
     None: None,
 }
 
@@ -102,16 +112,28 @@ def run_ci_status_poll_tick(
     now = datetime.now(timezone.utc)
 
     with session_scope(engine) as session:
-        # Find AWAITING_CI workflows + their latest implementer
-        # attempt's head_sha_after. The push SHA is what we need to
-        # query Forgejo's CI status against.
+        # Find AWAITING_CI workflows + the SHA whose CI to poll.
+        #
+        # CA6: both implementer AND conflict_resolver push commits.
+        # The previous query filtered ``a.role = 'implementer'`` so
+        # workflows that re-entered AWAITING_CI after CONFLICT_RESOLVING
+        # → IMPLEMENTING used a stale pre-conflict SHA. Union both roles
+        # and pick whichever attempt has the highest attempt_number.
+        #
+        # PD9: filter on ``a.outcome = 'resolved'`` (implementer) or
+        # ``a.outcome = 'resolved'`` (conflict_resolver) — a blocked
+        # attempt has ``head_sha_after`` equal to ``head_sha_before``
+        # (no push happened), so the previous query would treat the
+        # stale pre-blocked SHA as the "latest push" and false-green
+        # advance to REVIEWING for code that was never re-pushed.
         rows = session.execute(
             text(
                 "SELECT w.workflow_id, w.entity_number, "
                 "       (SELECT a.head_sha_after FROM workflow_attempts a "
                 "         WHERE a.workflow_id = w.workflow_id "
-                "           AND a.role = 'implementer' "
+                "           AND a.role IN ('implementer', 'conflict_resolver') "
                 "           AND a.status = 'complete' "
+                "           AND a.outcome = 'resolved' "
                 "           AND a.head_sha_after IS NOT NULL "
                 "         ORDER BY a.attempt_number DESC LIMIT 1) AS head_sha "
                 "  FROM workflows w "
@@ -150,6 +172,17 @@ def run_ci_status_poll_tick(
                 continue
 
             ci_state = ci.get("state")
+            # Warn on unknown Forgejo states so a new state string
+            # surfaces in operator logs instead of being silently
+            # treated as "pending" (TE9: previously indistinguishable
+            # from real pending — operators got no signal).
+            if ci_state not in _STATE_TO_EVENT:
+                logger.warning(
+                    "ci_status_poll: workflow %s reported unknown CI state "
+                    "%r (head=%s); treating as wait — add to _STATE_TO_EVENT "
+                    "if recurring",
+                    wf_id, ci_state, (head_sha or "")[:12],
+                )
             event = _STATE_TO_EVENT.get(ci_state)
             if event is None:
                 # Pending / queued / in_progress / unknown → wait.
@@ -199,9 +232,11 @@ def run_ci_status_poll_tick(
                 ),
                 {
                     "wf_id": wf_id, "ts": now,
-                    "event_type": (
-                        "ci-green" if event == "ci_green" else "ci-red"
-                    ),
+                    # PD8: use the state-machine event name (ci_green /
+                    # ci_red_retry_same_tier) directly so consumers
+                    # joining on event_type don't see schema drift
+                    # between sources.
+                    "event_type": event,
                     "to_state": new_state,
                     "payload": json.dumps({
                         "reason": event,
diff --git a/tools/controller/mcp/_builder_base.py b/tools/controller/mcp/_builder_base.py
index 8a5c6106d..1338ee967 100644
--- a/tools/controller/mcp/_builder_base.py
+++ b/tools/controller/mcp/_builder_base.py
@@ -58,6 +58,44 @@ class BuilderState:
         if self.finalized:
             raise BuilderError("response already finalized; no further mutations allowed")
 
+    def reset_for_new_attempt(self) -> None:
+        """Wipe per-attempt accumulator so a fresh ``*_start`` call sees
+        a clean slate. Called by every ``{role}_start`` body when the
+        arriving attempt_id differs from the stored identity's.
+
+        Required because the opencode.json-registered local MCP servers
+        are reused across multiple OpenCode sessions (each session ==
+        one controller attempt). Without reset the second attempt's
+        ``_start`` would either raise "already started" or inherit the
+        first attempt's accumulated fields/identity.
+
+        Force-reset semantics: if the prior attempt was interrupted
+        (timeout / lost-lock / OpenCode hang) it can leave
+        ``started=True, finalized=False`` indefinitely. Refusing to
+        reset would permanently wedge the MCP for the rest of the
+        OpenCode server's lifetime. Log a WARNING when this happens
+        so operators see abandoned attempts but proceed with reset.
+
+        Intra-session double-``_start`` (same attempt_id called twice)
+        is detected separately by callers AFTER reset, via the
+        ``_STATE.started`` check.
+        """
+        if self.started and not self.finalized:
+            import logging
+            logging.getLogger(__name__).warning(
+                "builder reset_for_new_attempt: prior attempt was not "
+                "finalized (identity=%s); force-resetting state. The "
+                "previous attempt likely hit a timeout or lost lock.",
+                self.identity,
+            )
+        with self._lock:
+            self.started = False
+            self.finalized = False
+            self.identity = {}
+            self.fields = {}
+            self.audit = []
+            self.started_at = None
+
     def record(self, tool: str, args: dict[str, Any]) -> None:
         """Append an audit entry. Args are str-coerced to avoid
         carrying large blobs (e.g. raw_log_excerpt) into the audit
@@ -160,30 +198,56 @@ def finalize_and_emit(
             validation_error=str(exc),
         )
     canonical = model.model_dump_json()
-    # Mark BEFORE emitting so duplicate finalize calls return error.
-    state.finalized = True
     out_path = output_path or os.environ.get("CONTROLLER_CANONICAL_OUTPUT_PATH")
     if out_path:
+        # CA3 path-injection defense: refuse paths that look obviously
+        # adversarial. Real defense lives at the workspace boundary
+        # (worker validates owner/repo + workspace_root is under /tmp),
+        # but a defense-in-depth check here surfaces bugs early.
+        if "\x00" in out_path:
+            return err(f"output_path contains NUL byte; refusing to write")
+        if not os.path.isabs(out_path):
+            return err(
+                f"output_path must be absolute; got {out_path!r}"
+            )
         # Production: file-based clean channel.
+        parent = os.path.dirname(out_path)
+        if parent and os.path.exists(parent) and not os.path.isdir(parent):
+            return err(
+                f"parent of output_path={out_path!r} exists but is a file, "
+                "not a directory; refusing to write"
+            )
         try:
             # Ensure the parent dir exists so the controller doesn't
             # have to pre-create it; same idempotent guarantee as
             # tempfile.mkstemp gave.
-            parent = os.path.dirname(out_path)
             if parent:
                 os.makedirs(parent, exist_ok=True)
-            with open(out_path, "w", encoding="utf-8") as f:
+            # Atomic write: write to .tmp then os.replace so a concurrent
+            # reader (the worker's _wait_for_canonical_output poller)
+            # never sees a half-written file.
+            tmp_path = f"{out_path}.tmp"
+            with open(tmp_path, "w", encoding="utf-8") as f:
                 f.write(canonical)
                 f.write("\n")
                 f.flush()
                 os.fsync(f.fileno())
+            os.replace(tmp_path, out_path)
         except OSError as exc:
+            # Best-effort cleanup of the .tmp leftover.
+            try:
+                os.unlink(f"{out_path}.tmp")
+            except OSError:
+                pass
             return err(f"failed to write canonical output to {out_path}: {exc}")
     else:
         # Direct-call tests: emit to stdout (captured by capsys).
         sys.stdout.write(canonical)
         sys.stdout.write("\n")
         sys.stdout.flush()
+    # Mark AFTER successful emission so a failed write allows the agent
+    # to retry finalize within the same session (CA12).
+    state.finalized = True
     return ok(
         committed=True, output_bytes=len(canonical),
         wrote_to=out_path or "stdout",
diff --git a/tools/controller/mcp/conflict_resolver_builder.py b/tools/controller/mcp/conflict_resolver_builder.py
index 67d14caf7..155a4e8cc 100644
--- a/tools/controller/mcp/conflict_resolver_builder.py
+++ b/tools/controller/mcp/conflict_resolver_builder.py
@@ -46,6 +46,10 @@ def conflict_start(
     args = {"workflow_id": workflow_id, "attempt_id": attempt_id, "pr_number": pr_number}
 
     def body() -> dict[str, Any]:
+        # Cross-session reset for OpenCode-reused MCP subprocesses.
+        prior_attempt = _STATE.identity.get("attempt_id")
+        if prior_attempt is not None and prior_attempt != attempt_id:
+            _STATE.reset_for_new_attempt()
         if _STATE.started:
             raise BuilderError("conflict_start may only be called once per attempt")
         _STATE.started = True
diff --git a/tools/controller/mcp/estimator_builder.py b/tools/controller/mcp/estimator_builder.py
index 0a3c6a686..9348eded9 100644
--- a/tools/controller/mcp/estimator_builder.py
+++ b/tools/controller/mcp/estimator_builder.py
@@ -34,6 +34,42 @@ _TIER_VALUES = {0, 1, 2}
 _CONFIDENCE_VALUES = {"high", "medium", "low"}
 
 
+@server.tool()
+def estimator_start(
+    workflow_id: int, attempt_id: int,
+    pr_number: int | None = None, head_sha: str | None = None,
+) -> dict[str, Any]:
+    """Initialize the estimator session. RECOMMENDED first call.
+
+    Setters auto-start without this for backward compat, but calling
+    this explicitly is the contract the prompt advertises + lets the
+    builder detect cross-session reuse (OpenCode reuses the local MCP
+    subprocess across sessions; without explicit start, attempt-id
+    discrimination is impossible)."""
+    args = {
+        "workflow_id": workflow_id, "attempt_id": attempt_id,
+        "pr_number": pr_number, "head_sha": head_sha,
+    }
+
+    def body() -> dict[str, Any]:
+        prior_attempt = _STATE.identity.get("attempt_id")
+        if prior_attempt is not None and prior_attempt != attempt_id:
+            _STATE.reset_for_new_attempt()
+        if _STATE.started:
+            raise BuilderError(
+                "estimator_start may only be called once per attempt"
+            )
+        _STATE.started = True
+        _STATE.started_at = datetime.now(timezone.utc)
+        _STATE.identity = {
+            "workflow_id": workflow_id, "attempt_id": attempt_id,
+            "pr_number": pr_number, "head_sha": head_sha,
+        }
+        return ok()
+
+    return call_with_invariants(_STATE, "estimator_start", body, args)
+
+
 @server.tool()
 def estimator_set_recommended_tier(tier: int) -> dict[str, Any]:
     args = {"tier": tier}
diff --git a/tools/controller/mcp/implementer_builder.py b/tools/controller/mcp/implementer_builder.py
index 087d916ea..cb6036f77 100644
--- a/tools/controller/mcp/implementer_builder.py
+++ b/tools/controller/mcp/implementer_builder.py
@@ -54,6 +54,15 @@ def implementer_start(
             "pr_number": pr_number, "tier": tier}
 
     def body() -> dict[str, Any]:
+        # OpenCode reuses local MCP subprocesses across sessions
+        # (each session == one controller attempt). If this is a fresh
+        # attempt landing on the same long-lived MCP process, wipe the
+        # prior attempt's state. Intra-session double-_start (same
+        # attempt_id, started but not finalized) is still rejected
+        # below as a real bug.
+        prior_attempt = _STATE.identity.get("attempt_id")
+        if prior_attempt is not None and prior_attempt != attempt_id:
+            _STATE.reset_for_new_attempt()
         if _STATE.started:
             raise BuilderError("implementer_start may only be called once per attempt")
         if tier not in _TIER_VALUES:
diff --git a/tools/controller/mcp/reviewer_builder.py b/tools/controller/mcp/reviewer_builder.py
index c7fe377b1..5853b1d3f 100644
--- a/tools/controller/mcp/reviewer_builder.py
+++ b/tools/controller/mcp/reviewer_builder.py
@@ -123,6 +123,12 @@ def reviewer_start(
     }
 
     def body() -> dict[str, Any]:
+        # Cross-session reset for OpenCode-reused MCP subprocesses.
+        prior_attempt = _STATE.identity.get("attempt_id")
+        if prior_attempt is not None and prior_attempt != attempt_id:
+            _STATE.reset_for_new_attempt()
+            _GATES_AUTO_ACK.clear()
+            _GATES_DISCUSSED.clear()
         if _STATE.started:
             raise BuilderError("reviewer_start may only be called once per attempt")
         _STATE.started = True
diff --git a/tools/controller/mcp/summarizer_builder.py b/tools/controller/mcp/summarizer_builder.py
index c6e9e3261..e20330038 100644
--- a/tools/controller/mcp/summarizer_builder.py
+++ b/tools/controller/mcp/summarizer_builder.py
@@ -35,6 +35,35 @@ server = FastMCP("summarizer-builder")
 _STATE = BuilderState()
 
 
+@server.tool()
+def summarizer_start(
+    workflow_id: int, attempt_id: int,
+) -> dict[str, Any]:
+    """Initialize the summarizer session. RECOMMENDED first call.
+
+    Setters auto-start without this for backward compat, but calling
+    this explicitly is the contract the prompt advertises + lets the
+    builder detect cross-session reuse."""
+    args = {"workflow_id": workflow_id, "attempt_id": attempt_id}
+
+    def body() -> dict[str, Any]:
+        prior_attempt = _STATE.identity.get("attempt_id")
+        if prior_attempt is not None and prior_attempt != attempt_id:
+            _STATE.reset_for_new_attempt()
+        if _STATE.started:
+            raise BuilderError(
+                "summarizer_start may only be called once per attempt"
+            )
+        _STATE.started = True
+        _STATE.started_at = datetime.now(timezone.utc)
+        _STATE.identity = {
+            "workflow_id": workflow_id, "attempt_id": attempt_id,
+        }
+        return ok()
+
+    return call_with_invariants(_STATE, "summarizer_start", body, args)
+
+
 @server.tool()
 def summarizer_set_summary(text: str) -> dict[str, Any]:
     """Set the synthesis text. 50-2000 chars (mirrors SummarizerOutputV1)."""
diff --git a/tools/controller/worker/__main__.py b/tools/controller/worker/__main__.py
index 06d2ce204..3a2640a60 100644
--- a/tools/controller/worker/__main__.py
+++ b/tools/controller/worker/__main__.py
@@ -114,19 +114,31 @@ def main(argv: list[str] | None = None) -> int:
 
     # Construct the Forgejo clone URL from env + owner/repo. The
     # agent will operate inside ``{workspace_dir}/worktree/``.
+    #
+    # PD4: ``.rstrip("/api/v1")`` is a CHARACTER-SET strip in Python
+    # (removes any of /, a, p, i, v, 1 from the end repeatedly), NOT a
+    # suffix strip — works by coincidence for hosts ending in
+    # ``.com`` / ``.io`` but corrupts URLs whose final char happens to
+    # be in the set. Use an explicit endswith() check instead.
+    def _strip_api_suffix(s: str) -> str:
+        s = s.rstrip("/")
+        if s.endswith("/api/v1"):
+            s = s[: -len("/api/v1")]
+        return s
+
     forgejo_base = os.environ.get(
         "FORGEJO_URL",
-        os.environ.get("FORGEJO_API_BASE", "").rstrip("/api/v1"),
+        _strip_api_suffix(os.environ.get("FORGEJO_API_BASE", "")),
     ).rstrip("/")
     forgejo_token = os.environ.get("FORGEJO_TOKEN", "")
 
     def _build_clone_url(owner: str, repo: str) -> str:
-        # Embed the token so the agent's git ops don't need a separate
-        # credential helper. Safe inside /tmp; not logged.
-        if forgejo_token:
-            scheme, _, rest = forgejo_base.partition("://")
-            scheme = scheme or "https"
-            return f"{scheme}://{forgejo_token}@{rest}/{owner}/{repo}.git"
+        # CA10: do NOT embed the token in the URL — git persists it
+        # into ``.git/config`` under [remote "origin"] where any agent
+        # with filesystem read can exfiltrate it via ``cat .git/config``
+        # / ``git remote -v``. The token is provided per-operation via
+        # ``credential.helper`` configured by PerPRWorkspace after
+        # clone (workspace.py).
         return f"{forgejo_base}/{owner}/{repo}.git"
 
     def agent_runner(**kw):
@@ -212,19 +224,17 @@ def main(argv: list[str] | None = None) -> int:
     signal.signal(signal.SIGTERM, _on_signal)
     signal.signal(signal.SIGINT, _on_signal)
 
+    # CA9: same bug round-4 P5 fixed in master/__main__.py — manual
+    # rebuild silently drops any future-added WorkerConfig field. Use
+    # dataclasses.replace so a new field added later doesn't revert
+    # to its default whenever --max-concurrent or --poll-interval is
+    # used.
+    import dataclasses as _dc
     cfg = WorkerConfig(roles=roles)
     if args.max_concurrent is not None:
-        cfg = WorkerConfig(
-            roles=cfg.roles, max_concurrent=args.max_concurrent,
-            poll_interval_s=cfg.poll_interval_s,
-            heartbeat_interval_s=cfg.heartbeat_interval_s,
-        )
+        cfg = _dc.replace(cfg, max_concurrent=args.max_concurrent)
     if args.poll_interval is not None:
-        cfg = WorkerConfig(
-            roles=cfg.roles, max_concurrent=cfg.max_concurrent,
-            poll_interval_s=args.poll_interval,
-            heartbeat_interval_s=cfg.heartbeat_interval_s,
-        )
+        cfg = _dc.replace(cfg, poll_interval_s=args.poll_interval)
 
     logger.info(
         "worker starting: roles=%s max_concurrent=%d poll=%.1fs opencode=%s db=%s",
diff --git a/tools/controller/worker/agent_runner.py b/tools/controller/worker/agent_runner.py
index a79467ff2..72991d62c 100644
--- a/tools/controller/worker/agent_runner.py
+++ b/tools/controller/worker/agent_runner.py
@@ -182,6 +182,28 @@ def production_agent_runner(
             )
             sidecar_path = None
 
+    # CA1 stale-file cleanup: the per-PR workspace dir is reused across
+    # attempts on the same PR, so a prior attempt's
+    # ``{workspace_dir}/{role}_output.json`` would be picked up by the
+    # poller as if it were this attempt's output. Wipe BOTH the MCP
+    # canonical path (defensive) and the fallback paths before the
+    # session starts.
+    fallback_paths: list[str] = []
+    if workspace_dir is not None:
+        fallback_paths.append(
+            str(Path(workspace_dir) / f"{role}_output.json")
+        )
+    for stale in [out_path] + fallback_paths:
+        try:
+            os.unlink(stale)
+        except FileNotFoundError:
+            pass
+        except OSError as exc:
+            logger.warning(
+                "could not unlink stale output file %s before session: %s",
+                stale, exc,
+            )
+
     try:
         # Hand the OpenCode session the MCP we just spawned + the
         # input payload. The session's job: drive the LLM through
@@ -200,6 +222,40 @@ def production_agent_runner(
         _session_start = _time.monotonic()
 
         def _capture_inline_json(parsed: dict) -> None:
+            # CA7: skip if we lost the lock — file written after lock-loss
+            # would poison a subsequent attempt on this same workspace.
+            if lost_lock_check():
+                logger.info(
+                    "inline-output capture skipped (lost lock) "
+                    "for attempt_id=%s role=%s", attempt_id, role,
+                )
+                return
+            # PD3: if the MCP already wrote canonical V1 to out_path,
+            # do NOT overwrite. The MCP path is authoritative; the
+            # inline-chat path is a fallback for agents that don't call
+            # the MCP. Inspect any existing file at out_path first.
+            for p in [out_path] + list(fallback_paths):
+                try:
+                    with open(p, "r", encoding="utf-8") as f:
+                        existing = f.read().strip()
+                except (FileNotFoundError, OSError):
+                    continue
+                if not existing:
+                    continue
+                try:
+                    existing_obj = _json_mod.loads(existing)
+                except (TypeError, ValueError):
+                    continue
+                if (
+                    isinstance(existing_obj, dict)
+                    and existing_obj.get("output_version") == "V1"
+                ):
+                    logger.debug(
+                        "MCP already emitted V1 to %s; skipping inline "
+                        "capture (would overwrite authoritative output)",
+                        p,
+                    )
+                    return
             # Adapt the legacy-shape JSON to the V1 contract before
             # writing. Without this, strict_parse against the V1
             # model rejects the legacy shape (missing output_version,
@@ -208,18 +264,26 @@ def production_agent_runner(
             adapted = adapt_to_v1(
                 role, parsed, tier=tier, wallclock_seconds=wallclock,
             )
+            # PD10: atomic write via .tmp + os.replace so the poller
+            # never sees a half-written file.
+            tmp_path = f"{out_path}.tmp"
             try:
-                with open(out_path, "w", encoding="utf-8") as f:
+                with open(tmp_path, "w", encoding="utf-8") as f:
                     f.write(_json_mod.dumps(adapted))
                     f.write("\n")
                     f.flush()
                     os.fsync(f.fileno())
+                os.replace(tmp_path, out_path)
             except OSError as exc:
                 logger.warning(
                     "inline-output capture write failed (%s): canonical "
                     "poller will fall through to other channels",
                     exc,
                 )
+                try:
+                    os.unlink(tmp_path)
+                except OSError:
+                    pass
 
         try:
             run_opencode_session(
@@ -240,17 +304,6 @@ def production_agent_runner(
 
         if lost_lock_check():
             raise WorkerLostLock("lost lock during MCP session")
-
-        # Wait for the MCP to emit its JSON to the canonical output
-        # file. Also poll the per-role fallback path under workspace_dir
-        # — used during the trial phase where the response-builder
-        # MCPs aren't wired into OpenCode and the agent writes its
-        # JSON via direct bash/edit.
-        fallback_paths: list[str] = []
-        if workspace_dir is not None:
-            fallback_paths.append(
-                str(Path(workspace_dir) / f"{role}_output.json")
-            )
         canonical = _wait_for_canonical_output(
             spawn, out_path, timeout_s=finalize_timeout_s,
             fallback_paths=fallback_paths,
diff --git a/tools/controller/worker/legacy_adapter.py b/tools/controller/worker/legacy_adapter.py
index 63c0a32f4..0cc60d0b5 100644
--- a/tools/controller/worker/legacy_adapter.py
+++ b/tools/controller/worker/legacy_adapter.py
@@ -38,19 +38,108 @@ _LEGACY_BOOL_TO_CONFIDENCE: dict[Any, str] = {
     False: "low",
 }
 
+# Cap on adapter-tolerated blocker length so a buggy agent can't blow
+# up the audit log / DB column with a 10MB blocker string. The MCP
+# path has no equivalent cap (PD15), but the legacy chat-JSON path is
+# unbounded by construction.
+_BLOCKER_CHAR_CAP = 4096
+_BLOCKING_ISSUE_DESCRIPTION_CAP = 4096
+
+import logging as _logging
+_logger = _logging.getLogger(__name__)
+
 
 def _confidence_from_legacy(payload: dict[str, Any]) -> str:
-    """Coerce a legacy is_confident bool to V1 confidence string."""
-    if "confidence" in payload and isinstance(payload["confidence"], str):
-        v = payload["confidence"].lower()
-        if v in {"high", "medium", "low"}:
-            return v
+    """Coerce a legacy is_confident bool to V1 confidence string.
+
+    PD1: if ``confidence`` is present but not a recognized string,
+    emit a WARNING so operators see prompt drift rather than silently
+    masking it as "medium"."""
+    if "confidence" in payload:
+        raw = payload["confidence"]
+        if isinstance(raw, str):
+            v = raw.lower().strip()
+            if v in {"high", "medium", "low"}:
+                return v
+            _logger.warning(
+                "adapter: unrecognized confidence value %r; falling back "
+                "to is_confident / 'medium' default", raw,
+            )
+        else:
+            _logger.warning(
+                "adapter: non-string confidence %r (type=%s); falling back "
+                "to is_confident / 'medium' default", raw, type(raw).__name__,
+            )
     is_conf = payload.get("is_confident")
     if isinstance(is_conf, bool):
         return _LEGACY_BOOL_TO_CONFIDENCE[is_conf]
     return "medium"
 
 
+def _clamp_tier(raw: Any, *, fallback: int = 0) -> int:
+    """Coerce ``raw`` to an int in {0, 1, 2}. PD2: estimator's
+    ``recommended_tier`` is ``Literal[0, 1, 2]`` — any unclamped int
+    fails Pydantic validation, defeating the adapter's purpose."""
+    try:
+        v = int(raw) if raw is not None else fallback
+    except (TypeError, ValueError):
+        _logger.warning("adapter: non-int tier %r; using fallback %d", raw, fallback)
+        return fallback
+    if v < 0:
+        _logger.warning("adapter: tier %d < 0; clamping to 0", v)
+        return 0
+    if v > 2:
+        _logger.warning("adapter: tier %d > 2; clamping to 2", v)
+        return 2
+    return v
+
+
+def _coerce_blocking_issues(raw: Any) -> list[dict[str, Any]]:
+    """PD7: V1's ``blocking_issues`` is ``list[BlockingIssue]`` (dicts
+    with required ``description`` field). Legacy agents often emit a
+    list of free-form strings. Wrap bare strings into the dict shape
+    with a sensible default severity so strict_parse accepts them."""
+    if not isinstance(raw, list):
+        return []
+    out: list[dict[str, Any]] = []
+    for item in raw:
+        if isinstance(item, dict):
+            # Already-shaped; pass through (let pydantic enforce schema).
+            out.append(item)
+        elif isinstance(item, str):
+            out.append({
+                "description": item[:_BLOCKING_ISSUE_DESCRIPTION_CAP],
+                "severity": "error",
+            })
+        else:
+            _logger.warning(
+                "adapter: blocking_issues entry has unexpected type %s; "
+                "dropping", type(item).__name__,
+            )
+    return out
+
+
+def _coerce_commits(raw: Any, *, singular_fallback: Any = None) -> list[str]:
+    """Normalize a commit-shas list. PD22: also accept a singular
+    ``commit_sha`` fallback for legacy roles that only emit one."""
+    if isinstance(raw, list):
+        out: list[str] = []
+        dropped = 0
+        for c in raw:
+            if isinstance(c, str) and c.strip():
+                out.append(c)
+            else:
+                dropped += 1
+        if dropped:
+            _logger.warning(
+                "adapter: dropped %d non-string commit entries", dropped,
+            )
+        return out
+    if isinstance(singular_fallback, str) and singular_fallback.strip():
+        return [singular_fallback]
+    return []
+
+
 def adapt_to_v1(
     role: str, payload: dict[str, Any], *,
     tier: int | None, wallclock_seconds: float,
@@ -67,13 +156,27 @@ def adapt_to_v1(
         # Nothing to adapt; let strict_parse raise its own error.
         return payload  # type: ignore[return-value]
     if payload.get("output_version") == "V1":
-        # Already V1-shape; pass through.
+        # Already V1-shape — passthrough. PD11: do NOT overwrite the
+        # agent's claimed wallclock; if the agent emitted V1 it should
+        # have set wallclock_seconds itself. The controller has its
+        # own measured value but trusting V1's contract is the whole
+        # point of having a V1 marker.
         return payload
 
+    # CA4: telemetry — log every fallback so operators can measure
+    # agent-migration progress (how often is the adapter still needed?).
+    # Cannot stamp the payload (V1 has extra="forbid"); the log line
+    # carries role + payload-shape fingerprint instead.
+    _logger.info(
+        "legacy_adapter: adapting role=%s (keys=%s) — chat-JSON fallback "
+        "active; agent should migrate to MCP",
+        role, sorted(payload.keys())[:10],
+    )
+
     if role == "estimator":
         return {
             "output_version": "V1",
-            "recommended_tier": int(payload.get("recommended_tier", 0)),
+            "recommended_tier": _clamp_tier(payload.get("recommended_tier")),
             "is_metadata_only": bool(payload.get("is_metadata_only", False)),
             "confidence": _confidence_from_legacy(payload),
             "reasoning": str(payload.get("reasoning", ""))[:2048],
@@ -88,28 +191,30 @@ def adapt_to_v1(
         files = payload.get("files_touched") or []
         if not isinstance(files, list):
             files = []
-        commits = payload.get("commit_shas")
-        if not isinstance(commits, list):
-            # legacy commonly has commit_sha (singular) or no commit info
-            single = payload.get("commit_sha")
-            commits = [single] if isinstance(single, str) and single else []
-        blockers = payload.get("blockers") or []
-        if v1_outcome == "blocked" and not blockers:
+        commits = _coerce_commits(
+            payload.get("commit_shas"),
+            singular_fallback=payload.get("commit_sha"),
+        )
+        blockers_raw = payload.get("blockers") or []
+        if v1_outcome == "blocked" and not blockers_raw:
             # V1 requires ≥1 blocker for outcome=blocked; synthesize one
             # from any free-form context the legacy agent emitted.
-            blockers = [
+            blockers_raw = [
                 str(payload.get("reason"))
                 if payload.get("reason")
                 else "agent emitted unresolved without explicit blocker"
             ]
+        # PD15: cap each blocker length so a buggy agent's 10MB blocker
+        # doesn't blow the audit log / DB column.
+        blockers = [str(b)[:_BLOCKER_CHAR_CAP] for b in blockers_raw]
         return {
             "output_version": "V1",
             "outcome": v1_outcome,
             "files_touched": [str(f) for f in files if isinstance(f, str)],
-            "commit_shas": [str(c) for c in commits if isinstance(c, str)],
+            "commit_shas": commits,
             "confidence": _confidence_from_legacy(payload),
-            "blockers": [str(b) for b in blockers],
-            "used_tier": int(tier or 0),
+            "blockers": blockers,
+            "used_tier": _clamp_tier(tier),
             "wallclock_seconds": float(wallclock_seconds),
         }
 
@@ -134,7 +239,11 @@ def adapt_to_v1(
         return {
             "output_version": "V1",
             "verdict": verdict,
-            "blocking_issues": payload.get("blocking_issues") or [],
+            # PD7: coerce list-of-strings into list-of-BlockingIssue
+            # dicts so strict_parse accepts the legacy shape.
+            "blocking_issues": _coerce_blocking_issues(
+                payload.get("blocking_issues"),
+            ),
             "approved_at_sha": approved_at,
             "suggested_next_action": next_action,
             "confidence": _confidence_from_legacy(payload),
@@ -146,7 +255,12 @@ def adapt_to_v1(
         valid = {"resolved", "partial", "irreconcilable", "competence-failure"}
         outcome = legacy_outcome if legacy_outcome in valid else "competence-failure"
         files = payload.get("files_modified") or payload.get("files_touched") or []
-        commits = payload.get("commit_shas") or []
+        # PD22: also accept ``commit_sha`` singular fallback as the
+        # implementer adapter does.
+        commits = _coerce_commits(
+            payload.get("commit_shas"),
+            singular_fallback=payload.get("commit_sha"),
+        )
         new_head = payload.get("new_head_sha")
         if outcome != "resolved":
             new_head = None
@@ -154,7 +268,7 @@ def adapt_to_v1(
             "output_version": "V1",
             "outcome": outcome,
             "files_modified": [str(f) for f in files if isinstance(f, str)],
-            "commit_shas": [str(c) for c in commits if isinstance(c, str)],
+            "commit_shas": commits,
             "new_head_sha": new_head,
             "reasoning": str(payload.get("reasoning", ""))[:2048],
             "confidence": _confidence_from_legacy(payload),
@@ -162,9 +276,20 @@ def adapt_to_v1(
         }
 
     if role == "summarizer":
+        # PD14: SummarizerOutputV1.summary requires min_length=50. A
+        # too-short legacy summary would silently slip through and then
+        # fail strict_parse — surface the problem with a padded marker
+        # so the agent's behavior is visible in the parsed output.
+        raw_summary = str(payload.get("summary", ""))
+        if len(raw_summary) < 50:
+            raw_summary = (
+                raw_summary
+                + " [adapter-padded: legacy summary below 50-char floor]"
+            )
+        summary = raw_summary[:2000]
         return {
             "output_version": "V1",
-            "summary": str(payload.get("summary", ""))[:2000],
+            "summary": summary,
             "covers_through_attempt": int(
                 payload.get("covers_through_attempt", 1),
             ),
diff --git a/tools/controller/worker/opencode_session.py b/tools/controller/worker/opencode_session.py
index 8ce2598e1..d2b80850d 100644
--- a/tools/controller/worker/opencode_session.py
+++ b/tools/controller/worker/opencode_session.py
@@ -151,23 +151,31 @@ def wire_opencode_session(
             ) from exc
 
         # Phase 1k++++ trial-path: harvest the inline JSON the agent
-        # emitted as its final response. The existing legacy pipeline's
-        # agent prompts (.opencode/agents/*.md) all instruct the agent
-        # to emit a single JSON object as the LAST machine-readable
-        # artifact — ``_opencode_worker.run_session_blocking`` extracts
-        # it into ``SessionResult.parsed_json`` via
-        # ``_extract_last_json_object``. The controller's MCP-builder
-        # path isn't reachable from OpenCode (MCPs not registered in
-        # opencode.json); this inline channel IS the working path.
+        # emitted as its final response. Legacy pipeline agents emit a
+        # single JSON object as their final message; OpenCode worker
+        # extracts it into ``SessionResult.parsed_json``. Since
+        # Phase 1m the response-builder MCPs ARE registered in
+        # opencode.json, so this inline channel is a fallback for
+        # agents that still emit chat-JSON. The agent_runner's
+        # callback skips the write if the MCP already wrote
+        # canonical V1 to the same path.
+        #
+        # PD16: re-raise WorkerLostLock from the callback so the
+        # runner's outer handler aborts properly. Other exceptions are
+        # converted to WorkerError so they classify as
+        # ``worker-internal-error`` instead of silently letting the
+        # canonical poller time out 30s later with no root cause.
         parsed = getattr(result, "parsed_json", None)
         if parsed is not None and inline_output_callback is not None:
             try:
                 inline_output_callback(parsed)
-            except Exception:
-                logger.exception(
-                    "inline_output_callback raised for attempt_id=%s",
-                    attempt_id,
-                )
+            except WorkerLostLock:
+                raise
+            except Exception as exc:
+                raise WorkerError(
+                    f"inline_output_callback raised: {exc}",
+                    outcome="worker-internal-error",
+                ) from exc
 
         # Inspect SessionResult.status.
         status = getattr(result, "status", None)
diff --git a/tools/controller/worker/prompts.py b/tools/controller/worker/prompts.py
index 43e8c7922..c209dbceb 100644
--- a/tools/controller/worker/prompts.py
+++ b/tools/controller/worker/prompts.py
@@ -516,6 +516,9 @@ def build_summarizer_prompt(input_payload: dict) -> str:
         "`summarizer_set_covers_through_attempt(value=N)`",
         f"3. `summarizer_finalize(output_path=\"{_fmt_or_unavailable(s.get('workspace_dir'))}/summarizer_output.json\")`",
         "",
+        "DO NOT emit a JSON object in your final chat message — the "
+        "new controller reads from the MCP-written file.",
+        "",
         "Be terse: capture what was tried, what failed, what's worth "
         "carrying forward. Skip redundant headers, file lists already "
         "in the next attempt's input.",
@@ -542,8 +545,17 @@ def build_prompt(role: str, tier: int | None, input_payload: dict) -> str:
     if role == "estimator":
         return build_estimator_prompt(input_payload)
     if role == "conflict_resolver":
+        # PD13: scheduler always sets tier for conflict_resolver
+        # (CONFLICT_RESOLVING uses workflow.current_tier), but defend
+        # against contract drift by defaulting to tier=1 with a warning
+        # instead of raising — a single attempt failure beats burning
+        # a pickup_count + stack trace inside production_agent_runner.
         if tier is None:
-            raise ValueError("conflict_resolver prompt requires tier")
+            import logging
+            logging.getLogger(__name__).warning(
+                "conflict_resolver prompt called without tier; defaulting to 1"
+            )
+            tier = 1
         return build_conflict_resolver_prompt(input_payload, tier=tier)
     if role == "summarizer":
         return build_summarizer_prompt(input_payload)
diff --git a/tools/controller/worker/workspace.py b/tools/controller/worker/workspace.py
index f278126fd..104db03ce 100644
--- a/tools/controller/worker/workspace.py
+++ b/tools/controller/worker/workspace.py
@@ -116,6 +116,19 @@ class PerPRWorkspace:
         self.workspace_dir.mkdir(parents=True, exist_ok=True)
         self.attempts_dir.mkdir(exist_ok=True)
 
+    # Credential-helper invocation that sources the Forgejo token from
+    # the ``FORGEJO_TOKEN`` env var at fetch/push time. Keeps the token
+    # out of ``.git/config`` (where any agent with filesystem read
+    # could ``cat .git/config`` to exfiltrate). The helper script is
+    # quoted because git's credential.helper accepts shell strings —
+    # see ``git help credentials``. ``test "$1" = get`` ensures we
+    # only respond to the get action; store/erase become no-ops.
+    _CREDENTIAL_HELPER = (
+        '!f() { test "$1" = "get" && '
+        'echo "username=x" && '
+        'echo "password=${FORGEJO_TOKEN:-}"; }; f'
+    )
+
     def clone_if_absent(self) -> None:
         """``git clone`` into ``worktree/`` if not already cloned.
 
@@ -123,6 +136,12 @@ class PerPRWorkspace:
         no-op. The clone uses ``--no-single-branch`` so subsequent
         ``git fetch`` can pick up new branches.
 
+        Token-handling: the clone_url MUST NOT contain credentials.
+        A local ``credential.helper`` is configured immediately after
+        clone so subsequent fetch/push operations source the token
+        from ``$FORGEJO_TOKEN`` at runtime — keeps the token out of
+        ``.git/config``.
+
         Tests can pre-populate ``worktree/`` (e.g., via ``git init``)
         and skip this call.
         """
@@ -136,10 +155,24 @@ class PerPRWorkspace:
         # Remove any partial state before cloning.
         if self.worktree_dir.exists():
             shutil.rmtree(self.worktree_dir)
+        # Use -c credential.helper at clone time so the initial fetch
+        # can authenticate; then bake the same helper into the local
+        # repo config for subsequent git ops.
         _git_run(
-            ["git", "clone", "--no-single-branch", self.clone_url, str(self.worktree_dir)],
+            [
+                "git", "-c", f"credential.helper={self._CREDENTIAL_HELPER}",
+                "clone", "--no-single-branch",
+                self.clone_url, str(self.worktree_dir),
+            ],
             cwd=None,
         )
+        _git_run(
+            [
+                "git", "config", "--local",
+                "credential.helper", self._CREDENTIAL_HELPER,
+            ],
+            cwd=self.worktree_dir,
+        )
 
     def fetch_and_validate(
         self, *, expected_head_sha: str, head_ref: str