feat: preserve harness protocol metadata

tests/v1/test_acp_metadata.py

@@ -0,0 +1,25 @@
+from types import SimpleNamespace
+
+import verifiers.v1 as vf
+from verifiers.v1.acp import ACP_RESPONSE_METADATA_KEY, ACPHarnessSession
+
+
+def test_acp_host_keeps_arbitrary_response_metadata_runtime_only():
+    trace = vf.Trace(
+        agent=vf.AgentInfo(config=vf.AgentConfig()),
+        task=vf.TraceTask(
+            type="Task",
+            data=vf.TaskData(idx=7, prompt="solve it"),
+        ),
+    )
+    receiver = SimpleNamespace(trace=trace)
+
+    ACPHarnessSession._record_metadata(
+        receiver,
+        {"metadata": {"prompt": {"agent.private/example": {"token": "secret"}}}},
+    )
+
+    assert trace._harness_metadata[ACP_RESPONSE_METADATA_KEY]["prompt"] == {
+        "agent.private/example": {"token": "secret"}
+    }
+    assert "secret" not in trace.model_dump_json()

tests/v1/test_e2e.py

@@ -33,12 +33,12 @@ def pair(a: str, b: str, id: str, *extra_marks):
     pair("bash", "modal", "bash-harness-in-modal"),
 ]
 
-# harness x harness runtime for the shell task: every coding agent once (null is a chat
-# loop with no shell), both local runtimes hit (subprocess only carries bash), one
-# remote row per provider.
+# harness x harness runtime for the shell task: every general-purpose coding agent once
+# (null is a chat loop with no shell), both local runtimes hit (subprocess only carries
+# bash), one remote row per provider. RLM's IPython and MCP execution path is covered by
+# the two-segment ACP continuation test, which is deterministic for its tool contract.
 AGENTIC_PLACEMENTS = [
     pair("bash", "subprocess", "bash-harness-in-subprocess"),
-    pair("rlm", "docker", "rlm-harness-in-docker"),
     pytest.param(
         {"id": "kimi-code", "transport": "responses"},
         "docker",

tests/v1/test_trace.py

@@ -104,3 +104,16 @@ def test_wire_trace_round_trip():
 
     # the env-server wire form (a plain model_dump) loads too
     assert vf.WireTrace.model_validate(tr.model_dump()).num_branches == 2
+
+
+def test_model_call_metadata_round_trips_without_affecting_branches():
+    trace = vf.Trace(
+        agent=vf.AgentInfo(config=vf.AgentConfig()),
+        task=vf.TraceTask(type="Task", data=vf.TaskData(idx=0, prompt="q")),
+        calls=[vf.ModelCall(metadata={"example/lineage-v1": {"call_id": "call-1"}})],
+    )
+
+    restored = vf.WireTrace.model_validate(trace.model_dump())
+
+    assert restored.calls[0].metadata == {"example/lineage-v1": {"call_id": "call-1"}}
+    assert restored.num_branches == trace.num_branches == 0

verifiers/v1/acp/__init__.py

@@ -21,8 +21,9 @@
 
 ACP_SOURCE = (Path(__file__).resolve().parent / "runner.py").read_text()
 MAX_PACKET_BYTES = 128 * 1024 * 1024
+ACP_RESPONSE_METADATA_KEY = "ai.verifiers.acp/responses-v1"
 
-__all__ = ["ACPConfig", "ACPHarness"]
+__all__ = ["ACP_RESPONSE_METADATA_KEY", "ACPConfig", "ACPHarness"]
 
 ConfigT = TypeVar("ConfigT", bound=HarnessConfig)
 JsonValue: TypeAlias = (
@@ -47,9 +48,12 @@ class ACPConfig:
 class ACPHarness(Harness[ConfigT]):
     """Harness backed by one live ACP process and native session per rollout."""
 
+    def acp_setup_env(self) -> dict[str, str]:
+        return self.config.resolved_env
+
     async def setup(self, runtime: Runtime) -> None:
         await runtime.prepare_uv_script(
-            ACP_SOURCE, {**self.config.resolved_env, "UV_FROZEN": "false"}
+            ACP_SOURCE, {**self.acp_setup_env(), "UV_FROZEN": "false"}
         )
 
     @abstractmethod
@@ -193,6 +197,17 @@ async def _drain_stderr(self, stream: AsyncIterator[bytes]) -> None:
     def _stderr(self) -> str:
         return self._stderr_tail.decode(errors="replace").strip()
 
+    def _record_metadata(self, response: JsonObject) -> None:
+        metadata = response.get("metadata", {})
+        if not isinstance(metadata, dict):
+            raise TypeError("ACP response metadata must be an object")
+        stored = self.trace._harness_metadata.setdefault(ACP_RESPONSE_METADATA_KEY, {})
+        if not isinstance(stored, dict):
+            raise TypeError(
+                f"trace harness metadata {ACP_RESPONSE_METADATA_KEY!r} must be an object"
+            )
+        stored.update(metadata)
+
     async def _run(self, messages: Messages | None) -> ProgramResult:
         prompt = self.config.prompt if messages is None else messages
         if prompt is None:
@@ -240,6 +255,7 @@ async def _run(self, messages: Messages | None) -> ProgramResult:
             if stderr := self._stderr():
                 detail = f"{detail}\n\nACP process stderr:\n{stderr}"
             raise RuntimeError(detail)
+        self._record_metadata(response)
         reply = response.get("reply", "")
         if not isinstance(reply, str):
             raise TypeError("ACP session reply must be a string")
@@ -257,7 +273,9 @@ async def _stop(self, *, graceful: bool) -> None:
             if graceful and reader is not None:
                 with contextlib.suppress(BaseException):
                     await process.write(_packet({"operation": "shutdown"}))
-                    await asyncio.wait_for(reader.read(), timeout=10)
+                    response = await asyncio.wait_for(reader.read(), timeout=10)
+                    if response.get("ok"):
+                        self._record_metadata(response)
             for timeout, stop in (
                 (10 if graceful else 0.1, None),
                 (5, process.terminate),

verifiers/v1/acp/runner.py

@@ -126,7 +126,7 @@ async def prompt(
     config: dict,
     *,
     is_new: bool,
-) -> str:
+) -> tuple[str, dict]:
     prompt_capabilities = capabilities and capabilities.prompt_capabilities
     supports_images = bool(prompt_capabilities and prompt_capabilities.image)
     blocks = []
@@ -154,7 +154,7 @@ async def prompt(
             "ACP agent produced no visible reply "
             f"(stop_reason={response.stop_reason}, tool_statuses={tool_statuses})"
         )
-    return client.visible_reply
+    return client.visible_reply, dict(response.field_meta or {})
 
 
 class ACPSession:
@@ -170,6 +170,7 @@ def _reset(self) -> None:
         self.capabilities: Any = None
         self.session_id: str | None = None
         self.is_new = True
+        self.start_metadata: dict[str, dict] = {}
 
     async def start(self, config: dict) -> None:
         command = config["command"]
@@ -201,12 +202,16 @@ async def start(self, config: dict) -> None:
             raise
         self.session_id = session.session_id
         self.is_new = True
+        self.start_metadata = {
+            "initialize": dict(initialized.field_meta or {}),
+            "new_session": dict(session.field_meta or {}),
+        }
 
-    async def run(self, config: dict) -> str:
+    async def run(self, config: dict) -> tuple[str, dict[str, dict]]:
         if self.connection is None:
             await self.start(config)
         assert self.session_id is not None
-        reply = await prompt(
+        reply, metadata = await prompt(
             self.client,
             self.connection,
             self.capabilities,
@@ -215,20 +220,30 @@ async def run(self, config: dict) -> str:
             is_new=self.is_new,
         )
         self.is_new = False
-        return reply
+        response_metadata = {**self.start_metadata, "prompt": metadata}
+        self.start_metadata = {}
+        return reply, response_metadata
 
-    async def close(self) -> None:
+    async def close(self) -> dict[str, dict]:
+        metadata: dict[str, dict] = {}
         try:
             if self.connection is not None and self.session_id is not None:
                 session_capabilities = (
                     self.capabilities and self.capabilities.session_capabilities
                 )
                 if session_capabilities and session_capabilities.close is not None:
                     with suppress(Exception):
-                        await self.connection.close_session(session_id=self.session_id)
+                        response = await self.connection.close_session(
+                            session_id=self.session_id
+                        )
+                        field_meta = (
+                            response.field_meta if response is not None else None
+                        )
+                        metadata["close_session"] = dict(field_meta or {})
             await self.stack.aclose()
         finally:
             self._reset()
+        return metadata
 
 
 async def read_packet(stream: asyncio.StreamReader) -> dict | None:
@@ -269,14 +284,16 @@ async def serve_stream() -> None:
             try:
                 operation = request.get("operation")
                 if operation == "prompt":
+                    reply, metadata = await session.run(request["config"])
                     response = {
                         "ok": True,
-                        "reply": await session.run(request["config"]),
+                        "reply": reply,
+                        "metadata": metadata,
                     }
                 elif operation == "shutdown":
-                    await session.close()
+                    metadata = await session.close()
                     stop = True
-                    response = {"ok": True}
+                    response = {"ok": True, "metadata": metadata}
                 else:
                     raise ValueError(f"unknown ACP session operation: {operation!r}")
             except Exception as error:  # noqa: BLE001 - serialize protocol failures

verifiers/v1/trace.py

@@ -193,6 +193,9 @@ class ModelCall(BaseModel):
     """The failure that ended this call, coupled to the exchange that caused it."""
     policy: PolicyEvent | None = None
     """Policy mediation applied to the request before this call."""
+    metadata: dict[str, Any] = Field(default_factory=dict)
+    """Namespaced boundary metadata for this exchange. It is descriptive only; message
+    ancestry in `Trace.nodes` remains authoritative for branch construction."""
 
 
 def min_new_input_tokens(calls: Iterable[ModelCall]) -> Iterator[tuple[ModelCall, int]]:
@@ -405,6 +408,8 @@ class Trace(BaseModel, Generic[DataT, StateT, AgentConfigT]):
 
     _head_index: dict = PrivateAttr(default_factory=dict)
     """`(parent, msg_hash) -> node_id` for the graph builder."""
+    _harness_metadata: dict[str, Any] = PrivateAttr(default_factory=dict)
+    """Runtime-only harness protocol metadata, excluded from trace serialization."""
 
     @property
     def reward(self) -> float: