refactor: join lineage by request ID

tests/v1/test_e2e.py

@@ -273,6 +273,11 @@ 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
+        lineage = trace.lineage
+        assert lineage is not None
+        request_ids = {request.request_id for request in lineage.requests}
+        assert all(call.lineage_request_id in request_ids for call in trace.calls)
+        assert sum(map(len, trace.calls_by_session.values())) == len(trace.calls)
     if harness.id == "prime-agent":
         lifecycle = trace.info["acp_lifecycle"]["ai.primeintellect.prime-agent"]
         assert len(lifecycle) == 2

tests/v1/test_trace.py

@@ -19,7 +19,7 @@
 from verifiers.v1.lineage import (
     ACP_LINEAGE_HEADERS,
     ACP_LINEAGE_METADATA_KEY,
-    extract_call_lineage,
+    extract_lineage_request_id,
 )
 from verifiers.v1.rollout import Rollout, RolloutTimeouts
 from verifiers.v1.types import AssistantMessage, UserMessage
@@ -285,47 +285,19 @@ def test_exact_lineage_groups_interleaved_calls_and_round_trips():
     tr.calls = [
         vf.ModelCall(
             node=1,
-            lineage=vf.CallLineage(
-                request_id="root-turn",
-                session_id=tr.id,
-                context_id="ctx-root",
-                transition="root",
-                depth=0,
-            ),
+            lineage_request_id="root-turn",
         ),
         vf.ModelCall(
             node=3,
-            lineage=vf.CallLineage(
-                request_id="child-turn",
-                session_id="child",
-                parent_session_id=tr.id,
-                context_id="ctx-child",
-                transition="spawn",
-                depth=1,
-            ),
+            lineage_request_id="child-turn",
         ),
         vf.ModelCall(
             node=5,
-            lineage=vf.CallLineage(
-                request_id="root-compact",
-                session_id=tr.id,
-                context_id="ctx-root",
-                transition="root",
-                compaction_id="compact-1",
-                depth=0,
-            ),
+            lineage_request_id="root-compact",
         ),
         vf.ModelCall(
             node=7,
-            lineage=vf.CallLineage(
-                request_id="root-after",
-                session_id=tr.id,
-                context_id="ctx-root-2",
-                previous_context_id="ctx-root",
-                transition="compact",
-                compaction_id="compact-1",
-                depth=0,
-            ),
+            lineage_request_id="root-after",
         ),
     ]
 
@@ -350,12 +322,12 @@ def test_exact_lineage_groups_interleaved_calls_and_round_trips():
     tr.reconcile_lineage(vf.LineageManifest.model_validate(manifest.model_dump()))
     calls_by_session = tr.calls_by_session
     assert list(calls_by_session) == [tr.id, "child", "idle-child"]
-    assert [call.lineage.request_id for call in calls_by_session[tr.id]] == [
+    assert [call.lineage_request_id for call in calls_by_session[tr.id]] == [
         "root-turn",
         "root-compact",
         "root-after",
     ]
-    assert [call.lineage.request_id for call in calls_by_session["child"]] == [
+    assert [call.lineage_request_id for call in calls_by_session["child"]] == [
         "child-turn"
     ]
     assert calls_by_session["idle-child"] == []
@@ -368,8 +340,8 @@ def test_exact_lineage_groups_interleaved_calls_and_round_trips():
 
     restored = vf.WireTrace.model_validate_json(tr.model_dump_json())
     assert restored.lineage == tr.lineage
-    assert [call.lineage for call in restored.calls] == [
-        call.lineage for call in tr.calls
+    assert [call.lineage_request_id for call in restored.calls] == [
+        call.lineage_request_id for call in tr.calls
     ]
     assert list(restored.calls_by_session) == [tr.id, "child", "idle-child"]
 
@@ -409,47 +381,31 @@ def test_exact_lineage_groups_interleaved_calls_and_round_trips():
     # A failed provider exchange and its SDK retry share one logical request ID.
     restored.calls.append(
         vf.ModelCall(
-            lineage=restored.calls[0].lineage, error=vf.Error(type="E", message="x")
+            lineage_request_id=restored.calls[0].lineage_request_id,
+            error=vf.Error(type="E", message="x"),
         )
     )
     restored.reconcile_lineage(_lineage_manifest(restored.id))
 
 
-def test_lineage_headers_are_complete_validated_and_stripped():
+def test_lineage_request_id_is_validated_and_stripped():
     headers = {
         "Authorization": "Bearer local",
-        "Idempotency-Key": "request-1",
+        "Idempotency-Key": "provider-key",
         "X-ACP-Lineage-Request-ID": "request-1",
-        "X-ACP-Lineage-Session-ID": "session-1",
-        "X-ACP-Lineage-Context-ID": "context-1",
-        "X-ACP-Lineage-Transition": "root",
-        "X-ACP-Lineage-Depth": "0",
         "OpenAI-Beta": "feature",
     }
-    lineage, forwarded = extract_call_lineage(headers)
-    assert lineage == vf.CallLineage(
-        request_id="request-1",
-        session_id="session-1",
-        context_id="context-1",
-        transition="root",
-        depth=0,
-    )
+    request_id, forwarded = extract_lineage_request_id(headers)
+    assert request_id == "request-1"
     assert not ACP_LINEAGE_HEADERS.intersection(map(str.lower, forwarded))
+    assert forwarded["Idempotency-Key"] == "provider-key"
     assert forwarded["OpenAI-Beta"] == "feature"
 
-    absent, unchanged = extract_call_lineage({"OpenAI-Beta": "feature"})
+    absent, unchanged = extract_lineage_request_id({"OpenAI-Beta": "feature"})
     assert absent is None and unchanged == {"OpenAI-Beta": "feature"}
 
-    with pytest.raises(ValueError, match="missing X-ACP-Lineage-Context-ID"):
-        extract_call_lineage(
-            {
-                "Idempotency-Key": "request-1",
-                "X-ACP-Lineage-Request-ID": "request-1",
-                "X-ACP-Lineage-Session-ID": "session-1",
-                "X-ACP-Lineage-Transition": "root",
-                "X-ACP-Lineage-Depth": "0",
-            }
-        )
+    with pytest.raises(ValueError, match="not a valid lineage ID"):
+        extract_lineage_request_id({"X-ACP-Lineage-Request-ID": "not/a/valid/id"})
 
 
 def test_acp_lineage_metadata_is_optional_and_agent_session_ids_are_opaque():

verifiers/v1/__init__.py

@@ -69,7 +69,6 @@
 )
 from verifiers.v1.lineage import (
     ACP_LINEAGE_METADATA_KEY,
-    CallLineage,
     LineageCompaction,
     LineageContext,
     LineageManifest,
@@ -236,7 +235,6 @@
     "EvalWorkInfo",
     "ModelCall",
     "ACP_LINEAGE_METADATA_KEY",
-    "CallLineage",
     "LineageManifest",
     "LineageSession",
     "LineageContext",

verifiers/v1/harnesses/rlm/harness.py

@@ -36,7 +36,7 @@ class _SessionSnapshot(BaseModel):
 
 class RLMHarnessConfig(HarnessConfig):
     version: str = Field(
-        default="48e2a761d4d5b525d90783e006cc76fd24bd11b8", min_length=1
+        default="e26b37a0e8f06d64bff8d7a627ed261be41726a3", min_length=1
     )
     """Git ref (branch, tag, or commit) of nano-rlm to install."""
     max_depth: int = 0

verifiers/v1/interception/server.py

@@ -59,7 +59,7 @@
     TunnelConfig,
     make_tunnel,
 )
-from verifiers.v1.lineage import CallLineage, extract_call_lineage
+from verifiers.v1.lineage import extract_lineage_request_id
 from verifiers.v1.session import IdempotentRequest, ReplayResponse, RolloutSession
 from verifiers.v1.trace import Error, ModelCall, PolicyEvent, TimeSpan
 from verifiers.v1.types import FinishReason, Request, Response, Usage
@@ -423,7 +423,7 @@ def record_call(
         usage: "Usage | None" = None,
         error: BaseException | None = None,
         policy_paths: list[str] | None = None,
-        lineage: CallLineage | None = None,
+        lineage_request_id: str | None = None,
     ) -> None:
         """Append one provider exchange to the trace's per-call records (`Trace.calls`):
         the model + effective settings that went upstream, timing, and — when the call
@@ -471,7 +471,7 @@ def record_call(
                 )
                 if policy_paths
                 else None,
-                lineage=lineage,
+                lineage_request_id=lineage_request_id,
             )
         )
 
@@ -496,7 +496,9 @@ async def handle_request(
         body = dialect.apply_overrides(body, session.ctx.model, session.ctx.sampling)
         streaming = dialect.streaming(body)
         try:
-            lineage, upstream_headers = extract_call_lineage(request.headers)
+            lineage_request_id, upstream_headers = extract_lineage_request_id(
+                request.headers
+            )
         except ValueError as error:
             return web.json_response(dialect.error_body(str(error)), status=400)
         logger.debug(
@@ -515,7 +517,7 @@ async def handle_request(
         replay_key: str | None = None
         binding = (request.path, req_hash)
         if idempotency_key:
-            if streaming and lineage is None:
+            if streaming and lineage_request_id is None:
                 return web.json_response(
                     dialect.error_body(
                         "Idempotency-Key is not supported for streaming requests"
@@ -524,15 +526,6 @@ async def handle_request(
                 )
             if not streaming:
                 replay_key = f"explicit:{idempotency_key}"
-            if lineage is not None:
-                # Lineage uses the key only to bind its logical request id; streaming
-                # replay/coalescing remains unsupported. Ordinary requests keep their
-                # provider-facing key, but this private lineage identity stays local.
-                upstream_headers = {
-                    name: value
-                    for name, value in upstream_headers.items()
-                    if name.lower() != IDEMPOTENCY_KEY_HEADER.lower()
-                }
         elif not streaming:
             replay_key = f"retry:{request.path}:{req_hash.hex()}"
 
@@ -655,7 +648,7 @@ async def coalesced(
                 turn=turn,
                 inspect_response=inspect_response,
                 policy_paths=policy_paths,
-                lineage=lineage,
+                lineage_request_id=lineage_request_id,
                 upstream_headers=upstream_headers,
             )
 
@@ -782,7 +775,7 @@ async def sample() -> web.Response:
                     usage=call_response.usage if call_response else None,
                     error=error,
                     policy_paths=policy_paths,
-                    lineage=lineage,
+                    lineage_request_id=lineage_request_id,
                 )
             return serve(call_response)
 
@@ -799,7 +792,7 @@ async def _stream(
         turn: graph.PendingTurn,
         inspect_response: bool,
         policy_paths: list[str] | None = None,
-        lineage: CallLineage | None = None,
+        lineage_request_id: str | None = None,
         upstream_headers: Mapping[str, str] | None = None,
     ) -> web.StreamResponse:
         """A streamed (SSE) model turn: relay the provider's stream through to the program,
@@ -1059,7 +1052,7 @@ async def _stream(
                 usage=response.usage if response is not None else None,
                 error=error,
                 policy_paths=policy_paths,
-                lineage=lineage,
+                lineage_request_id=lineage_request_id,
             )
 
     async def handle_aux(