diff --git a/tests/v1/test_e2e.py b/tests/v1/test_e2e.py index cef9bf7e1..2c9b3f82a 100644 --- a/tests/v1/test_e2e.py +++ b/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 diff --git a/tests/v1/test_trace.py b/tests/v1/test_trace.py index 69d5895a3..a586b99eb 100644 --- a/tests/v1/test_trace.py +++ b/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(): diff --git a/verifiers/v1/__init__.py b/verifiers/v1/__init__.py index 222fd56f0..65e0ffb8c 100644 --- a/verifiers/v1/__init__.py +++ b/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", diff --git a/verifiers/v1/harnesses/rlm/harness.py b/verifiers/v1/harnesses/rlm/harness.py index 1c51c2f85..7cdd22c4b 100644 --- a/verifiers/v1/harnesses/rlm/harness.py +++ b/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 diff --git a/verifiers/v1/interception/server.py b/verifiers/v1/interception/server.py index 2f5ff4fc0..af3b321e6 100644 --- a/verifiers/v1/interception/server.py +++ b/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( diff --git a/verifiers/v1/lineage.py b/verifiers/v1/lineage.py index 97646550b..c89141d1a 100644 --- a/verifiers/v1/lineage.py +++ b/verifiers/v1/lineage.py @@ -8,6 +8,7 @@ from __future__ import annotations +import re from collections.abc import Mapping from typing import Literal @@ -17,27 +18,8 @@ ACP_LINEAGE_METADATA_KEY = "ai.prime.acp/lineage-v1" ACP_REQUEST_ID_HEADER = "X-ACP-Lineage-Request-ID" -ACP_SESSION_ID_HEADER = "X-ACP-Lineage-Session-ID" -ACP_PARENT_SESSION_ID_HEADER = "X-ACP-Lineage-Parent-Session-ID" -ACP_CONTEXT_ID_HEADER = "X-ACP-Lineage-Context-ID" -ACP_PREVIOUS_CONTEXT_ID_HEADER = "X-ACP-Lineage-Previous-Context-ID" -ACP_TRANSITION_HEADER = "X-ACP-Lineage-Transition" -ACP_COMPACTION_ID_HEADER = "X-ACP-Lineage-Compaction-ID" -ACP_DEPTH_HEADER = "X-ACP-Lineage-Depth" -IDEMPOTENCY_KEY_HEADER = "Idempotency-Key" - -ACP_LINEAGE_HEADERS = frozenset( - { - ACP_REQUEST_ID_HEADER.lower(), - ACP_SESSION_ID_HEADER.lower(), - ACP_PARENT_SESSION_ID_HEADER.lower(), - ACP_CONTEXT_ID_HEADER.lower(), - ACP_PREVIOUS_CONTEXT_ID_HEADER.lower(), - ACP_TRANSITION_HEADER.lower(), - ACP_COMPACTION_ID_HEADER.lower(), - ACP_DEPTH_HEADER.lower(), - } -) + +ACP_LINEAGE_HEADERS = frozenset({ACP_REQUEST_ID_HEADER.lower()}) """Private ACP extension headers consumed at interception and never sent upstream.""" LineageTransition = Literal["root", "spawn", "compact"] @@ -50,37 +32,6 @@ class _StrictLineageModel(BaseModel): model_config = ConfigDict(extra="forbid", strict=True) -class CallLineage(_StrictLineageModel): - """Provenance copied from one intercepted model request.""" - - request_id: str = Field(pattern=LINEAGE_ID_PATTERN) - session_id: str = Field(pattern=LINEAGE_ID_PATTERN) - parent_session_id: str | None = Field(default=None, pattern=LINEAGE_ID_PATTERN) - context_id: str = Field(pattern=LINEAGE_ID_PATTERN) - previous_context_id: str | None = Field(default=None, pattern=LINEAGE_ID_PATTERN) - transition: LineageTransition - compaction_id: str | None = Field(default=None, pattern=LINEAGE_ID_PATTERN) - depth: int = Field(ge=0) - - @model_validator(mode="after") - def validate_transition(self) -> CallLineage: - if self.transition == "root": - if self.parent_session_id is not None: - raise ValueError("a root context cannot have a parent session") - if self.previous_context_id is not None: - raise ValueError("a root context cannot have a previous context") - elif self.transition == "spawn": - if self.parent_session_id is None: - raise ValueError("a spawned context requires a parent session") - if self.previous_context_id is not None: - raise ValueError("a spawned context cannot have a previous context") - elif self.previous_context_id is None: - raise ValueError("a compacted context requires a previous context") - if self.transition == "compact" and self.compaction_id is None: - raise ValueError("a compacted context requires a compaction id") - return self - - class LineageSession(_StrictLineageModel): session_id: str = Field(pattern=LINEAGE_ID_PATTERN) parent_session_id: str | None = Field(default=None, pattern=LINEAGE_ID_PATTERN) @@ -340,14 +291,13 @@ def validate_references(self) -> LineageManifest: return self -def extract_call_lineage( +def extract_lineage_request_id( headers: Mapping[str, str], -) -> tuple[CallLineage | None, dict[str, str]]: - """Parse and remove private ACP lineage headers from a provider-bound request. +) -> tuple[str | None, dict[str, str]]: + """Parse and remove the ACP lineage correlation ID from a provider request. - A partial lineage envelope is rejected: silently accepting it would turn an exact - provenance channel into a heuristic one. Ordinary requests with none of the private - headers remain backward-compatible and carry no lineage. + The full execution graph arrives independently in ACP metadata. Ordinary requests + without the private correlation header remain unchanged. """ forwarded = { @@ -356,50 +306,9 @@ def extract_call_lineage( if name.lower() not in ACP_LINEAGE_HEADERS } normalized = {name.lower(): value for name, value in headers.items()} - values = { - name: value for name, value in normalized.items() if name in ACP_LINEAGE_HEADERS - } - if not values: + request_id = normalized.get(ACP_REQUEST_ID_HEADER.lower()) + if request_id is None: return None, forwarded - - def get(name: str) -> str | None: - value = values.get(name.lower()) - return value if value not in (None, "") else None - - required = ( - ACP_REQUEST_ID_HEADER, - ACP_SESSION_ID_HEADER, - ACP_CONTEXT_ID_HEADER, - ACP_TRANSITION_HEADER, - ACP_DEPTH_HEADER, - ) - missing = [name for name in required if get(name) is None] - if missing: - raise ValueError( - "incomplete ACP lineage headers: missing " + ", ".join(missing) - ) - request_id = get(ACP_REQUEST_ID_HEADER) - idempotency_key = normalized.get(IDEMPOTENCY_KEY_HEADER.lower()) - if idempotency_key is None: - raise ValueError("ACP lineage requires Idempotency-Key") - if idempotency_key != request_id: - raise ValueError("ACP lineage request id must match Idempotency-Key") - try: - depth = int(get(ACP_DEPTH_HEADER) or "") - except ValueError as error: - raise ValueError( - f"{ACP_DEPTH_HEADER} must be a non-negative integer" - ) from error - return ( - CallLineage( - request_id=request_id, - session_id=get(ACP_SESSION_ID_HEADER), - parent_session_id=get(ACP_PARENT_SESSION_ID_HEADER), - context_id=get(ACP_CONTEXT_ID_HEADER), - previous_context_id=get(ACP_PREVIOUS_CONTEXT_ID_HEADER), - transition=get(ACP_TRANSITION_HEADER), - compaction_id=get(ACP_COMPACTION_ID_HEADER), - depth=depth, - ), - forwarded, - ) + if re.fullmatch(LINEAGE_ID_PATTERN, request_id) is None: + raise ValueError(f"{ACP_REQUEST_ID_HEADER} is not a valid lineage ID") + return request_id, forwarded diff --git a/verifiers/v1/trace.py b/verifiers/v1/trace.py index 280270ed6..2c6ff3bcf 100644 --- a/verifiers/v1/trace.py +++ b/verifiers/v1/trace.py @@ -18,7 +18,7 @@ from verifiers.v1.configs.agent import AgentConfig, WireAgentConfig from verifiers.v1.errors import ProviderError from verifiers.v1.graph import MessageNode -from verifiers.v1.lineage import CallLineage, LineageManifest +from verifiers.v1.lineage import LINEAGE_ID_PATTERN, LineageManifest, LineageRequest from verifiers.v1.runtimes import RuntimeInfo from verifiers.v1.state import State, StateT from verifiers.v1.task import DataT, WireTaskData @@ -172,8 +172,8 @@ 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.""" - lineage: CallLineage | None = None - """Exact recursive-session provenance supplied by the harness, when available.""" + lineage_request_id: str | None = Field(default=None, pattern=LINEAGE_ID_PATTERN) + """Opaque request ID joined to the optional ACP lineage manifest.""" def min_new_input_tokens(calls: Iterable[ModelCall]) -> Iterator[tuple[ModelCall, int]]: @@ -199,6 +199,16 @@ class Branch(BaseModel): mm_token_type_id_map: dict[int, int] = Field(default_factory=dict) """The trace's `mm_token_type_id_map`, carried so `mm_token_type_ids` is self-contained.""" + _lineage_requests: dict[str, LineageRequest] = PrivateAttr(default_factory=dict) + + def _requests_in_call_order(self) -> Iterator[LineageRequest]: + for call in self.calls: + if call.lineage_request_id is None: + continue + request = self._lineage_requests.get(call.lineage_request_id) + if request is not None: + yield request + @property def messages(self) -> Messages: return [n.message for n in self.nodes] @@ -208,9 +218,7 @@ def session_ids(self) -> tuple[str, ...]: """Recursive sessions represented on this physical message path, in call order.""" return tuple( dict.fromkeys( - call.lineage.session_id - for call in self.calls - if call.lineage is not None + request.session_id for request in self._requests_in_call_order() ) ) @@ -219,9 +227,7 @@ def context_ids(self) -> tuple[str, ...]: """Context epochs represented on this physical message path, in call order.""" return tuple( dict.fromkeys( - call.lineage.context_id - for call in self.calls - if call.lineage is not None + request.context_id for request in self._requests_in_call_order() ) ) @@ -230,9 +236,9 @@ def compaction_ids(self) -> tuple[str, ...]: """Compactions correlated with calls on this physical message path.""" return tuple( dict.fromkeys( - call.lineage.compaction_id - for call in self.calls - if call.lineage is not None and call.lineage.compaction_id is not None + request.compaction_id + for request in self._requests_in_call_order() + if request.compaction_id is not None ) ) @@ -513,6 +519,11 @@ def usage(self) -> Usage | None: def branches(self) -> list[Branch]: """One root-to-leaf path per graph leaf, its calls attached in path order.""" by_node = {c.node: c for c in self.calls if c.node is not None} + lineage_requests = ( + {request.request_id: request for request in self.lineage.requests} + if self.lineage is not None + else {} + ) branches: list[Branch] = [] for i, leaf in enumerate(graph.leaves(self)): path: list[int] = [] @@ -521,14 +532,14 @@ def branches(self) -> list[Branch]: path.append(nid) nid = self.nodes[nid].parent path.reverse() - branches.append( - Branch( - index=i, - nodes=[self.nodes[n] for n in path], - calls=[by_node[n] for n in path if n in by_node], - mm_token_type_id_map=self.mm_token_type_id_map, - ) + branch = Branch( + index=i, + nodes=[self.nodes[n] for n in path], + calls=[by_node[n] for n in path if n in by_node], + mm_token_type_id_map=self.mm_token_type_id_map, ) + branch._lineage_requests = lineage_requests + branches.append(branch) return branches @property @@ -537,9 +548,15 @@ def calls_by_session(self) -> dict[str, list[ModelCall]]: grouped: dict[str, list[ModelCall]] = {} if self.lineage is not None: grouped = {session.session_id: [] for session in self.lineage.sessions} + requests = { + request.request_id: request for request in self.lineage.requests + } + else: + requests = {} for call in self.calls: - if call.lineage is not None: - grouped.setdefault(call.lineage.session_id, []).append(call) + request = requests.get(call.lineage_request_id or "") + if request is not None: + grouped.setdefault(request.session_id, []).append(call) return grouped @property @@ -554,43 +571,19 @@ def branches_by_session(self) -> dict[str, list[Branch]]: } def reconcile_lineage(self, manifest: LineageManifest) -> None: - """Validate an ACP lineage snapshot against call headers and attach it. + """Join recorded model calls to an ACP lineage snapshot by request ID. - Nothing is inferred from the message graph: every recorded call must have a complete - lineage envelope and a matching request in the snapshot. + Nothing is inferred from the message graph: every recorded call must carry an + opaque correlation ID naming a request in the self-validating manifest. """ - sessions = {session.session_id: session for session in manifest.sessions} - contexts = {context.context_id: context for context in manifest.contexts} requests = {request.request_id: request for request in manifest.requests} for index, call in enumerate(self.calls): - item = call.lineage - if item is None: - raise ValueError(f"model call {index} has no ACP lineage headers") - session = sessions.get(item.session_id) - context = contexts.get(item.context_id) - request = requests.get(item.request_id) - if session is None or ( - session.parent_session_id != item.parent_session_id - or session.depth != item.depth - ): - raise ValueError( - f"model call {item.request_id!r} does not match its lineage session" - ) - if context is None or ( - context.session_id != item.session_id - or context.previous_context_id != item.previous_context_id - or context.transition != item.transition - ): - raise ValueError( - f"model call {item.request_id!r} does not match its lineage context" - ) - if request is None or ( - request.session_id != item.session_id - or request.context_id != item.context_id - or request.compaction_id != item.compaction_id - ): + request_id = call.lineage_request_id + if request_id is None: + raise ValueError(f"model call {index} has no ACP lineage request ID") + if request_id not in requests: raise ValueError( - f"model call {item.request_id!r} does not match its lineage request" + f"model call {request_id!r} has no matching lineage request" ) self.lineage = manifest