feat: add optional ACP semantic edges

pyproject.toml

@@ -41,7 +41,7 @@ dependencies = [
     "pyzmq>=27.1.0",
     "msgpack>=1.1.2",
     "httpx>=0.27.0",
-    "aiohttp>=3.14.0",
+    "aiohttp>=3.14.1",
     "prime-pydantic-config[toml]>=0.4.3",
     "uvloop>=0.21.0; sys_platform != 'win32' and sys_platform != 'cygwin' and platform_python_implementation != 'PyPy'",
     "loguru>=0.7.0",

tests/v1/test_e2e.py

@@ -273,6 +273,12 @@ async def test_acp_resume_with_tool(run_v1, harness, harness_runtime, tmp_path):
     assert segments[1]["tool_outputs"]
     if harness.id == "rlm":
         assert "turns_since_last_compaction" in trace.metrics
+        assert all(call.acp is not None for call in trace.calls)
+        assert all(
+            trace.nodes[parent.node].sampled and node.sampled
+            for node in trace.nodes
+            for parent in node.semantic_parents
+        )
     if harness.id == "prime-agent":
         lifecycle = trace.info["acp_lifecycle"]["ai.primeintellect.prime-agent"]
         assert len(lifecycle) == 2

tests/v1/test_trace.py

@@ -11,7 +11,17 @@
 import verifiers.v1 as vf
 from verifiers.v1.agent import Interaction
 from verifiers.v1.graph import MessageNode
+from verifiers.v1.harnesses.rlm.harness import (
+    RLM_SESSION_METADATA_KEY,
+    RLMHarness,
+    RLMHarnessConfig,
+)
 from verifiers.v1.rollout import Rollout, RolloutTimeouts
+from verifiers.v1.semantic import (
+    ACP_EXTENSION_HEADERS,
+    ACP_SEMANTIC_EDGES_METADATA_KEY,
+    extract_acp_info,
+)
 from verifiers.v1.types import AssistantMessage, UserMessage
 
 
@@ -170,3 +180,301 @@ 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 _semantic_edge_set() -> vf.SemanticEdgeSet:
+    return vf.SemanticEdgeSet(
+        edges=[
+            vf.SemanticEdge(
+                source_request_id="root-turn",
+                target_request_id="root-compact",
+                type="continuation",
+            ),
+            vf.SemanticEdge(
+                source_request_id="root-turn",
+                target_request_id="child-turn",
+                type="subagent_call",
+            ),
+            vf.SemanticEdge(
+                source_request_id="child-turn",
+                target_request_id="root-after",
+                type="subagent_return",
+            ),
+            vf.SemanticEdge(
+                source_request_id="root-compact",
+                target_request_id="root-after",
+                type="compaction",
+            ),
+            vf.SemanticEdge(
+                source_request_id="root-turn",
+                target_request_id="root-after",
+                type="critic_review",
+            ),
+        ],
+    )
+
+
+def test_semantic_edges_resolve_to_message_nodes_and_round_trip():
+    """Request edges resolve by exact IDs, not call adjacency or graph shape."""
+    tr = vf.Trace(
+        agent=vf.AgentInfo(config=vf.AgentConfig()),
+        task=vf.TraceTask(type="Task", data=vf.TaskData(idx=0, prompt="q")),
+        nodes=[
+            MessageNode(parent=None, message=UserMessage(content="root")),
+            MessageNode(
+                parent=0, message=AssistantMessage(content="root turn"), sampled=True
+            ),
+            MessageNode(parent=None, message=UserMessage(content="child")),
+            MessageNode(
+                parent=2, message=AssistantMessage(content="child turn"), sampled=True
+            ),
+            MessageNode(parent=None, message=UserMessage(content="summarize")),
+            MessageNode(
+                parent=4, message=AssistantMessage(content="summary"), sampled=True
+            ),
+            MessageNode(parent=None, message=UserMessage(content="resume")),
+            MessageNode(
+                parent=6, message=AssistantMessage(content="done"), sampled=True
+            ),
+        ],
+    )
+    tr.calls = [
+        vf.ModelCall(
+            node=1,
+            acp=vf.ACPInfo(request_id="root-turn"),
+        ),
+        vf.ModelCall(
+            node=3,
+            acp=vf.ACPInfo(request_id="child-turn"),
+        ),
+        vf.ModelCall(
+            node=5,
+            acp=vf.ACPInfo(request_id="root-compact"),
+        ),
+        vf.ModelCall(
+            node=7,
+            acp=vf.ACPInfo(request_id="root-after"),
+        ),
+    ]
+
+    edge_set = _semantic_edge_set()
+    tr.add_semantic_edges(vf.SemanticEdgeSet(edges=edge_set.edges[:2]))
+    first_semantic_parents = tr.nodes[3].semantic_parents
+    tr.add_semantic_edges(vf.SemanticEdgeSet.model_validate(edge_set.model_dump()))
+    expected_parents = [
+        [],
+        [],
+        [],
+        [vf.ParentLink(node=1, type="subagent_call")],
+        [],
+        [vf.ParentLink(node=1, type="continuation")],
+        [],
+        [
+            vf.ParentLink(node=3, type="subagent_return"),
+            vf.ParentLink(node=5, type="compaction"),
+            vf.ParentLink(node=1, type="critic_review"),
+        ],
+    ]
+    assert [node.semantic_parents for node in tr.nodes] == expected_parents
+    assert tr.nodes[3].semantic_parents is first_semantic_parents
+
+    restored = vf.WireTrace.model_validate_json(tr.model_dump_json())
+    assert [node.semantic_parents for node in restored.nodes] == expected_parents
+    assert [call.acp for call in restored.calls] == [call.acp for call in tr.calls]
+
+    # The base ACP layer resolves the generic edge set before harness-owned metadata.
+    harness = RLMHarness(RLMHarnessConfig(id="rlm"))
+    turn_metadata = {
+        ACP_SEMANTIC_EDGES_METADATA_KEY: _semantic_edge_set().model_dump(mode="json"),
+        RLM_SESSION_METADATA_KEY: {
+            "session_id": restored.id,
+            "metrics": {"turns": 4},
+        },
+    }
+    harness._consume_protocol_metadata(restored, turn_metadata)
+    harness.acp_turn_result(
+        restored, vf.ACPTurn(reply="done", response_metadata=turn_metadata)
+    )
+    assert restored.metrics["turns"] == 4
+    assert [node.semantic_parents for node in restored.nodes] == expected_parents
+
+    # session/close may publish the same cumulative edge set again.
+    close_metadata = {
+        ACP_SEMANTIC_EDGES_METADATA_KEY: _semantic_edge_set().model_dump(mode="json"),
+        RLM_SESSION_METADATA_KEY: {
+            "session_id": restored.id,
+            "metrics": {"turns": 4},
+        },
+    }
+    harness._consume_protocol_metadata(restored, close_metadata)
+    harness.acp_close_result(restored, close_metadata)
+    assert restored.metrics["turns"] == 4
+    assert [node.semantic_parents for node in restored.nodes] == expected_parents
+
+    # A failed provider exchange and its SDK retry share one logical request ID.
+    restored.calls.append(
+        vf.ModelCall(
+            acp=restored.calls[0].acp,
+            error=vf.Error(type="E", message="x"),
+        )
+    )
+    restored.add_semantic_edges(_semantic_edge_set())
+    assert [node.semantic_parents for node in restored.nodes] == expected_parents
+
+
+def test_semantic_edge_uses_last_committed_retry_node():
+    tr = vf.Trace(
+        agent=vf.AgentInfo(config=vf.AgentConfig()),
+        task=vf.TraceTask(type="Task", data=vf.TaskData(idx=0, prompt="q")),
+        nodes=[
+            MessageNode(parent=None, message=UserMessage(content="root")),
+            MessageNode(
+                parent=0, message=AssistantMessage(content="attempt 1"), sampled=True
+            ),
+            MessageNode(
+                parent=0, message=AssistantMessage(content="attempt 2"), sampled=True
+            ),
+            MessageNode(
+                parent=None, message=AssistantMessage(content="next"), sampled=True
+            ),
+        ],
+        calls=[
+            vf.ModelCall(node=1, acp=vf.ACPInfo(request_id="retried")),
+            vf.ModelCall(node=2, acp=vf.ACPInfo(request_id="retried")),
+            vf.ModelCall(node=3, acp=vf.ACPInfo(request_id="next")),
+        ],
+    )
+
+    tr.add_semantic_edges(
+        vf.SemanticEdgeSet(
+            edges=[
+                vf.SemanticEdge(
+                    source_request_id="retried",
+                    target_request_id="next",
+                    type="continuation",
+                )
+            ]
+        )
+    )
+
+    assert tr.nodes[3].semantic_parents == [vf.ParentLink(node=2, type="continuation")]
+
+
+def test_semantic_edge_cycle_is_rejected_without_partial_mutation():
+    tr = vf.Trace(
+        agent=vf.AgentInfo(config=vf.AgentConfig()),
+        task=vf.TraceTask(type="Task", data=vf.TaskData(idx=0, prompt="q")),
+        nodes=[
+            MessageNode(parent=None, message=UserMessage(content="start")),
+            MessageNode(
+                parent=0, message=AssistantMessage(content="first"), sampled=True
+            ),
+            MessageNode(parent=1, message=UserMessage(content="continue")),
+            MessageNode(
+                parent=2, message=AssistantMessage(content="second"), sampled=True
+            ),
+        ],
+        calls=[
+            vf.ModelCall(node=1, acp=vf.ACPInfo(request_id="first")),
+            vf.ModelCall(node=3, acp=vf.ACPInfo(request_id="second")),
+        ],
+    )
+
+    with pytest.raises(ValueError, match="cycle in the message graph"):
+        tr.add_semantic_edges(
+            vf.SemanticEdgeSet(
+                edges=[
+                    vf.SemanticEdge(
+                        source_request_id="second",
+                        target_request_id="first",
+                        type="custom",
+                    )
+                ]
+            )
+        )
+
+    assert all(not node.semantic_parents for node in tr.nodes)
+
+
+def test_acp_info_is_validated_and_stripped():
+    headers = {
+        "Authorization": "Bearer local",
+        "Idempotency-Key": "provider-key",
+        "X-ACP-Model-Request-ID": "request-1",
+        "OpenAI-Beta": "feature",
+    }
+    acp, forwarded = extract_acp_info(headers)
+    assert acp == vf.ACPInfo(request_id="request-1")
+    assert not ACP_EXTENSION_HEADERS.intersection(map(str.lower, forwarded))
+    assert forwarded["Idempotency-Key"] == "provider-key"
+    assert forwarded["OpenAI-Beta"] == "feature"
+
+    absent, unchanged = extract_acp_info({"OpenAI-Beta": "feature"})
+    assert absent is None and unchanged == {"OpenAI-Beta": "feature"}
+
+    with pytest.raises(ValueError, match="not a valid ACP request ID"):
+        extract_acp_info({"X-ACP-Model-Request-ID": "not/a/valid/id"})
+
+
+def test_acp_semantic_edge_metadata_is_optional():
+    trace = vf.Trace(
+        agent=vf.AgentInfo(config=vf.AgentConfig()),
+        task=vf.TraceTask(type="Task", data=vf.TaskData(idx=0, prompt="q")),
+    )
+    harness = RLMHarness(RLMHarnessConfig(id="rlm"))
+
+    harness._consume_protocol_metadata(trace, {})
+
+    assert all(not node.semantic_parents for node in trace.nodes)
+
+    harness._consume_protocol_metadata(
+        trace, {ACP_SEMANTIC_EDGES_METADATA_KEY: {"edges": []}}
+    )
+
+    assert all(not node.semantic_parents for node in trace.nodes)
+
+
+def test_semantic_edge_set_rejects_duplicate_self_and_cyclic_edges():
+    edge_set = _semantic_edge_set().model_dump(mode="json")
+    edge_set["edges"].append(edge_set["edges"][0])
+    with pytest.raises(ValueError, match="duplicate semantic edge"):
+        vf.SemanticEdgeSet.model_validate(edge_set)
+
+    with pytest.raises(ValueError, match="cannot link a request to itself"):
+        vf.SemanticEdgeSet.model_validate(
+            {
+                "edges": [
+                    {
+                        "source_request_id": "request-1",
+                        "target_request_id": "request-1",
+                        "type": "custom",
+                    }
+                ]
+            }
+        )
+
+    edge_set = _semantic_edge_set().model_dump(mode="json")
+    edge_set["edges"].append(
+        {
+            "source_request_id": "root-after",
+            "target_request_id": "root-turn",
+            "type": "custom",
+        }
+    )
+    with pytest.raises(ValueError, match="semantic edge cycle"):
+        vf.SemanticEdgeSet.model_validate(edge_set)
+
+
+def test_semantic_edge_set_accepts_deep_acyclic_chain():
+    edge_set = vf.SemanticEdgeSet(
+        edges=[
+            vf.SemanticEdge(
+                source_request_id=f"request-{index}",
+                target_request_id=f"request-{index + 1}",
+                type="continuation",
+            )
+            for index in range(2_000)
+        ]
+    )
+
+    assert len(edge_set.edges) == 2_000

verifiers/v1/__init__.py

@@ -82,6 +82,13 @@
     RuntimeProcess,
     SubprocessConfig,
 )
+from verifiers.v1.semantic import (
+    ACP_SEMANTIC_EDGES_METADATA_KEY,
+    ACPInfo,
+    ParentLink,
+    SemanticEdge,
+    SemanticEdgeSet,
+)
 from verifiers.v1.state import State, StateT
 from verifiers.v1.task import Task, TaskData, TaskResources, TaskTimeout, WireTaskData
 from verifiers.v1.taskset import Taskset
@@ -226,6 +233,11 @@
     "EvalRunInfo",
     "EvalWorkInfo",
     "ModelCall",
+    "ACPInfo",
+    "ACP_SEMANTIC_EDGES_METADATA_KEY",
+    "ParentLink",
+    "SemanticEdge",
+    "SemanticEdgeSet",
     "PolicyEvent",
     "TrainRunInfo",
     "TrainWorkInfo",