diff --git a/tests/auto_agents/controller/test_legacy_adapter.py b/tests/auto_agents/controller/test_legacy_adapter.py new file mode 100644 index 000000000..4b7c3073d --- /dev/null +++ b/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 diff --git a/tools/controller/worker/agent_runner.py b/tools/controller/worker/agent_runner.py index 2b7030d8a..a79467ff2 100644 --- a/tools/controller/worker/agent_runner.py +++ b/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 diff --git a/tools/controller/worker/legacy_adapter.py b/tools/controller/worker/legacy_adapter.py new file mode 100644 index 000000000..63c0a32f4 --- /dev/null +++ b/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"] diff --git a/tools/controller/worker/opencode_session.py b/tools/controller/worker/opencode_session.py index bb2cd7fc3..8ce2598e1 100644 --- a/tools/controller/worker/opencode_session.py +++ b/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":