fix(controller): batch M — post-Phase-1m adversarial review fixes
Three rounds of adversarial review (Chief Architect / Principal Dev /
Senior Test Engineer) on commits 3ca794be7..db12f45ac surfaced ~35
issues. This commit addresses 25+ across criticals, highs, and
mediums, and adds 40 new tests covering the changes plus key gaps the
review identified.
CRITICALS (M1):
- CA1: stale {role}_output.json from a prior attempt on the same
per-PR workspace was readable as "fresh" output of the new attempt.
agent_runner now unlinks the MCP-canonical path AND every fallback
path BEFORE the session runs.
- CA2/PD5: opencode.json-registered MCP subprocesses persist across
OpenCode sessions, but BuilderState was module-singleton. Added
reset_for_new_attempt() + cross-session detection (compare
identity.attempt_id) to every *_start; force-resets with WARN if
prior attempt was interrupted (timeout / lost lock).
- PD3: inline-JSON callback could overwrite an MCP-written canonical
V1 file with adapted-from-prose garbage. Callback now inspects
existing files and skips when V1 is already present.
- PD4: FORGEJO_URL = .rstrip("/api/v1") is a character-set strip —
catastrophic for hosts whose path contains /v1 in the middle.
Replaced with explicit endswith()-based suffix strip.
- CA10: clone URL embedded $FORGEJO_TOKEN, persisted into
.git/config where any agent could cat it. Token now sourced via
local credential.helper at clone-time, URL kept clean.
- CA12: state.finalized was set BEFORE the file write, so disk-full
/ OSError left the agent unable to retry finalize. Reordered.
HIGHS (M2):
- CA3/PD12: output_path validation (NUL-byte rejection, must be
absolute, parent-not-file check) in finalize_and_emit.
- CA6: ci_status_poll SELECT only considered implementer attempts;
conflict_resolver also pushes commits. SQL now unions both roles.
- PD9: ci_status_poll could advance on a stale "resolved" SHA from a
blocked attempt (whose head_sha_after == head_sha_before). Added
outcome='resolved' filter.
- CA8: cancelled/stale CI states mapped to ci_red_retry_same_tier,
burning pickup_count on healthy PRs. Both now wait (treated as
operator/system action, not failure). timed_out stays red.
- TE9: unknown Forgejo CI states now WARN-log instead of silently
being treated as pending — operators see new state strings.
- PD8: ci_status_poll event_type strings standardized to match the
state-machine event names (ci_green / ci_red_retry_same_tier)
instead of legacy ci-green / ci-red.
- CA7: inline-JSON callback now checks lost_lock_check BEFORE write
so a file isn't staged after lock loss.
- PD10: atomic .tmp + os.replace writes in both MCP finalize and
inline callback so the poller never sees a half-written file.
- PD16: inline_output_callback exceptions now re-raise as WorkerError
instead of being silently logged (root cause was buried 30s later
in a canonical-output timeout).
- CA9: WorkerConfig manual rebuild on --max-concurrent/--poll-interval
silently dropped new fields. Use dataclasses.replace, matching
round-4 P5 fix in master/__main__.py.
MEDIUMS (M3) — legacy_adapter quality upgrades:
- PD1: unrecognized confidence values now WARN instead of silently
defaulting to "medium" — surfaces agent prompt drift.
- PD2: estimator recommended_tier clamped to {0,1,2} so an out-of-
range int doesn't bypass the adapter's whole purpose.
- PD7: reviewer blocking_issues list-of-strings coerced into the
list-of-BlockingIssue-dict shape strict_parse requires.
- PD13: conflict_resolver prompt defaults tier=1 + warns instead of
raising; the scheduler always sets it but defends against drift.
- PD14: summarizer summary < 50 chars padded with a clear marker so
strict_parse accepts it (and the truncation is visible).
- PD15: implementer blockers capped at 4096 chars each so a buggy
agent can't blow up audit log / DB column.
- PD17: launch script accepts either FORGEJO_TOKEN or GITEA_TOKEN
with a clear error if both are unset.
- PD22: conflict_resolver adapter accepts singular commit_sha
fallback, matching implementer.
- CA4: every adapter invocation logs role + payload key fingerprint
so operators can measure agent-migration progress.
- estimator + summarizer now have explicit _start tools (the prompts
already referenced them; previously absent → first call would fail).
TESTS (M4) — added 40 tests in test_post_review_fixes.py:
- Cross-session MCP state reset (implementer + reviewer + estimator
+ summarizer; intra-session double-start still rejected).
- finalize_and_emit output_path precedence (arg > env > stdout),
parent-dir creation, rejection of relative/NUL paths, failed-write
leaves state retryable.
- legacy_adapter quality: tier clamping, blocker cap, non-string
commit warning, blocking_issues string coercion, conflict_resolver
full roundtrip + non-resolved head clearing, summarizer padding,
confidence warning, V1-passthrough no-log.
- opencode.json registration parity: every MCP the prompts name is
registered with the correct module path.
- Per-role prompts mention {role}_output.json (canonical poller path)
+ the "DO NOT emit chat-JSON" directive.
- FORGEJO_URL suffix-strip parametrized table.
- agent_runner stale-file cleanup: prior-attempt file is unlinked
before a new session can read it as phantom output.
Also updated 2 pre-existing tests for the CA8 / PD8 / PD13 behavior
changes (cancelled→wait, event_type renaming, conflict_resolver
default-tier warning).
Total: 741 → 781 tests, 0 regressions.
DEFERRED (M5 follow-up — non-trial-blocking):
- CA5: head_sha verification via git cat-file (requires subprocess).
- CA11: discovery_interval_s wall-time cadence (vs iteration count).
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -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(
|
||||
|
||||
@@ -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"
|
||||
@@ -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"):
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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}
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)."""
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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),
|
||||
),
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user