fix(controller): trial-path output channel — harvest inline JSON + legacy-shape adapter

Trial run-1 surfaced the real failure mode: the existing agent
prompts emit their result as a single JSON object in the FINAL
assistant message (the legacy dispatch_implementer/dispatch_review
contract). They DON'T call the controller's response-builder MCPs
(not registered in opencode.json) and they DON'T write a file. The
worker's canonical-output poller timed out every attempt.

Fix (two pieces):

1. INLINE JSON HARVEST
   ``_opencode_worker.run_session_blocking`` already extracts the
   last JSON object from the agent's final response into
   ``SessionResult.parsed_json``. The controller's opencode_session
   adapter was ignoring it.
   - ``opencode_session.py:run_opencode_session`` now accepts an
     ``inline_output_callback`` kwarg. After the session completes,
     if parsed_json is set, the callback fires.
   - ``agent_runner.py`` provides the callback: writes the parsed
     JSON to the canonical-output path so the existing poller picks
     it up uniformly with the MCP + file-write channels.

2. LEGACY → V1 SHAPE ADAPTER (``worker/legacy_adapter.py``)
   The legacy JSON shape doesn't match V1 contracts:
   - estimator: legacy {recommended_tier, is_confident, reasoning}
     vs V1 {output_version, recommended_tier, is_metadata_only,
     confidence, reasoning, wallclock_seconds}
   - implementer: legacy {outcome: resolved|unresolved, ...} vs V1
     {outcome: resolved|rebase-failed|blocked|noop|competence-failure,
     commit_shas, blockers, used_tier, ...}
   - reviewer: legacy {verdict, ...} vs V1 {output_version, verdict,
     blocking_issues, suggested_next_action, ...}

   ``adapt_to_v1(role, payload, tier, wallclock_seconds)`` normalizes
   each role's legacy shape into the corresponding V1 dict:
   - adds ``output_version="V1"``
   - maps outcome enums (``unresolved`` → ``blocked``)
   - synthesizes missing required fields with sensible defaults
   - coerces singular commit_sha → list[commit_shas]
   - converts legacy is_confident bool → confidence string

   Already-V1 payloads pass through unchanged. The agent_runner
   wraps the inline_output_callback to run the adapter before
   writing to the canonical path.

Tests (+12 in test_legacy_adapter.py):
- Each role's adapt: legacy in, V1-validating-via-strict_parse out
- Already-V1 passthrough
- Outcome enum mappings + invalid-outcome fallback
- Singular commit_sha normalization
- Non-dict / unknown-role passthrough (lets strict_parse raise)

Total: 738 controller tests pass (+12 net), 0 regressions.

Long-term: update agent prompts to emit V1 directly, register the
MCPs in opencode.json. For the trial, this adapter lets the existing
agents flow through the new controller unchanged.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

tests/auto_agents/controller/test_legacy_adapter.py

@@ -0,0 +1,146 @@
+"""Tests for the legacy-shape → V1-shape adapter (trial path)."""
+from __future__ import annotations
+
+import pytest
+
+from tools.controller.contracts.v1 import (
+    EstimatorOutputV1,
+    ImplementerOutputV1,
+    ReviewerOutputV1,
+)
+from tools.controller.worker.legacy_adapter import adapt_to_v1
+
+
+class TestEstimatorAdapt:
+    def test_legacy_estimator_v1_passes_strict_parse(self):
+        legacy = {
+            "recommended_tier": 1,
+            "is_confident": True,
+            "reasoning": "tier 1 because cross-file refactor",
+        }
+        v1 = adapt_to_v1(
+            "estimator", legacy, tier=None, wallclock_seconds=12.5,
+        )
+        parsed = EstimatorOutputV1.model_validate(v1)
+        assert parsed.recommended_tier == 1
+        assert parsed.confidence == "high"
+        assert parsed.is_metadata_only is False
+        assert parsed.wallclock_seconds == 12.5
+
+    def test_already_v1_passthrough(self):
+        v1 = {
+            "output_version": "V1", "recommended_tier": 0,
+            "is_metadata_only": False, "confidence": "medium",
+            "reasoning": "ok", "wallclock_seconds": 1.0,
+        }
+        out = adapt_to_v1("estimator", v1, tier=None, wallclock_seconds=1.0)
+        assert out is v1  # no rebuild needed
+
+
+class TestImplementerAdapt:
+    def test_legacy_resolved_with_commits(self):
+        legacy = {
+            "outcome": "resolved",
+            "files_touched": ["a.py", "b.py"],
+            "commit_shas": ["abc1234"],
+        }
+        v1 = adapt_to_v1(
+            "implementer", legacy, tier=0, wallclock_seconds=42.0,
+        )
+        parsed = ImplementerOutputV1.model_validate(v1)
+        assert parsed.outcome == "resolved"
+        assert parsed.files_touched == ["a.py", "b.py"]
+        assert parsed.commit_shas == ["abc1234"]
+        assert parsed.used_tier == 0
+
+    def test_legacy_unresolved_maps_to_blocked(self):
+        legacy = {
+            "outcome": "unresolved",
+            "files_touched": [],
+            "reason": "tool denied",
+        }
+        v1 = adapt_to_v1(
+            "implementer", legacy, tier=1, wallclock_seconds=10.0,
+        )
+        parsed = ImplementerOutputV1.model_validate(v1)
+        assert parsed.outcome == "blocked"
+        # blockers synthesized from reason field
+        assert parsed.blockers == ["tool denied"]
+        assert parsed.used_tier == 1
+
+    def test_legacy_singular_commit_sha_normalized(self):
+        legacy = {
+            "outcome": "resolved",
+            "files_touched": ["a.py"],
+            "commit_sha": "deadbeef",  # singular, not plural
+        }
+        v1 = adapt_to_v1(
+            "implementer", legacy, tier=0, wallclock_seconds=1.0,
+        )
+        parsed = ImplementerOutputV1.model_validate(v1)
+        assert parsed.commit_shas == ["deadbeef"]
+
+    def test_legacy_no_commit_info_means_empty_list(self):
+        legacy = {"outcome": "noop", "files_touched": []}
+        v1 = adapt_to_v1(
+            "implementer", legacy, tier=0, wallclock_seconds=1.0,
+        )
+        parsed = ImplementerOutputV1.model_validate(v1)
+        assert parsed.commit_shas == []
+        assert parsed.outcome == "noop"
+
+    def test_unknown_outcome_falls_to_competence_failure(self):
+        legacy = {"outcome": "weird", "files_touched": []}
+        v1 = adapt_to_v1(
+            "implementer", legacy, tier=0, wallclock_seconds=1.0,
+        )
+        parsed = ImplementerOutputV1.model_validate(v1)
+        assert parsed.outcome == "competence-failure"
+
+
+class TestReviewerAdapt:
+    def test_legacy_approve(self):
+        legacy = {
+            "verdict": "approve",
+            "approved_at_sha": "abc",
+            "is_confident": True,
+        }
+        v1 = adapt_to_v1(
+            "reviewer", legacy, tier=None, wallclock_seconds=5.0,
+        )
+        parsed = ReviewerOutputV1.model_validate(v1)
+        assert parsed.verdict == "approve"
+        assert parsed.approved_at_sha == "abc"
+        assert parsed.suggested_next_action == "merge"
+
+    def test_legacy_request_changes(self):
+        legacy = {"verdict": "request-changes", "is_confident": False}
+        v1 = adapt_to_v1(
+            "reviewer", legacy, tier=None, wallclock_seconds=5.0,
+        )
+        parsed = ReviewerOutputV1.model_validate(v1)
+        assert parsed.verdict == "request-changes"
+        assert parsed.confidence == "low"
+
+    def test_invalid_verdict_falls_to_abstain(self):
+        legacy = {"verdict": "yay"}
+        v1 = adapt_to_v1(
+            "reviewer", legacy, tier=None, wallclock_seconds=5.0,
+        )
+        parsed = ReviewerOutputV1.model_validate(v1)
+        assert parsed.verdict == "abstain"
+
+
+class TestNonDictPassthrough:
+    def test_string_passthrough(self):
+        result = adapt_to_v1(
+            "estimator", "not-a-dict", tier=None, wallclock_seconds=1.0,
+        )
+        assert result == "not-a-dict"
+
+    def test_unknown_role_passthrough(self):
+        payload = {"outcome": "x"}
+        result = adapt_to_v1(
+            "weird_role", payload, tier=None, wallclock_seconds=1.0,
+        )
+        assert result is payload

tools/controller/worker/agent_runner.py

@@ -42,6 +42,7 @@ from ..contracts.v1 import (
     ReviewerOutputV1,
     SummarizerOutputV1,
 )
+from .legacy_adapter import adapt_to_v1
 from .runner import WorkerError, WorkerLostLock
 from .session_sidecar import (
     WorkerSession,
@@ -186,12 +187,47 @@ def production_agent_runner(
         # input payload. The session's job: drive the LLM through
         # its tool-calling sequence; the MCP captures the assembled
         # state + emits canonical JSON via its finalize() tool.
+        #
+        # Trial-path: legacy-style agents emit a single JSON object
+        # as their final response message; the OpenCode worker
+        # extracts it into ``SessionResult.parsed_json``. We hand
+        # the opencode_session adapter a callback that writes that
+        # JSON to the canonical-output file so the post-session poller
+        # picks it up uniformly with the MCP + file-write channels.
+        import json as _json_mod
+
+        import time as _time
+        _session_start = _time.monotonic()
+
+        def _capture_inline_json(parsed: dict) -> None:
+            # 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,
+            # different outcome enum, etc.).
+            wallclock = _time.monotonic() - _session_start
+            adapted = adapt_to_v1(
+                role, parsed, tier=tier, wallclock_seconds=wallclock,
+            )
+            try:
+                with open(out_path, "w", encoding="utf-8") as f:
+                    f.write(_json_mod.dumps(adapted))
+                    f.write("\n")
+                    f.flush()
+                    os.fsync(f.fileno())
+            except OSError as exc:
+                logger.warning(
+                    "inline-output capture write failed (%s): canonical "
+                    "poller will fall through to other channels",
+                    exc,
+                )
+
         try:
             run_opencode_session(
                 role=role, tier=tier, input_payload=input_payload,
                 mcp_process=spawn.process,
                 attempt_id=attempt_id, instance_id=instance_id,
                 lost_lock_check=lost_lock_check,
+                inline_output_callback=_capture_inline_json,
             )
         except WorkerLostLock:
             raise

tools/controller/worker/legacy_adapter.py

@@ -0,0 +1,178 @@
+"""Legacy-shape → V1-shape adapter for the controller's worker.
+
+The existing pipeline's agents (``.opencode/agents/*.md``) emit a
+small JSON object as their final response. The shape pre-dates the
+controller's V1 contracts:
+
+- estimator: ``{recommended_tier, is_confident, reasoning}``
+- implementer: ``{outcome: resolved|unresolved, files_touched, ...}``
+- reviewer: ``{verdict, ...}``
+
+The controller's V1 contracts require additional fields
+(``output_version``, ``confidence``, ``used_tier``, ``wallclock_seconds``,
+``blockers``, etc.). Strict-parse against the V1 model fails on the
+legacy shape.
+
+For the Phase 2 trial we adapt the legacy shape to V1 here so we can
+exercise the controller's pipeline against the existing agents
+WITHOUT modifying every agent prompt. Long-term, the agent prompts
+should be updated to emit V1 directly.
+"""
+from __future__ import annotations
+
+from typing import Any
+
+
+# Map legacy outcome strings to V1 outcome enum values.
+_IMPL_OUTCOME_LEGACY_TO_V1: dict[str, str] = {
+    "resolved": "resolved",
+    "unresolved": "blocked",
+    "blocked": "blocked",
+    "noop": "noop",
+    "rebase-failed": "rebase-failed",
+    "competence-failure": "competence-failure",
+}
+
+_LEGACY_BOOL_TO_CONFIDENCE: dict[Any, str] = {
+    True: "high",
+    False: "low",
+}
+
+
+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
+    is_conf = payload.get("is_confident")
+    if isinstance(is_conf, bool):
+        return _LEGACY_BOOL_TO_CONFIDENCE[is_conf]
+    return "medium"
+
+
+def adapt_to_v1(
+    role: str, payload: dict[str, Any], *,
+    tier: int | None, wallclock_seconds: float,
+) -> dict[str, Any]:
+    """Return a V1-shape dict by filling in missing fields from
+    sensible defaults.
+
+    ``role`` selects which V1 contract to adapt to (implementer,
+    reviewer, estimator, conflict_resolver, summarizer).
+    ``tier`` + ``wallclock_seconds`` are runtime values the agent
+    doesn't carry — the runner provides them.
+    """
+    if not isinstance(payload, dict):
+        # 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.
+        return payload
+
+    if role == "estimator":
+        return {
+            "output_version": "V1",
+            "recommended_tier": int(payload.get("recommended_tier", 0)),
+            "is_metadata_only": bool(payload.get("is_metadata_only", False)),
+            "confidence": _confidence_from_legacy(payload),
+            "reasoning": str(payload.get("reasoning", ""))[:2048],
+            "wallclock_seconds": float(wallclock_seconds),
+        }
+
+    if role == "implementer":
+        legacy_outcome = payload.get("outcome", "competence-failure")
+        v1_outcome = _IMPL_OUTCOME_LEGACY_TO_V1.get(
+            legacy_outcome, "competence-failure",
+        )
+        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:
+            # V1 requires ≥1 blocker for outcome=blocked; synthesize one
+            # from any free-form context the legacy agent emitted.
+            blockers = [
+                str(payload.get("reason"))
+                if payload.get("reason")
+                else "agent emitted unresolved without explicit blocker"
+            ]
+        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)],
+            "confidence": _confidence_from_legacy(payload),
+            "blockers": [str(b) for b in blockers],
+            "used_tier": int(tier or 0),
+            "wallclock_seconds": float(wallclock_seconds),
+        }
+
+    if role == "reviewer":
+        legacy_verdict = payload.get("verdict", "abstain")
+        # V1 allows: approve / request-changes / comment / abstain
+        valid = {"approve", "request-changes", "comment", "abstain"}
+        verdict = legacy_verdict if legacy_verdict in valid else "abstain"
+        next_action = payload.get(
+            "suggested_next_action",
+            "merge" if verdict == "approve" else "human-attention",
+        )
+        valid_actions = {
+            "merge", "wait-for-ci", "re-implement",
+            "human-attention", "abandon",
+        }
+        if next_action not in valid_actions:
+            next_action = "human-attention"
+        approved_at = payload.get("approved_at_sha")
+        if verdict != "approve":
+            approved_at = None
+        return {
+            "output_version": "V1",
+            "verdict": verdict,
+            "blocking_issues": payload.get("blocking_issues") or [],
+            "approved_at_sha": approved_at,
+            "suggested_next_action": next_action,
+            "confidence": _confidence_from_legacy(payload),
+            "wallclock_seconds": float(wallclock_seconds),
+        }
+
+    if role == "conflict_resolver":
+        legacy_outcome = payload.get("outcome", "competence-failure")
+        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 []
+        new_head = payload.get("new_head_sha")
+        if outcome != "resolved":
+            new_head = None
+        return {
+            "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)],
+            "new_head_sha": new_head,
+            "reasoning": str(payload.get("reasoning", ""))[:2048],
+            "confidence": _confidence_from_legacy(payload),
+            "wallclock_seconds": float(wallclock_seconds),
+        }
+
+    if role == "summarizer":
+        return {
+            "output_version": "V1",
+            "summary": str(payload.get("summary", ""))[:2000],
+            "covers_through_attempt": int(
+                payload.get("covers_through_attempt", 1),
+            ),
+            "wallclock_seconds": float(wallclock_seconds),
+        }
+
+    # Unknown role — pass through (strict_parse will raise).
+    return payload
+
+
+__all__ = ["adapt_to_v1"]

tools/controller/worker/opencode_session.py

@@ -114,6 +114,7 @@ def wire_opencode_session(
         *, role: str, tier: int | None, input_payload: dict[str, Any],
         mcp_process, attempt_id: int, instance_id: str,
         lost_lock_check: Callable[[], bool],
+        inline_output_callback: Callable[[dict[str, Any]], None] | None = None,
     ) -> None:
         """The injected callable. Drives the OpenCode session +
         propagates lost_lock_check via the on_poll callback."""
@@ -149,6 +150,25 @@ def wire_opencode_session(
                 outcome="worker-internal-error",
             ) 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.
+        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,
+                )
+
         # Inspect SessionResult.status.
         status = getattr(result, "status", None)
         if status == "completed":