From 13445b8fb0983b600ec8791b5293e5ae60fb4d2a Mon Sep 17 00:00:00 2001 From: hallerite Date: Fri, 21 Aug 2026 18:28:05 +0200 Subject: [PATCH 1/2] refactor(v1): rely on Prime Agent ACP completion --- tests/v1/test_e2e.py | 12 +- tests/v1/test_prime_agent_acp_lifecycle.py | 824 ------------------ verifiers/v1/acp/__init__.py | 96 +- verifiers/v1/acp/runner.py | 220 +---- verifiers/v1/harnesses/prime_agent/harness.py | 34 +- verifiers/v1/utils/score.py | 5 +- 6 files changed, 53 insertions(+), 1138 deletions(-) delete mode 100644 tests/v1/test_prime_agent_acp_lifecycle.py diff --git a/tests/v1/test_e2e.py b/tests/v1/test_e2e.py index d1a1e63b46..84e4b084ab 100644 --- a/tests/v1/test_e2e.py +++ b/tests/v1/test_e2e.py @@ -85,6 +85,12 @@ def pair(a: str, b: str, id: str, *extra_marks): pair("openclaw", "docker", "openclaw-acp-in-docker"), pair("pool", "prime", "pool-acp-in-prime"), pair("rlm", "prime", "rlm-acp-in-prime-vm"), + pytest.param( + "prime-agent", + "prime", + marks=[mark.prime], + id="prime-agent-acp-in-prime-vm", + ), ] # harness runtime x tool placement: every axis value once plus the two-container case @@ -257,8 +263,10 @@ async def test_acp_resume_with_tool(run_v1, harness, harness_runtime, tmp_path): assert len(segments) == 2 assert segments[0]["terminated"] is False assert segments[1]["terminated"] is False - # Kimi Code is broken upstream: its Responses adapter drops message `phase` on replay. - if harness.id != "kimi-code": + # Kimi Code drops message `phase` on replay. Prime Agent's Chat Completions + # replay omits provider-only response state; the interception server retains + # those model calls as separate branches instead of guessing a false lineage. + if harness.id not in ("kimi-code", "prime-agent"): assert trace.num_branches == 1 # Native MCP tools need not appear in the intercepted model request that # populates trace.tools; the ACP transcript is the source of truth for use. diff --git a/tests/v1/test_prime_agent_acp_lifecycle.py b/tests/v1/test_prime_agent_acp_lifecycle.py deleted file mode 100644 index 51d5fd3cf5..0000000000 --- a/tests/v1/test_prime_agent_acp_lifecycle.py +++ /dev/null @@ -1,824 +0,0 @@ -"""Focused tests for the Prime Agent ACP lifecycle consumer.""" - -import asyncio -import importlib.util -import sys -import types -from pathlib import Path - -import pytest - -from verifiers.v1.acp import ACPHarnessSession, _record_lifecycle_status -from verifiers.v1.harnesses.prime_agent.harness import ( - PrimeAgentHarnessConfig, - _autonomous_args, -) -from verifiers.v1.utils.score import read_answer_file_or_last_reply - -NAMESPACE = "ai.primeintellect.prime-agent" -CONFIG = { - "user_contents": ["task"], - "system_prompt": "", - "lifecycle_meta_namespace": NAMESPACE, -} - - -def load_runner(monkeypatch: pytest.MonkeyPatch): - acp = types.ModuleType("acp") - acp.PROTOCOL_VERSION = "0.11" - acp.Client = object - - class RequestError(RuntimeError): - def __init__(self, message, *, data=None): - super().__init__(message) - self.data = data - - acp.RequestError = RequestError - acp.image_block = lambda data, media_type: (data, media_type) - acp.spawn_agent_process = None - acp.text_block = lambda text: text - schema = types.ModuleType("acp.schema") - for name in ( - "AgentMessageChunk", - "AllowedOutcome", - "ClientCapabilities", - "DeniedOutcome", - "HttpMcpServer", - "PermissionOption", - "RequestPermissionResponse", - "SessionInfoUpdate", - "TextContentBlock", - "ToolCall", - "ToolCallUpdate", - ): - setattr(schema, name, type(name, (), {})) - monkeypatch.setitem(sys.modules, "acp", acp) - monkeypatch.setitem(sys.modules, "acp.schema", schema) - path = Path(__file__).parents[2] / "verifiers/v1/acp/runner.py" - spec = importlib.util.spec_from_file_location("prime_agent_acp_runner", path) - assert spec and spec.loader - module = importlib.util.module_from_spec(spec) - spec.loader.exec_module(module) - monkeypatch.setattr(module, "LATE_REPLY_GRACE_SECONDS", 0.01) - return module - - -def event(sequence, phase="event", turn=1, **values): - metadata = { - "promptTurnId": turn, - "eventSequence": sequence, - "phase": phase, - **values, - } - if phase == "responseBoundary": - metadata.setdefault("terminalQuiescenceExpected", True) - return metadata - - -def update(runner, metadata, text=None): - if text is None: - value = sys.modules["acp.schema"].SessionInfoUpdate() - else: - content = runner.TextContentBlock() - content.text = text - value = runner.AgentMessageChunk() - value.content = content - value.message_id = "message" - value.field_meta = {NAMESPACE: metadata} - return value - - -def tool_update(runner, metadata, status="completed"): - value = runner.ToolCallUpdate() - value.tool_call_id = "tool" - value.status = status - value.field_meta = {NAMESPACE: metadata} - return value - - -async def run_prompt(runner, client, updates, stop_reason="end_turn", config=CONFIG): - class Connection: - async def prompt(self, **kwargs): - for value in updates: - await client.session_update("session", value) - return types.SimpleNamespace(stop_reason=stop_reason) - - return await runner.prompt( - client, Connection(), None, "session", config, is_new=True - ) - - -@pytest.mark.asyncio -async def test_only_correlated_terminal_quiescence_completes(monkeypatch): - runner = load_runner(monkeypatch) - client = runner.VerifiersACPClient() - updates = [ - update(runner, event(1, compaction={"summary": "not an answer"}), "compact"), - update(runner, event(2, refinement={"status": "complete"}), "refine"), - update(runner, event(3, subagents=[{"id": "child"}]), "child"), - update(runner, event(4, turn=0), "foreign"), - update(runner, event(5), "final answer"), - update(runner, event(6, "responseBoundary", outcome="result")), - update( - runner, - event( - 7, - "terminalQuiescence", - outcome="result", - quiescence={ - "outstandingSubagents": 0, - "remainingAutonomousContinuations": 3, - }, - ), - ), - ] - - result = await run_prompt(runner, client, updates, "max_turn_requests") - - assert result["reply"] == "final answer" - assert result["stop_reason"] == "max_turn_requests" - assert result["lifecycle"]["phase"] == "terminalQuiescence" - - -@pytest.mark.asyncio -async def test_outer_timeout_bounds_waiting_text_without_terminal(monkeypatch): - runner = load_runner(monkeypatch) - client = runner.VerifiersACPClient() - updates = [ - update(runner, event(1), "waiting for children"), - update(runner, event(2, "responseBoundary", outcome="result")), - ] - - with pytest.raises(asyncio.TimeoutError): - await asyncio.wait_for(run_prompt(runner, client, updates), timeout=0.03) - - -@pytest.mark.asyncio -async def test_prompt_waits_for_delayed_terminal_quiescence(monkeypatch): - runner = load_runner(monkeypatch) - client = runner.VerifiersACPClient() - - class Connection: - async def prompt(self, **kwargs): - await client.session_update( - "session", update(runner, event(1), "final answer") - ) - await client.session_update( - "session", - update(runner, event(2, "responseBoundary", outcome="result")), - ) - - async def settle(): - await asyncio.sleep(0.03) - await client.session_update( - "session", - update( - runner, - event( - 3, - "terminalQuiescence", - outcome="result", - quiescence={ - "outstandingSubagents": 0, - "remainingAutonomousContinuations": 0, - }, - ), - ), - ) - - asyncio.create_task(settle()) - return types.SimpleNamespace(stop_reason="end_turn") - - result = await runner.prompt( - client, Connection(), None, "session", CONFIG, is_new=True - ) - - assert result["reply"] == "final answer" - assert result["lifecycle"]["phase"] == "terminalQuiescence" - - -@pytest.mark.asyncio -@pytest.mark.parametrize( - "events", - [ - [ - event(1, "responseBoundary", turn=True, outcome="result"), - event( - 2, - "terminalQuiescence", - turn=True, - outcome="result", - quiescence={ - "outstandingSubagents": False, - "remainingAutonomousContinuations": 0, - }, - ), - ], - [ - event(1, "responseBoundary", outcome="result"), - event( - 2, - "terminalQuiescence", - outcome="result", - quiescence={ - "outstandingSubagents": False, - "remainingAutonomousContinuations": 0, - }, - ), - ], - [ - event( - 1, - "terminalQuiescence", - outcome="result", - quiescence={ - "outstandingSubagents": 0, - "remainingAutonomousContinuations": 0, - }, - ) - ], - ], -) -async def test_malformed_or_unordered_terminal_never_completes(monkeypatch, events): - runner = load_runner(monkeypatch) - client = runner.VerifiersACPClient() - updates = [update(runner, value) for value in events] - - with pytest.raises(RuntimeError, match="Prime Agent"): - await run_prompt(runner, client, updates) - - -@pytest.mark.asyncio -@pytest.mark.parametrize("sequences", [[1, 1], [2, 1]]) -async def test_non_monotonic_lifecycle_sequence_fails(monkeypatch, sequences): - runner = load_runner(monkeypatch) - client = runner.VerifiersACPClient() - updates = [ - update(runner, event(sequences[0]), "answer"), - update(runner, event(sequences[1], "responseBoundary", outcome="result")), - ] - - with pytest.raises(RuntimeError, match="eventSequence"): - await run_prompt(runner, client, updates) - - -@pytest.mark.asyncio -async def test_terminal_error_is_not_autonomous_completion(monkeypatch): - runner = load_runner(monkeypatch) - client = runner.VerifiersACPClient() - updates = [ - update(runner, event(1), "partial answer"), - update(runner, event(2, "responseBoundary", outcome="error")), - update( - runner, - event( - 3, - "terminalQuiescence", - outcome="error", - quiescence={ - "outstandingSubagents": 0, - "remainingAutonomousContinuations": 0, - }, - ), - ), - ] - - with pytest.raises(RuntimeError, match="terminal lifecycle error"): - await run_prompt(runner, client, updates) - - -def test_prime_agent_eval_defaults_to_autonomous_with_lifecycle_opt_in(): - assert PrimeAgentHarnessConfig().autonomous is True - assert PrimeAgentHarnessConfig(autonomous=False).autonomous is False - assert PrimeAgentHarnessConfig().require_terminal_quiescence is False - assert ( - PrimeAgentHarnessConfig( - require_terminal_quiescence=True - ).require_terminal_quiescence - is True - ) - - -def test_autonomous_mode_receives_verifiers_rollout_budget(): - trace = types.SimpleNamespace( - agent=types.SimpleNamespace( - config=types.SimpleNamespace( - max_turns=9, - max_total_tokens=65_536, - timeout=types.SimpleNamespace(rollout=123.4561), - ) - ) - ) - - assert _autonomous_args(True, trace) == [ - "--autonomous", - "--autonomous-max-turns", - "9", - "--autonomous-max-tokens", - "65536", - "--autonomous-timeout-ms", - "123457", - ] - assert _autonomous_args(False, trace) == [] - - -@pytest.mark.parametrize("version", [".", "..", "+", "-unsafe"]) -def test_prime_agent_version_rejects_path_like_values(version): - with pytest.raises(ValueError): - PrimeAgentHarnessConfig(version=version) - - -@pytest.mark.asyncio -async def test_legacy_agent_remains_compatible_without_lifecycle_namespace(monkeypatch): - runner = load_runner(monkeypatch) - client = runner.VerifiersACPClient() - legacy_config = {**CONFIG, "lifecycle_meta_namespace": None} - - result = await run_prompt( - runner, - client, - [update(runner, event(1), "legacy answer")], - config=legacy_config, - ) - - assert result == { - "reply": "legacy answer", - "stop_reason": "end_turn", - "response_boundary": None, - "lifecycle": None, - } - assert legacy_config["lifecycle_meta_namespace"] is None - - -@pytest.mark.asyncio -async def test_lifecycle_state_persists_across_prompt_turns(monkeypatch): - runner = load_runner(monkeypatch) - client = runner.VerifiersACPClient() - first = [ - update(runner, event(1), "first answer"), - update(runner, event(2, "responseBoundary", outcome="result")), - update( - runner, - event( - 3, - "terminalQuiescence", - outcome="result", - quiescence={ - "outstandingSubagents": 0, - "remainingAutonomousContinuations": 0, - }, - ), - ), - ] - second = [ - update(runner, event(4, turn=2), "second answer"), - update( - runner, - event(5, "responseBoundary", turn=2, outcome="result"), - ), - update( - runner, - event( - 6, - "terminalQuiescence", - turn=2, - outcome="result", - quiescence={ - "outstandingSubagents": 0, - "remainingAutonomousContinuations": 0, - }, - ), - ), - ] - - first_result = await run_prompt(runner, client, first) - second_result = await run_prompt(runner, client, second) - - assert first_result["reply"] == "first answer" - assert second_result["reply"] == "second answer" - assert second_result["lifecycle"]["promptTurnId"] == 2 - - -@pytest.mark.asyncio -async def test_request_error_waits_for_correlated_terminal_error(monkeypatch): - runner = load_runner(monkeypatch) - client = runner.VerifiersACPClient() - - class Connection: - async def prompt(self, **kwargs): - async def settle(): - await asyncio.sleep(0) - await client.session_update( - "session", - update(runner, event(1, "responseBoundary", outcome="error")), - ) - await client.session_update( - "session", - update( - runner, - event( - 2, - "terminalQuiescence", - outcome="error", - quiescence={ - "outstandingSubagents": 0, - "remainingAutonomousContinuations": 0, - }, - ), - ), - ) - - asyncio.create_task(settle()) - raise runner.RequestError( - "request failed", data={"details": "model request failed"} - ) - - with pytest.raises(RuntimeError, match="model request failed"): - await runner.prompt(client, Connection(), None, "session", CONFIG, is_new=True) - assert client.terminal_quiescence is not None - assert client.terminal_quiescence["outcome"] == "error" - - -@pytest.mark.asyncio -async def test_foreign_tool_update_cannot_complete_current_turn(monkeypatch): - runner = load_runner(monkeypatch) - client = runner.VerifiersACPClient() - updates = [ - tool_update(runner, event(1, turn=0)), - update(runner, event(2, "responseBoundary", outcome="result")), - update( - runner, - event( - 3, - "terminalQuiescence", - outcome="result", - quiescence={ - "outstandingSubagents": 0, - "remainingAutonomousContinuations": 0, - }, - ), - ), - ] - config = {**CONFIG, "allow_empty_tool_reply": True} - - class Connection: - async def prompt(self, **kwargs): - for value in updates: - await client.session_update("session", value) - return types.SimpleNamespace(stop_reason="end_turn") - - with pytest.raises(RuntimeError, match="no visible reply"): - await runner.prompt(client, Connection(), None, "session", config, is_new=True) - - -@pytest.mark.asyncio -async def test_precommit_request_error_does_not_wait_for_terminal(monkeypatch): - runner = load_runner(monkeypatch) - client = runner.VerifiersACPClient() - - class Connection: - async def prompt(self, **kwargs): - async def publish_boundary(): - await asyncio.sleep(0) - await client.session_update( - "session", - update( - runner, - event( - 1, - "responseBoundary", - outcome="error", - terminalQuiescenceExpected=False, - ), - ), - ) - - asyncio.create_task(publish_boundary()) - raise runner.RequestError( - "request failed", data={"details": "admission failed"} - ) - - with pytest.raises(RuntimeError, match="admission failed"): - await runner.prompt(client, Connection(), None, "session", CONFIG, is_new=True) - assert client.response_boundary is not None - assert client.response_boundary["terminalQuiescenceExpected"] is False - assert client.terminal_quiescence is None - - -@pytest.mark.asyncio -async def test_malformed_lifecycle_envelope_fails_promptly(monkeypatch): - runner = load_runner(monkeypatch) - client = runner.VerifiersACPClient() - malformed = sys.modules["acp.schema"].SessionInfoUpdate() - malformed.field_meta = {NAMESPACE: "not an object"} - - with pytest.raises(RuntimeError, match="metadata must be an object"): - await run_prompt(runner, client, [malformed]) - - -@pytest.mark.asyncio -async def test_ignored_update_kind_wakes_lifecycle_error_waiter(monkeypatch): - runner = load_runner(monkeypatch) - client = runner.VerifiersACPClient() - - class IgnoredUpdate: - def __init__(self): - self.field_meta = {NAMESPACE: "not an object"} - - class Connection: - async def prompt(self, **kwargs): - await client.session_update( - "session", update(runner, event(1), "partial answer") - ) - - async def publish_malformed_update(): - await asyncio.sleep(0) - await client.session_update("session", IgnoredUpdate()) - - asyncio.create_task(publish_malformed_update()) - return types.SimpleNamespace(stop_reason="end_turn") - - with pytest.raises(RuntimeError, match="metadata must be an object"): - await asyncio.wait_for( - runner.prompt(client, Connection(), None, "session", CONFIG, is_new=True), - timeout=0.03, - ) - - -def test_lifecycle_status_is_separate_from_benchmark_reward(): - trace = types.SimpleNamespace( - info={}, rewards={"benchmark": 0.75}, stop_condition=None - ) - boundary = event(8, "responseBoundary", turn=4, outcome="result") - terminal = event( - 9, - "terminalQuiescence", - turn=4, - outcome="result", - quiescence={"outstandingSubagents": 0}, - ) - _record_lifecycle_status( - trace, - NAMESPACE, - { - "ok": True, - "reply": "main answer", - "stop_reason": "max_turn_requests", - "response_boundary": boundary, - "lifecycle": terminal, - }, - ) - - assert trace.info["acp_lifecycle"][NAMESPACE][0] == { - "prompt_turn_id": 4, - "stop_reason": "max_turn_requests", - "infrastructure_status": "ok", - "autonomous_completion": True, - "terminal_quiescence_observed": True, - "last_lifecycle_phase": "terminalQuiescence", - "response_boundary": boundary, - "terminal_quiescence": terminal, - } - assert trace.info["acp_answer_fallback"] == "main answer" - assert trace.rewards == {"benchmark": 0.75} - assert trace.stop_condition is None - - -def test_lifecycle_status_preserves_available_response_boundary_phase(): - trace = types.SimpleNamespace(info={}) - boundary = event( - 3, - "responseBoundary", - outcome="error", - terminalQuiescenceExpected=False, - ) - - _record_lifecycle_status( - trace, - NAMESPACE, - {"ok": False, "response_boundary": boundary}, - ) - - status = trace.info["acp_lifecycle"][NAMESPACE][0] - assert status["infrastructure_status"] == "error" - assert status["autonomous_completion"] is False - assert status["terminal_quiescence_observed"] is False - assert status["last_lifecycle_phase"] == "responseBoundary" - - -class _BlockingProcess: - async def write(self, data): - assert data - - -class _BlockingReader: - async def read(self): - await asyncio.Event().wait() - - -def _incomplete_session(namespace=NAMESPACE): - session = object.__new__(ACPHarnessSession) - session.config = types.SimpleNamespace( - prompt="task", - command=["agent"], - system_prompt=None, - session_meta=None, - allow_empty_tool_reply=False, - lifecycle_meta_namespace=namespace, - ) - session.mcp_urls = {} - session._lock = asyncio.Lock() - session._closed = False - session._process = _BlockingProcess() - session._reader = _BlockingReader() - session.trace = types.SimpleNamespace( - info={}, rewards={"benchmark": 0.75}, calls=[], stop_condition=None - ) - stopped = [] - - async def stop(*, graceful): - stopped.append(graceful) - - session._stop = stop - return session, stopped - - -_INCOMPLETE_STATUS = { - "prompt_turn_id": None, - "stop_reason": None, - "infrastructure_status": "error", - "autonomous_completion": False, - "terminal_quiescence_observed": False, - "last_lifecycle_phase": None, - "response_boundary": None, - "terminal_quiescence": None, -} - - -@pytest.mark.asyncio -@pytest.mark.parametrize("cancel_with_timeout", [False, True]) -async def test_cancelled_turn_records_incomplete_lifecycle_status(cancel_with_timeout): - session, stopped = _incomplete_session() - turn = asyncio.create_task(session._run(None)) - await asyncio.sleep(0) - if cancel_with_timeout: - with pytest.raises(asyncio.TimeoutError): - await asyncio.wait_for(turn, timeout=0.01) - else: - turn.cancel() - with pytest.raises(asyncio.CancelledError): - await turn - - assert session.trace.info["acp_lifecycle"][NAMESPACE] == [_INCOMPLETE_STATUS] - assert "acp_answer_fallback" not in session.trace.info - assert session.trace.rewards == {"benchmark": 0.75} - assert stopped == [False] - - -@pytest.mark.asyncio -async def test_lock_wait_cancellation_records_status_without_stopping_process(): - session, stopped = _incomplete_session() - await session._lock.acquire() - turn = asyncio.create_task(session._run(None)) - await asyncio.sleep(0) - - turn.cancel() - with pytest.raises(asyncio.CancelledError): - await turn - session._lock.release() - - assert session.trace.info["acp_lifecycle"][NAMESPACE] == [_INCOMPLETE_STATUS] - assert stopped == [] - - -@pytest.mark.asyncio -async def test_start_cancellation_records_status_and_stops_locked_turn(): - session, stopped = _incomplete_session() - session._process = None - session._reader = None - start_entered = asyncio.Event() - - async def start(): - start_entered.set() - await asyncio.Event().wait() - - session._start = start - turn = asyncio.create_task(session._run(None)) - await start_entered.wait() - turn.cancel() - - with pytest.raises(asyncio.CancelledError): - await turn - - assert session.trace.info["acp_lifecycle"][NAMESPACE] == [_INCOMPLETE_STATUS] - assert stopped == [False] - - -@pytest.mark.asyncio -async def test_failed_turn_teardown_finishes_before_next_turn_starts(): - session, _ = _incomplete_session() - read_entered = asyncio.Event() - fail_read = asyncio.Event() - teardown_entered = asyncio.Event() - finish_teardown = asyncio.Event() - second_start = asyncio.Event() - second_write = asyncio.Event() - - class FirstReader: - async def read(self): - read_entered.set() - await fail_read.wait() - raise RuntimeError("packet read failed") - - class SecondProcess: - async def write(self, data): - assert data - second_write.set() - - async def start(): - second_start.set() - session._process = SecondProcess() - session._reader = _BlockingReader() - - async def stop(*, graceful): - assert graceful is False - teardown_entered.set() - await finish_teardown.wait() - session._process = None - session._reader = None - - session._reader = FirstReader() - session._start = start - session._stop = stop - - first = asyncio.create_task(session._run(None)) - await read_entered.wait() - second = asyncio.create_task(session._run(None)) - fail_read.set() - await teardown_entered.wait() - await asyncio.sleep(0) - - assert not second_start.is_set() - assert not second_write.is_set() - - finish_teardown.set() - with pytest.raises(RuntimeError, match="packet read failed"): - await first - await second_start.wait() - await second_write.wait() - second.cancel() - with pytest.raises(asyncio.CancelledError): - await second - - -@pytest.mark.asyncio -async def test_cancelled_turn_without_lifecycle_namespace_only_stops_process(): - session, stopped = _incomplete_session(namespace=None) - turn = asyncio.create_task(session._run(None)) - await asyncio.sleep(0) - turn.cancel() - - with pytest.raises(asyncio.CancelledError): - await turn - - assert session.trace.info == {} - assert stopped == [False] - - -@pytest.mark.asyncio -async def test_status_recording_failure_does_not_mask_turn_exception(monkeypatch): - session, stopped = _incomplete_session() - original = RuntimeError("packet read failed") - - class FailingReader: - async def read(self): - raise original - - def fail_recording(*args, **kwargs): - raise TypeError("malformed trace.info") - - session._reader = FailingReader() - monkeypatch.setattr("verifiers.v1.acp._record_lifecycle_status", fail_recording) - - with pytest.raises(RuntimeError, match="packet read failed") as error: - await session._run(None) - - assert error.value is original - assert stopped == [False] - - -@pytest.mark.asyncio -async def test_answer_fallback_cannot_select_child_branch_text(): - class MissingAnswerRuntime: - async def read(self, path): - raise FileNotFoundError(path) - - trace = types.SimpleNamespace( - info={"acp_answer_fallback": "main answer"}, - last_reply="child branch text", - ) - answer = await read_answer_file_or_last_reply( - MissingAnswerRuntime(), "/missing/answer", trace - ) - - assert answer == "main answer" diff --git a/verifiers/v1/acp/__init__.py b/verifiers/v1/acp/__init__.py index 6d3d85620a..e770e14f39 100644 --- a/verifiers/v1/acp/__init__.py +++ b/verifiers/v1/acp/__init__.py @@ -42,8 +42,6 @@ class ACPConfig: system_prompt: str | None = None session_meta: JsonObject | None = None allow_empty_tool_reply: bool = False - lifecycle_meta_namespace: str | None = None - """Namespaced ACP lifecycle contract required to finish each prompt turn.""" class ACPHarness(Harness[ConfigT]): @@ -118,45 +116,6 @@ def _packet(value: JsonObject) -> bytes: return len(data).to_bytes(8, "big") + data -def _record_lifecycle_status( - trace: Trace, namespace: str, response: JsonObject -) -> None: - """Record agent lifecycle as infrastructure status, never benchmark reward.""" - stop_reason = response.get("stop_reason") - response_boundary = response.get("response_boundary") - lifecycle = response.get("lifecycle") - if stop_reason is not None and not isinstance(stop_reason, str): - raise TypeError("ACP stop reason must be a string or null") - if response_boundary is not None and not isinstance(response_boundary, dict): - raise TypeError("ACP response boundary must be an object or null") - if lifecycle is not None and not isinstance(lifecycle, dict): - raise TypeError("ACP lifecycle status must be an object or null") - infrastructure_ok = response.get("ok") is True - last_lifecycle = lifecycle or response_boundary - terminal_quiescence_observed = bool( - lifecycle and lifecycle.get("phase") == "terminalQuiescence" - ) - status = { - "prompt_turn_id": (last_lifecycle or {}).get("promptTurnId"), - "stop_reason": stop_reason, - "infrastructure_status": "ok" if infrastructure_ok else "error", - "autonomous_completion": bool( - infrastructure_ok - and terminal_quiescence_observed - and lifecycle - and lifecycle.get("outcome") == "result" - ), - "terminal_quiescence_observed": terminal_quiescence_observed, - "last_lifecycle_phase": (last_lifecycle or {}).get("phase"), - "response_boundary": response_boundary, - "terminal_quiescence": lifecycle, - } - trace.info.setdefault("acp_lifecycle", {}).setdefault(namespace, []).append(status) - reply = response.get("reply") - if status["autonomous_completion"]: - trace.info["acp_answer_fallback"] = reply if isinstance(reply, str) else "" - - def _require_model_turn(trace: Trace, calls_before: int, result: ProgramResult) -> None: if ( result.exit_code @@ -257,44 +216,25 @@ async def _run(self, messages: Messages | None) -> ProgramResult: "system_prompt": self.config.system_prompt or "", "session_meta": self.config.session_meta or {}, "allow_empty_tool_reply": self.config.allow_empty_tool_reply, - "lifecycle_meta_namespace": self.config.lifecycle_meta_namespace, } - lock_acquired = False - try: - async with self._lock: - lock_acquired = True - locked_turn_started = False - try: - # Closed-session rejection is validation, not an attempted turn. - if self._closed: - raise HarnessError( - f"harness {self.harness.config.id!r} session is already closed" - ) - locked_turn_started = True - if self._process is None: - await self._start() - assert self._process is not None - assert self._reader is not None - calls_before = len(self.trace.calls) - await self._process.write( - _packet({"operation": "prompt", "config": config}) - ) - response = await self._reader.read() - except BaseException: - if locked_turn_started: - with contextlib.suppress(BaseException): - if namespace := self.config.lifecycle_meta_namespace: - _record_lifecycle_status(self.trace, namespace, {}) - await run_shielded(self._stop(graceful=False)) - raise - except asyncio.CancelledError: - if not lock_acquired: - with contextlib.suppress(BaseException): - if namespace := self.config.lifecycle_meta_namespace: - _record_lifecycle_status(self.trace, namespace, {}) - raise - if namespace := self.config.lifecycle_meta_namespace: - _record_lifecycle_status(self.trace, namespace, response) + async with self._lock: + if self._closed: + raise HarnessError( + f"harness {self.harness.config.id!r} session is already closed" + ) + if self._process is None: + await self._start() + assert self._process is not None + assert self._reader is not None + calls_before = len(self.trace.calls) + try: + await self._process.write( + _packet({"operation": "prompt", "config": config}) + ) + response = await self._reader.read() + except BaseException: + await run_shielded(self._stop(graceful=False)) + raise if not response.get("ok"): detail = response.get("error") or "ACP session request failed" if stderr := self._stderr(): diff --git a/verifiers/v1/acp/runner.py b/verifiers/v1/acp/runner.py index 6181f0f8e9..a014cbaf2c 100644 --- a/verifiers/v1/acp/runner.py +++ b/verifiers/v1/acp/runner.py @@ -35,147 +35,38 @@ ) MAX_PACKET_BYTES = 128 * 1024 * 1024 -LATE_REPLY_GRACE_SECONDS = 1.0 +LATE_UPDATE_GRACE_SECONDS = 1.0 class VerifiersACPClient(Client): - """ACP output collector with an opt-in correlated lifecycle consumer.""" - - _NON_ANSWER_META = frozenset(("compaction", "refinement", "subagents")) - def __init__(self) -> None: self.visible_reply = "" self.message_id: str | None = None self.tool_calls: dict[str, str] = {} self.output_changed = asyncio.Condition() - self.lifecycle_namespace: str | None = None - self.stop_reason: str | None = None - self.prompt_turn_id = 0 - self.response_boundary: dict[str, Any] | None = None - self.terminal_quiescence: dict[str, Any] | None = None - self.lifecycle_error: str | None = None - self._last_event_sequence = 0 - - def reset(self, lifecycle_namespace: str | None = None) -> None: + + def reset(self) -> None: self.visible_reply = "" self.message_id = None self.tool_calls = {} - self.lifecycle_namespace = lifecycle_namespace - self.stop_reason = None - self.response_boundary = None - self.terminal_quiescence = None - self.lifecycle_error = None - if lifecycle_namespace is not None: - self.prompt_turn_id += 1 - - def _lifecycle_meta(self, update: Any) -> dict[str, Any] | None: - if self.lifecycle_namespace is None: - return None - field_meta = getattr(update, "field_meta", None) - if not isinstance(field_meta, dict): - return None - if self.lifecycle_namespace not in field_meta: - return None - event = field_meta[self.lifecycle_namespace] - if not isinstance(event, dict): - self.lifecycle_error = "Prime Agent lifecycle metadata must be an object" - return None - return event - - def _consume_lifecycle(self, event: dict[str, Any] | None) -> None: - if event is None: - return - sequence = event.get("eventSequence") - if type(sequence) is not int or sequence <= self._last_event_sequence: - self.lifecycle_error = "Prime Agent lifecycle eventSequence is invalid" - return - self._last_event_sequence = sequence - turn_id = event.get("promptTurnId") - if type(turn_id) is not int: - self.lifecycle_error = "Prime Agent lifecycle promptTurnId is invalid" - return - if turn_id != self.prompt_turn_id: - return - phase = event.get("phase") - if phase == "responseBoundary": - outcome = event.get("outcome") - terminal_expected = event.get("terminalQuiescenceExpected") - if ( - outcome not in ("result", "error") - or type(terminal_expected) is not bool - or (outcome == "result" and not terminal_expected) - ): - self.lifecycle_error = "Prime Agent responseBoundary is malformed" - elif self.response_boundary is not None: - self.lifecycle_error = "Prime Agent emitted duplicate responseBoundary" - else: - self.response_boundary = event - return - if phase != "terminalQuiescence": - return - quiescence = event.get("quiescence") - outstanding = ( - quiescence.get("outstandingSubagents") - if isinstance(quiescence, dict) - else None - ) - remaining = ( - quiescence.get("remainingAutonomousContinuations") - if isinstance(quiescence, dict) - else None - ) - if ( - self.response_boundary is None - or event.get("outcome") != self.response_boundary.get("outcome") - or type(outstanding) is not int - or outstanding != 0 - or type(remaining) is not int - or remaining < 0 - ): - self.lifecycle_error = "Prime Agent terminalQuiescence is malformed" - return - if self.terminal_quiescence is not None: - self.lifecycle_error = "Prime Agent emitted duplicate terminalQuiescence" - return - self.terminal_quiescence = event - - def _is_current_turn_event(self, event: dict[str, Any] | None) -> bool: - if self.lifecycle_namespace is None: - return True - return bool( - event - and type(event.get("promptTurnId")) is int - and event.get("promptTurnId") == self.prompt_turn_id - and event.get("phase") == "event" - ) - - def _is_answer_chunk(self, event: dict[str, Any] | None) -> bool: - return self._is_current_turn_event(event) and not ( - event and self._NON_ANSWER_META.intersection(event) - ) async def session_update(self, session_id: str, update: Any, **kwargs: Any) -> None: async with self.output_changed: - event = self._lifecycle_meta(update) - self._consume_lifecycle(event) if isinstance(update, ToolCall): - if self._is_current_turn_event(event): - self.tool_calls[update.tool_call_id] = update.status or "pending" + self.tool_calls[update.tool_call_id] = update.status or "pending" elif isinstance(update, ToolCallUpdate): - if update.status and self._is_current_turn_event(event): + if update.status: self.tool_calls[update.tool_call_id] = update.status - elif ( - isinstance(update, AgentMessageChunk) - and isinstance(update.content, TextContentBlock) - and self._is_answer_chunk(event) + elif isinstance(update, AgentMessageChunk) and isinstance( + update.content, TextContentBlock ): message_id = getattr(update, "message_id", None) if message_id is not None and message_id != self.message_id: self.visible_reply = "" self.message_id = message_id self.visible_reply += update.content.text - # Lifecycle metadata can accompany update kinds we otherwise ignore. - # Always wake lifecycle waiters after consuming it. + else: + return self.output_changed.notify_all() async def request_permission( @@ -241,7 +132,7 @@ async def prompt( config: dict, *, is_new: bool, -) -> dict[str, Any]: +) -> str: prompt_capabilities = capabilities and capabilities.prompt_capabilities supports_images = bool(prompt_capabilities and prompt_capabilities.image) blocks = [] @@ -250,13 +141,12 @@ async def prompt( blocks.extend(user_content_blocks(config["user_contents"], supports_images)) if not blocks: raise ValueError("ACP prompt has no content") - client.reset(config.get("lifecycle_meta_namespace")) - prompt_error: RequestError | None = None + client.reset() try: response = await connection.prompt(session_id=session_id, prompt=blocks) - client.stop_reason = response.stop_reason except RequestError as error: - prompt_error = error + detail = error.data.get("details") if isinstance(error.data, dict) else None + raise RuntimeError(detail or str(error)) from error # ACP 0.11 dispatches notifications in background tasks but resolves a request # response directly in its receive loop. An agent that sends its final @@ -266,96 +156,29 @@ async def prompt( def has_visible_reply() -> bool: return bool(client.visible_reply.strip()) - if prompt_error is None and not has_visible_reply(): + if not has_visible_reply(): async with client.output_changed: try: await asyncio.wait_for( client.output_changed.wait_for(has_visible_reply), - timeout=LATE_REPLY_GRACE_SECONDS, - ) - except asyncio.TimeoutError: # noqa: UP041 - Python 3.10 compatibility - pass - - if ( - client.lifecycle_namespace is not None - and prompt_error is not None - and client.response_boundary is None - and client.lifecycle_error is None - ): - # Prime drains its boundary notification before returning a request error, - # while ACP 0.11 dispatches that notification in a background task. - async with client.output_changed: - try: - await asyncio.wait_for( - client.output_changed.wait_for( - lambda: ( - client.response_boundary is not None - or client.lifecycle_error is not None - ) - ), - timeout=LATE_REPLY_GRACE_SECONDS, + timeout=LATE_UPDATE_GRACE_SECONDS, ) except asyncio.TimeoutError: # noqa: UP041 - Python 3.10 compatibility pass - terminal_expected = prompt_error is None or bool( - client.response_boundary - and client.response_boundary.get("terminalQuiescenceExpected") is True - ) - if ( - client.lifecycle_namespace is not None - and terminal_expected - and client.terminal_quiescence is None - and client.lifecycle_error is None - ): - # Do not impose a short protocol grace here: descendants can settle long after - # the prompt response. The owning rollout/action timeout remains the hard bound. - async with client.output_changed: - await client.output_changed.wait_for( - lambda: ( - client.terminal_quiescence is not None - or client.lifecycle_error is not None - ) - ) - if client.lifecycle_error is not None: - raise RuntimeError(client.lifecycle_error) - if ( - client.lifecycle_namespace is not None - and terminal_expected - and client.terminal_quiescence is None - ): - raise RuntimeError( - "Prime Agent prompt returned without correlated terminalQuiescence " - f"(stop_reason={client.stop_reason})" - ) - if prompt_error is not None: - data = getattr(prompt_error, "data", None) - detail = data.get("details") if isinstance(data, dict) else None - raise RuntimeError(detail or str(prompt_error)) from prompt_error - if ( - client.terminal_quiescence is not None - and client.terminal_quiescence["outcome"] == "error" - ): - raise RuntimeError("Prime Agent reported a terminal lifecycle error") - tool_statuses = list(client.tool_calls.values()) completed_tool_turn = ( config.get("allow_empty_tool_reply", False) - and client.stop_reason == "end_turn" + and response.stop_reason == "end_turn" and bool(tool_statuses) and all(status in ("completed", "failed") for status in tool_statuses) ) if not has_visible_reply() and not completed_tool_turn: raise RuntimeError( "ACP agent produced no visible reply " - f"(stop_reason={client.stop_reason}, tool_statuses={tool_statuses})" + f"(stop_reason={response.stop_reason}, tool_statuses={tool_statuses})" ) - return { - "reply": client.visible_reply, - "stop_reason": client.stop_reason, - "response_boundary": client.response_boundary, - "lifecycle": client.terminal_quiescence, - } + return client.visible_reply class ACPSession: @@ -403,7 +226,7 @@ async def start(self, config: dict) -> None: self.session_id = session.session_id self.is_new = True - async def run(self, config: dict) -> dict[str, Any]: + async def run(self, config: dict) -> str: if self.connection is None: await self.start(config) assert self.session_id is not None @@ -472,7 +295,7 @@ async def serve_stream() -> None: if operation == "prompt": response = { "ok": True, - **await session.run(request["config"]), + "reply": await session.run(request["config"]), } elif operation == "shutdown": await session.close() @@ -485,9 +308,6 @@ async def serve_stream() -> None: response = { "ok": False, "error": f"{type(error).__name__}: {error}", - "stop_reason": session.client.stop_reason, - "response_boundary": session.client.response_boundary, - "lifecycle": session.client.terminal_quiescence, } write_packet(sys.stdout.buffer, response) if stop: diff --git a/verifiers/v1/harnesses/prime_agent/harness.py b/verifiers/v1/harnesses/prime_agent/harness.py index 3925428856..26157d3de5 100644 --- a/verifiers/v1/harnesses/prime_agent/harness.py +++ b/verifiers/v1/harnesses/prime_agent/harness.py @@ -3,7 +3,6 @@ import hashlib import json import logging -import math import shlex from pydantic import Field @@ -23,26 +22,10 @@ STATE_ROOT = "/tmp/vf-prime-agent-runs" SKILLS_DIR = ".agents/skills" PROVIDER = "intercept" -LIFECYCLE_META_NAMESPACE = "ai.primeintellect.prime-agent" KEY_VAR = "PRIME_AGENT_INTERCEPT_KEY" ENV_AGENT_DIR = "PRIME_AGENT_CODING_AGENT_DIR" -def _autonomous_args(enabled: bool, trace: Trace) -> list[str]: - if not enabled: - return [] - config = trace.agent.config - args = ["--autonomous"] - if config.max_turns is not None and config.max_turns > 0: - args += ["--autonomous-max-turns", str(config.max_turns)] - if config.max_total_tokens is not None and config.max_total_tokens > 0: - args += ["--autonomous-max-tokens", str(config.max_total_tokens)] - rollout_timeout = config.timeout.rollout - if rollout_timeout is not None and rollout_timeout > 0: - args += ["--autonomous-timeout-ms", str(math.ceil(rollout_timeout * 1000))] - return args - - INSTALL = r""" set -e export PATH="/var/tmp/vf-node/bin:$PATH" @@ -58,15 +41,12 @@ def _autonomous_args(enabled: bool, trace: Trace) -> list[str]: class PrimeAgentHarnessConfig(HarnessConfig): - version: str = Field(default="0.7.3", pattern=r"^[A-Za-z0-9][A-Za-z0-9._+-]*$") + version: str = Field( + default="0.7.4-beta.533.1.848081e", + pattern=r"^[A-Za-z0-9][A-Za-z0-9._+-]*$", + ) """Prime Agent release to install, pinned for reproducibility.""" - autonomous: bool = True - """Run Prime Agent in autonomous mode unless an evaluation opts out.""" - - require_terminal_quiescence: bool = False - """Require the correlated lifecycle contract provided by compatible releases.""" - class PrimeAgentHarness(ACPHarness[PrimeAgentHarnessConfig]): APPENDS_SYSTEM_PROMPT = True @@ -174,7 +154,6 @@ async def prepare_acp( f"{root}/daemon.sock", "--offline", ] - args.extend(_autonomous_args(self.config.autonomous, trace)) for skill in self.config.skills: args += ["--skill", f"{SKILLS_DIR}/{skill.resolve().name}"] if system_prompt: @@ -201,11 +180,6 @@ async def prepare_acp( command=[wrapper], prompt=prompt, allow_empty_tool_reply=True, - lifecycle_meta_namespace=( - LIFECYCLE_META_NAMESPACE - if self.config.require_terminal_quiescence - else None - ), ) async def cleanup(self, trace: Trace, runtime: Runtime) -> None: diff --git a/verifiers/v1/utils/score.py b/verifiers/v1/utils/score.py index 785f5d5154..f3a9087670 100644 --- a/verifiers/v1/utils/score.py +++ b/verifiers/v1/utils/score.py @@ -45,10 +45,7 @@ async def read_answer_file_or_last_reply( answer = (await runtime.read(path)).decode(errors="replace").strip() except (FileNotFoundError, OSError, SandboxError): answer = "" - acp_fallback = trace.info.get("acp_answer_fallback") - return answer or ( - acp_fallback if isinstance(acp_fallback, str) else trace.last_reply - ) + return answer or trace.last_reply def parse_judge_choice( From 2d023b3f35303cccae10d72621ef8a8c2ca0a4d4 Mon Sep 17 00:00:00 2001 From: hallerite Date: Fri, 21 Aug 2026 19:30:55 +0200 Subject: [PATCH 2/2] test(v1): require linear Prime Agent traces --- tests/v1/test_e2e.py | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/tests/v1/test_e2e.py b/tests/v1/test_e2e.py index 84e4b084ab..283784e9c6 100644 --- a/tests/v1/test_e2e.py +++ b/tests/v1/test_e2e.py @@ -263,10 +263,8 @@ async def test_acp_resume_with_tool(run_v1, harness, harness_runtime, tmp_path): assert len(segments) == 2 assert segments[0]["terminated"] is False assert segments[1]["terminated"] is False - # Kimi Code drops message `phase` on replay. Prime Agent's Chat Completions - # replay omits provider-only response state; the interception server retains - # those model calls as separate branches instead of guessing a false lineage. - if harness.id not in ("kimi-code", "prime-agent"): + # Kimi Code is broken upstream: its Responses adapter drops message `phase` on replay. + if harness.id != "kimi-code": assert trace.num_branches == 1 # Native MCP tools need not appear in the intercepted model request that # populates trace.tools; the ACP transcript is the source of truth for use.