diff --git a/polylogue/schemas/packages.py b/polylogue/schemas/packages.py index 601623d2fd..849ecfc531 100644 --- a/polylogue/schemas/packages.py +++ b/polylogue/schemas/packages.py @@ -38,6 +38,7 @@ def _string_int_dict(value: object) -> dict[str, int]: SchemaResolutionReason: TypeAlias = Literal[ "bundle_scope", "exact_structure", + "package_catalog", "package_default", "profile_family", ] diff --git a/polylogue/schemas/runtime_registry.py b/polylogue/schemas/runtime_registry.py index 5765740cff..4fde3a0453 100644 --- a/polylogue/schemas/runtime_registry.py +++ b/polylogue/schemas/runtime_registry.py @@ -45,6 +45,7 @@ "exact_structure": 3, "bundle_scope": 2, "profile_family": 1, + "package_catalog": 0, "package_default": 0, } diff --git a/polylogue/schemas/synthetic/wire_formats.py b/polylogue/schemas/synthetic/wire_formats.py index 557fb85343..1a7c804a35 100644 --- a/polylogue/schemas/synthetic/wire_formats.py +++ b/polylogue/schemas/synthetic/wire_formats.py @@ -10,11 +10,13 @@ import hashlib import json import re +from collections import Counter from collections.abc import Collection, Mapping, Sequence from dataclasses import dataclass from typing import TYPE_CHECKING, Any, Literal, Protocol, TypeAlias, cast from polylogue.archive.raw_payload.decode import JSONValue +from polylogue.core.enums import BlockType, MaterialOrigin, MessageType, Role if TYPE_CHECKING: from polylogue.schemas.synthetic.models import SchemaRecord, SyntheticGenerationBatch @@ -22,6 +24,7 @@ WireEncoding: TypeAlias = Literal["json", "jsonl"] WireCapabilityStatus: TypeAlias = Literal["supported", "unsupported"] +WireSupportEntryKey: TypeAlias = tuple[str, str | None, str | None] class UnsupportedSyntheticWireRouteError(ValueError): @@ -142,6 +145,33 @@ def healthy(self) -> bool: ) +def wire_support_key( + provider: str, + package_version: str | None, + element_kind: str | None, +) -> WireSupportEntryKey: + """Return the canonical identity used by persisted wire-support entries.""" + + return (provider, package_version, element_kind) + + +def wire_support_entry_key(entry: WireSupportEntry) -> WireSupportEntryKey: + return wire_support_key(entry.provider, entry.package_version, entry.element_kind) + + +def validate_wire_support_entry_keys( + entries: Sequence[WireSupportEntry], + *, + boundary: str, +) -> None: + seen: set[WireSupportEntryKey] = set() + for entry in entries: + key = wire_support_entry_key(entry) + if key in seen: + raise ValueError(f"{boundary} contains duplicate wire support entry key: {key!r}") + seen.add(key) + + @dataclass(frozen=True) class WireSupportReceipt: """Registry-derived, deterministic support and coverage receipt.""" @@ -149,6 +179,13 @@ class WireSupportReceipt: catalog_providers: tuple[str, ...] entries: tuple[WireSupportEntry, ...] missing_routes: tuple[str, ...] + witness_seed: int = 20260805 + catalog_scope: Literal["registry-default", "explicit"] = "explicit" + + def __post_init__(self) -> None: + validate_wire_support_entry_keys(self.entries, boundary="wire support receipt") + if self.catalog_scope not in {"registry-default", "explicit"}: + raise ValueError(f"wire support receipt has invalid catalog scope: {self.catalog_scope!r}") @property def supported_count(self) -> int: @@ -169,10 +206,12 @@ def complete(self) -> bool: def to_dict(self) -> dict[str, object]: return { "catalog_providers": list(self.catalog_providers), + "catalog_scope": self.catalog_scope, "supported_count": self.supported_count, "unsupported_count": self.unsupported_count, "validated_supported_count": self.validated_supported_count, "missing_routes": list(self.missing_routes), + "witness_seed": self.witness_seed, "complete": self.complete, "entries": [ { @@ -255,6 +294,7 @@ def to_dict(self) -> dict[str, object]: # All catalog providers must have a route here. The unsupported routes # are explicit capabilities, not implicit generator fallbacks. +CATALOG_ELEMENT_UNSUPPORTED_REASON = "catalog element is marked unsupported" PROVIDER_WIRE_ROUTES: dict[str, WireRoute] = { **{ provider: WireRoute(status="supported", wire_format=wire_format) @@ -333,7 +373,12 @@ def _runtime_coverage_path(path: str) -> str: return path -def _route_nonrepresentable_reasons(provider: str, missing_keywords: Collection[str]) -> dict[str, str]: +def _route_nonrepresentable_reasons( + provider: str, + missing_keywords: Collection[str], + *, + package_version: str, +) -> dict[str, str]: """Prove nodes discarded by a provider's wire normalizer are unreachable.""" reasons: dict[str, str] = {} prefixes: tuple[tuple[str, str], ...] @@ -382,8 +427,17 @@ def _route_nonrepresentable_reasons(provider: str, missing_keywords: Collection[ "$.properties.chunkedPrompt.properties.chunks.items[*].properties.text", "$.properties.chunkedPrompt.properties.chunks.items[*].properties.createTime", } + chatgpt_v1_media_prefix = ( + "$.properties.mapping.additionalProperties.*.properties.message.anyOf[1].properties.content." + "properties.parts.items[*].anyOf[1]" + ) for keyword in missing_keywords: path = keyword.split("@", 1)[1] if "@" in keyword else "$" + if provider == "chatgpt" and package_version == "v1" and path.startswith(chatgpt_v1_media_prefix): + reasons[keyword] = ( + "ChatGPT v1 wire shaping discards only the export-only media branch at this exact package selection" + ) + continue if ( provider == "chatgpt" and keyword.startswith("type:null@") @@ -1022,10 +1076,7 @@ def _parser_artifact_evidence( node_texts = tuple(_normalise_evidence_text(text) for text in _payload_string_values(node)) identity_bound = not message_identity or message_identity in node_texts content_bound = normalized in node_texts - structured_content_bound = ( - bool(message.blocks) and bool(message_identity) and identity_bound and bool(node_texts) - ) - if not identity_bound or not (content_bound or structured_content_bound): + if not identity_bound or not content_bound: continue node_bytes = json.dumps(node, sort_keys=True, separators=(",", ":")).encode("utf-8") digest = hashlib.sha256(node_bytes).hexdigest() @@ -1041,7 +1092,529 @@ def _parser_artifact_evidence( return tuple(evidence) -def build_wire_support_receipt(*, registry: object | None = None, seed: int = 20260805) -> WireSupportReceipt: +def _parser_artifact_message_keys( + sessions: Sequence[ParsedSession], +) -> tuple[str, ...]: + """Return the complete conversational identity/content set from parsed output.""" + keys: list[str] = [] + for session in sessions: + for message in session.messages: + if message.provider_message_id: + keys.append(f"id:{message.provider_message_id}") + elif isinstance(message.text, str) and message.text.strip(): + keys.append(f"text:{_normalise_evidence_text(message.text)}") + return tuple(keys) + + +def _parser_artifact_message_semantic_witnesses( + sessions: Sequence[ParsedSession], +) -> tuple[tuple[str, Role, MaterialOrigin], ...]: + """Return parsed identity/text plus authoredness for semantic witnesses.""" + return tuple( + (key, message.role, message.material_origin) + for session in sessions + for message in session.messages + for key in ( + f"id:{message.provider_message_id}" + if message.provider_message_id + else f"text:{_normalise_evidence_text(message.text)}" + if isinstance(message.text, str) and message.text.strip() + else None, + ) + if key is not None + ) + + +def _parser_artifact_expected_nodes(provider: str, payload: JSONValue) -> tuple[Mapping[str, JSONValue], ...]: + """Return the parser-owned raw nodes used for message coverage.""" + native_payload = payload.get("raw_provider_payload") if isinstance(payload, dict) else None + if provider == "claude-ai" and isinstance(native_payload, (dict, list)): + nodes = _parser_evidence_nodes(provider, native_payload) + elif isinstance(payload, dict): + session = payload.get("session") + turns = session.get("turns") if isinstance(session, dict) else None + if isinstance(turns, list): + nodes = tuple( + turn + for turn in turns + if isinstance(turn, dict) + and isinstance(turn.get("provider_turn_id"), str) + and isinstance(turn.get("role"), str) + and isinstance(turn.get("text"), str) + ) + else: + nodes = () + else: + nodes = () + if ( + provider != "claude-ai" + and isinstance(payload, dict) + and isinstance(payload.get("raw_provider_payload"), (dict, list)) + ): + payload = {key: value for key, value in payload.items() if key != "raw_provider_payload"} + return nodes or _parser_evidence_nodes(provider, payload) + + +def _parser_artifact_node_identity(provider: str, node: Mapping[str, JSONValue]) -> str | None: + """Extract the provider message identity from one parser-owned raw node.""" + identity: object = None + if provider == "chatgpt": + message = node.get("message") + if isinstance(message, Mapping): + identity = message.get("id") + if not isinstance(identity, str) or not identity: + identity = node.get("provider_turn_id") + if not isinstance(identity, str) or not identity: + identity = node.get("uuid") + if not isinstance(identity, str) or not identity: + identity = node.get("id") + if not isinstance(identity, str) or not identity: + message = node.get("message") + if isinstance(message, Mapping): + identity = message.get("id") + return identity if isinstance(identity, str) and identity else None + + +def _parser_artifact_node_role(provider: str, node: Mapping[str, JSONValue]) -> Role: + """Normalize the role asserted by one raw parser-owned message node.""" + raw_role: object = None + if provider == "chatgpt": + message = node.get("message") + author = message.get("author") if isinstance(message, Mapping) else None + raw_role = author.get("role") if isinstance(author, Mapping) else None + if not isinstance(raw_role, str) or not raw_role: + raw_role = node.get("role") + elif provider == "claude-ai": + raw_role = node.get("sender") + elif provider == "claude-code": + message = node.get("message") + raw_role = message.get("role") if isinstance(message, Mapping) else node.get("type") + else: + raw_role = node.get("role") + if not isinstance(raw_role, str) or not raw_role: + return Role.UNKNOWN + try: + role = Role.normalize(raw_role) + except ValueError: + return Role.UNKNOWN + if provider in {"claude-ai", "claude-code"} and role is Role.USER: + tool_envelope: Mapping[str, JSONValue] = node + if provider == "claude-code": + nested_message = node.get("message") + if isinstance(nested_message, Mapping): + tool_envelope = nested_message + content = tool_envelope.get("content") + if ( + isinstance(content, list) + and content + and all(isinstance(item, Mapping) and item.get("type") == "tool_result" for item in content) + ): + return Role.TOOL + return role + + +def _parser_artifact_node_tool_witnesses( + provider: str, + node: Mapping[str, JSONValue], +) -> tuple[tuple[BlockType, str, bool | None, int | None], ...]: + """Project load-bearing tool identity/outcome fields from a raw node. + + Synthetic routes exercise Claude's shared content-segment protocol for + both Claude Code and Codex. A Claude Code background-task acknowledgement + deliberately distrusts its immediate ``is_error`` result: the result only + reports that the task started, not its terminal outcome. Match that + production semantic here instead of treating the intentional downgrade as + a parser loss. + """ + if provider not in {"claude-ai", "claude-code", "codex"}: + return () + if provider == "claude-code": + message = node.get("message") + content = message.get("content") if isinstance(message, Mapping) else None + tool_use_result = node.get("toolUseResult") + background_task = ( + isinstance(tool_use_result, Mapping) + and isinstance(tool_use_result.get("backgroundTaskId"), str) + and bool(tool_use_result.get("backgroundTaskId")) + ) + elif provider == "codex": + content = node.get("content") + background_task = False + else: + content = node.get("content") + background_task = False + if not isinstance(content, list): + return () + + witnesses: list[tuple[BlockType, str, bool | None, int | None]] = [] + for item in content: + if not isinstance(item, Mapping): + continue + item_type = item.get("type") + if item_type == "tool_use": + tool_id = item.get("id") + if isinstance(tool_id, str) and tool_id: + witnesses.append((BlockType.TOOL_USE, tool_id, None, None)) + elif item_type == "tool_result": + tool_id = item.get("tool_use_id") + if not isinstance(tool_id, str) or not tool_id: + continue + raw_is_error = item.get("is_error") + raw_exit_code = item.get("exit_code") + is_error = raw_is_error if isinstance(raw_is_error, bool) and not background_task else None + exit_code = ( + raw_exit_code if isinstance(raw_exit_code, int) and not isinstance(raw_exit_code, bool) else None + ) + if background_task: + exit_code = None + witnesses.append((BlockType.TOOL_RESULT, tool_id, is_error, exit_code)) + return tuple(witnesses) + + +def _parser_artifact_node_message_type( + provider: str, + node: Mapping[str, JSONValue], +) -> MessageType: + """Derive the raw node's message type using shared structural semantics.""" + from polylogue.archive.message.artifacts import classify_block_message_type, classify_text_message_type + + if provider == "claude-code": + return MessageType.CONTEXT if node.get("isMeta") else MessageType.MESSAGE + if provider == "codex": + raw_role = node.get("role") + return MessageType.CONTEXT if raw_role in {"system", "developer"} else MessageType.MESSAGE + if provider == "chatgpt" and _parser_artifact_node_role(provider, node) is Role.TOOL: + return MessageType.TOOL_RESULT + block_types = tuple(witness[0] for witness in _parser_artifact_node_tool_witnesses(provider, node)) + if block_message_type := classify_block_message_type(block_types): + return block_message_type + text = "\n".join(_parser_artifact_node_content_texts(provider, node)) + return classify_text_message_type(text) or MessageType.MESSAGE + + +def _parser_artifact_node_material_origin( + provider: str, + node: Mapping[str, JSONValue], + role: Role, +) -> MaterialOrigin: + """Derive authoredness from raw role, structure, and documented provenance.""" + from polylogue.archive.message.artifacts import classify_material_origin + from polylogue.sources.parsers.base_support import human_authored_override + + message_type = _parser_artifact_node_message_type(provider, node) + # Codex message records retain their message type from their role/text; + # inline content segments do not reclassify the whole record into a tool + # turn. Its parser makes the same deliberately narrow distinction. + block_types = ( + () + if provider == "codex" + else tuple(witness[0] for witness in _parser_artifact_node_tool_witnesses(provider, node)) + ) + text = "\n".join(_parser_artifact_node_content_texts(provider, node)) + material_origin = classify_material_origin( + role=role, + message_type=message_type, + text=text, + block_types=block_types, + ) + if provider != "claude-code": + return human_authored_override(role, message_type, material_origin) + if material_origin is not MaterialOrigin.UNKNOWN: + return material_origin + message = node.get("message") + content = message.get("content") if isinstance(message, Mapping) else None + has_tool_result = isinstance(content, list) and any( + isinstance(item, Mapping) and item.get("type") == "tool_result" for item in content + ) + if ( + node.get("type") == "user" + and role is Role.USER + and message_type is MessageType.MESSAGE + and not node.get("isMeta") + and not node.get("isCompactSummary") + and node.get("toolUseResult") is None + and not has_tool_result + ): + return MaterialOrigin.HUMAN_AUTHORED + return material_origin + + +def _parser_artifact_expected_message_keys( + provider: str, + payload: JSONValue, +) -> tuple[str, ...]: + """Derive the full conversational identity/content set from one artifact.""" + keys: list[str] = [] + for node in _parser_artifact_expected_nodes(provider, payload): + identity = _parser_artifact_node_identity(provider, node) + if identity is not None: + keys.append(f"id:{identity}") + continue + keys.extend(f"text:{text}" for text in _parser_artifact_node_content_texts(provider, node)) + return tuple(keys) + + +def _parser_artifact_expected_message_semantic_witnesses( + provider: str, + payload: JSONValue, +) -> tuple[tuple[str, Role, MaterialOrigin], ...]: + """Derive identity/text and authoredness from the canonical raw witness.""" + witnesses: list[tuple[str, Role, MaterialOrigin]] = [] + for node in _parser_artifact_expected_nodes(provider, payload): + identity = _parser_artifact_node_identity(provider, node) + role = _parser_artifact_node_role(provider, node) + material_origin = _parser_artifact_node_material_origin(provider, node, role) + if identity is not None: + witnesses.append((f"id:{identity}", role, material_origin)) + continue + witnesses.extend( + (f"text:{text}", role, material_origin) for text in _parser_artifact_node_content_texts(provider, node) + ) + return tuple(witnesses) + + +def _parser_artifact_expected_tool_witnesses( + provider: str, + payload: JSONValue, +) -> tuple[tuple[BlockType, str, bool | None, int | None], ...]: + """Collect every raw structured tool assertion owned by this artifact.""" + return tuple( + witness + for node in _parser_artifact_expected_nodes(provider, payload) + for witness in _parser_artifact_node_tool_witnesses(provider, node) + ) + + +def _parser_artifact_expected_session_id(provider: str, payload: JSONValue, fallback_id: str) -> str | None: + """Return the one provider session identity asserted by this wire artifact.""" + if isinstance(payload, Mapping): + native_payload = payload.get("raw_provider_payload") + if provider == "claude-ai" and isinstance(native_payload, Mapping): + for field in ("uuid", "id", "conversation_id", "conversationId"): + session_id = native_payload.get(field) + if isinstance(session_id, str) and session_id: + return session_id + captured_session = payload.get("session") + captured_session_id = ( + captured_session.get("provider_session_id") if isinstance(captured_session, Mapping) else None + ) + if isinstance(captured_session_id, str) and captured_session_id: + return captured_session_id + if provider == "chatgpt": + for field in ("id", "uuid", "conversation_id", "conversationId"): + session_id = payload.get(field) + if isinstance(session_id, str) and session_id: + return session_id + if provider == "claude-ai": + for field in ("uuid", "id", "conversation_id", "conversationId"): + session_id = payload.get(field) + if isinstance(session_id, str) and session_id: + return session_id + if provider == "claude-code" and isinstance(payload, list): + session_ids = { + session_id + for record in payload + if isinstance(record, Mapping) + for session_id in (record.get("sessionId"),) + if isinstance(session_id, str) and session_id + } + return next(iter(session_ids)) if len(session_ids) == 1 else None + return fallback_id + + +def _parser_artifact_has_expected_session_grouping( + sessions: Sequence[ParsedSession], + provider: str, + payload: JSONValue, + fallback_id: str, +) -> bool: + """Require this single-session artifact to retain its provider session key.""" + expected_session_id = _parser_artifact_expected_session_id(provider, payload, fallback_id) + return ( + expected_session_id is not None + and len(sessions) == 1 + and sessions[0].provider_session_id == expected_session_id + ) + + +def _parser_artifact_tool_witnesses( + sessions: Sequence[ParsedSession], +) -> tuple[tuple[BlockType, str, bool | None, int | None], ...]: + """Collect parsed structured tool identity and provider outcome fields.""" + return tuple( + (block.type, block.tool_id, block.is_error, block.exit_code) + for session in sessions + for message in session.messages + for block in message.blocks + if block.type in {BlockType.TOOL_USE, BlockType.TOOL_RESULT} and block.tool_id + ) + + +def _parser_artifact_messages_have_artifact_bound_content( + sessions: Sequence[ParsedSession], + provider: str, + payload: JSONValue, +) -> bool: + """Require every raw text segment to survive on its identified message.""" + raw_texts_by_identity: dict[str, Counter[str]] = {} + anonymous_raw_texts: Counter[str] = Counter() + for node in _parser_artifact_expected_nodes(provider, payload): + expected_texts = _parser_artifact_node_content_texts(provider, node) + if not expected_texts: + continue + identity = _parser_artifact_node_identity(provider, node) + if identity is None: + anonymous_raw_texts.update(expected_texts) + else: + raw_texts_by_identity.setdefault(identity, Counter()).update(expected_texts) + + observed_anonymous_texts: Counter[str] = Counter() + for session in sessions: + for message in session.messages: + observed_message_text = tuple( + segment + for text in (message.text,) + if isinstance(text, str) + for segment in _parser_artifact_observed_text_segments(text) + ) + observed_block_texts = tuple( + segment + for block in message.blocks + if isinstance(block.text, str) + for segment in _parser_artifact_observed_text_segments(block.text) + ) + expected_segments = raw_texts_by_identity.get(message.provider_message_id) + if expected_segments is not None and not any( + _parser_artifact_text_segments_are_covered(expected_segments, candidate) + for candidate in (observed_message_text, observed_block_texts) + ): + return False + if expected_segments is None: + observed_anonymous_texts.update((*observed_message_text, *observed_block_texts)) + return _parser_artifact_text_segments_are_covered(anonymous_raw_texts, tuple(observed_anonymous_texts.elements())) + + +def _parser_artifact_text_segments_are_covered( + expected: Counter[str], + observed: Sequence[str], +) -> bool: + """Check every raw segment, including repeated segments, in parser-owned text.""" + return all(sum(text.count(segment) for text in observed) >= count for segment, count in expected.items()) + + +def _parser_artifact_observed_text_segments(text: str) -> tuple[str, ...]: + """Expose text embedded in a parser's structured-text serialization.""" + normalized = _normalise_evidence_text(text) + segments = [normalized] if normalized else [] + try: + decoded = json.loads(text) + except json.JSONDecodeError: + return tuple(segments) + if isinstance(decoded, (dict, list)): + segments.extend( + normalized_value + for value in _payload_string_values(decoded) + if (normalized_value := _normalise_evidence_text(value)) + ) + return tuple(segments) + + +def _parser_artifact_node_content_texts(provider: str, node: Mapping[str, JSONValue]) -> tuple[str, ...]: + """Extract text-bearing message content, excluding structured tool arguments.""" + if provider == "chatgpt": + message = node.get("message") + content = message.get("content") if isinstance(message, Mapping) else None + parts = content.get("parts") if isinstance(content, Mapping) else None + values = _payload_string_values(parts) if isinstance(parts, list) else () + elif provider == "codex": + content = node.get("content") + content_values = ( + tuple( + text + for item in content + if isinstance(item, Mapping) and item.get("type") in {"input_text", "output_text", "thinking"} + for text in (item.get("text"), item.get("thinking")) + if isinstance(text, str) + ) + if isinstance(content, list) + else () + ) + message = node.get("message") + values = (*content_values, message) if isinstance(message, str) else content_values + elif provider in {"claude-ai", "gemini"}: + text = node.get("text") + values = (text,) if isinstance(text, str) else () + elif provider == "claude-code": + message = node.get("message") + content = message.get("content") if isinstance(message, Mapping) else None + if isinstance(content, str): + values = (content,) + elif isinstance(content, list): + values = tuple( + text + for item in content + if isinstance(item, Mapping) + for text in ( + (item.get("text"),) + if item.get("type") == "text" + else (item.get("thinking") or item.get("text"),) + if item.get("type") == "thinking" + else _payload_string_values(item.get("content")) + if item.get("type") == "tool_result" + else () + ) + if isinstance(text, str) + ) + else: + values = () + else: + values = () + return tuple(_normalise_evidence_text(value) for value in values if _normalise_evidence_text(value)) + + +def _parser_artifact_has_complete_message_coverage( + sessions: Sequence[ParsedSession], + provider: str, + payload: JSONValue, + fallback_id: str, +) -> bool: + """Require every parser-relevant generated node to survive parsing.""" + expected = _parser_artifact_expected_message_keys(provider, payload) + observed = _parser_artifact_message_keys(sessions) + return ( + bool(expected) + and Counter(expected) == Counter(observed) + and _parser_artifact_has_expected_session_grouping(sessions, provider, payload, fallback_id) + and _parser_artifact_messages_have_artifact_bound_content(sessions, provider, payload) + ) + + +def _parser_artifact_has_complete_semantic_coverage( + sessions: Sequence[ParsedSession], + provider: str, + payload: JSONValue, + fallback_id: str, +) -> bool: + """Require every raw witness to retain authoredness and tool outcomes.""" + expected = _parser_artifact_expected_message_semantic_witnesses(provider, payload) + observed = _parser_artifact_message_semantic_witnesses(sessions) + expected_tools = _parser_artifact_expected_tool_witnesses(provider, payload) + observed_tools = _parser_artifact_tool_witnesses(sessions) + return ( + bool(expected) + and Counter(expected) == Counter(observed) + and Counter(expected_tools) == Counter(observed_tools) + and _parser_artifact_has_expected_session_grouping(sessions, provider, payload, fallback_id) + and _parser_artifact_messages_have_artifact_bound_content(sessions, provider, payload) + ) + + +def build_wire_support_receipt( + *, + registry: object | None = None, + seed: int = 20260805, + providers: Sequence[str] | None = None, +) -> WireSupportReceipt: """Validate executable routes through the selected schema and parser. The provider set comes from the package registry. The parser call is the @@ -1060,218 +1633,280 @@ def build_wire_support_receipt(*, registry: object | None = None, seed: int = 20 from polylogue.schemas.validator import SchemaValidator, ValidationResult from polylogue.sources.dispatch import parse_payload, require_positive_conversational_evidence - catalog_providers = tuple(sorted(registry.list_providers())) # type: ignore[attr-defined] + catalog_providers = tuple( + sorted(dict.fromkeys(registry.list_providers() if providers is None else providers)) # type: ignore[attr-defined] + ) entries: list[WireSupportEntry] = [] missing_routes: list[str] = [] for provider in catalog_providers: route = PROVIDER_WIRE_ROUTES.get(provider) - package = registry.get_package(provider, version="default") # type: ignore[attr-defined] - package_version = package.version if package is not None else None - element_kind = package.default_element_kind if package is not None else None if route is None: missing_routes.append(provider) - entries.append( - WireSupportEntry( - provider=provider, - status="unsupported", - reason="no explicit synthetic wire route", - package_version=package_version, - element_kind=element_kind, - schema_valid=None, - parsed_session_count=0, - parsed_message_count=0, - construct_coverage=None, - validation_error="missing route", - ) + catalog = registry.load_package_catalog(provider) # type: ignore[attr-defined] + selections = tuple( + (package, element) + for package in (catalog.packages if catalog is not None else ()) + for element in package.elements + ) + selections = tuple( + sorted( + selections, + key=lambda item: ( + item[0].version if item[0] is not None else "", + item[1].element_kind if item[1] is not None else "", + ), ) - continue - if route.status == "unsupported": - entries.append( - WireSupportEntry( - provider=provider, - status=route.status, - reason=route.reason, - package_version=package_version, - element_kind=element_kind, - schema_valid=None, - parsed_session_count=0, - parsed_message_count=0, - construct_coverage=None, - ) + ) + if not selections: + package = registry.get_package(provider, version="default") # type: ignore[attr-defined] + selections = ((package, None),) + + for package, element in selections: + package_version = package.version if package is not None else None + element_kind = ( + element.element_kind + if element is not None + else package.default_element_kind + if package is not None + else None ) - continue - - if package is None or route.wire_format is None: - entries.append( - WireSupportEntry( - provider=provider, - status=route.status, - reason=None, - package_version=package_version, - element_kind=element_kind, - schema_valid=False, - parsed_session_count=0, - parsed_message_count=0, - construct_coverage=None, - validation_error="selected package schema is unavailable", + if element is not None and not element.supported: + entries.append( + WireSupportEntry( + provider=provider, + status="unsupported", + reason=CATALOG_ELEMENT_UNSUPPORTED_REASON, + package_version=package_version, + element_kind=element_kind, + schema_valid=None, + parsed_session_count=0, + parsed_message_count=0, + construct_coverage=None, + ) ) - ) - continue - - schema_valid = False - parsed_sessions = [] - payloads: list[JSONValue] = [] - validation_error: str | None = None - selection = None - parser_witnesses: list[WireParserWitness] = [] - parser_errors: list[str] = [] - try: - selection = select_synthetic_schema( - provider, - version="default", - element_kind=element_kind, - registry_factory=cast(Any, lambda: registry), - ) - if selection.package_version != package.version or selection.element_kind != element_kind: - raise ValueError( - "synthetic selection identity diverged from receipt package: " - f"{selection.package_version}/{selection.element_kind} != {package.version}/{element_kind}" + continue + if route is None: + entries.append( + WireSupportEntry( + provider=provider, + status="unsupported", + reason="no explicit synthetic wire route", + package_version=package_version, + element_kind=element_kind, + schema_valid=None, + parsed_session_count=0, + parsed_message_count=0, + construct_coverage=None, + validation_error="missing route", + ) + ) + continue + if route.status == "unsupported": + entries.append( + WireSupportEntry( + provider=provider, + status=route.status, + reason=route.reason, + package_version=package_version, + element_kind=element_kind, + schema_valid=None, + parsed_session_count=0, + parsed_message_count=0, + construct_coverage=None, + ) ) - if selection.wire_format != route.wire_format: - raise ValueError("synthetic selection wire format diverged from declared route") - - corpus = SyntheticCorpus.from_selection(selection) - raw_items = [ - corpus.generate_batch(count=1, messages_per_session=range(4, 5), seed=seed).raw_items[0], - *generate_coverage_witnesses(corpus, seed=seed + 1), - ] - validator = SchemaValidator(selection.schema, strict=False) - validation_results: list[ValidationResult] = [] - schema_resolution = None - if selection.element_kind is not None: - from polylogue.schemas.packages import SchemaResolution - - schema_resolution = SchemaResolution( - provider=selection.provider, - package_version=selection.package_version, - element_kind=selection.element_kind, - exact_structure_id=None, - bundle_scope=None, - reason="package_default", + continue + + if package is None or route.wire_format is None or element_kind is None: + entries.append( + WireSupportEntry( + provider=provider, + status=route.status, + reason=None, + package_version=package_version, + element_kind=element_kind, + schema_valid=False, + parsed_session_count=0, + parsed_message_count=0, + construct_coverage=None, + validation_error="selected package schema is unavailable", + ) ) - for index, raw in enumerate(raw_items): - if route.wire_format.encoding == "jsonl": - payload: JSONValue = [json.loads(line) for line in raw.decode("utf-8").splitlines() if line.strip()] - payload_items = payload if isinstance(payload, list) else [] - else: - payload = json.loads(raw) - payload_items = [payload] if isinstance(payload, (dict, list)) else [] - artifact_results = [validator.validate(item) for item in payload_items] - validation_results.extend(artifact_results) - artifact_validation_error = ( - "; ".join(error for result in artifact_results for error in result.errors) or None + continue + + schema_valid = False + parsed_sessions = [] + payloads: list[JSONValue] = [] + validation_error: str | None = None + selection = None + parser_witnesses: list[WireParserWitness] = [] + parser_errors: list[str] = [] + try: + selection = select_synthetic_schema( + provider, + version=package.version, + element_kind=element_kind, + registry_factory=cast(Any, lambda: registry), ) - artifact_coverage = construct_coverage(selection.schema, payload_items) - parse_error: str | None = None - artifact_sessions = [] - try: - parser_payload = payload - if provider == "chatgpt" and isinstance(payload, dict): - # Validate and account for the complete envelope, but - # keep the optional native subpayload from selecting a - # second schema-shaped tree during parser dispatch. - parser_payload = dict(payload) - parser_payload.pop("raw_provider_payload", None) - parsed_sessions_for_artifact = parse_payload( + if selection.package_version != package.version or selection.element_kind != element_kind: + raise ValueError( + "synthetic selection identity diverged from receipt package: " + f"{selection.package_version}/{selection.element_kind} != {package.version}/{element_kind}" + ) + if selection.wire_format != route.wire_format: + raise ValueError("synthetic selection wire format diverged from declared route") + + corpus = SyntheticCorpus.from_selection(selection) + raw_items = [ + corpus.generate_batch(count=1, messages_per_session=range(4, 5), seed=seed).raw_items[0], + *generate_coverage_witnesses(corpus, seed=seed + 1), + ] + validator = SchemaValidator(selection.schema, strict=False) + validation_results: list[ValidationResult] = [] + schema_resolution = None + if selection.element_kind is not None: + from polylogue.schemas.packages import SchemaResolution + + schema_resolution = SchemaResolution( + provider=selection.provider, + package_version=selection.package_version, + element_kind=selection.element_kind, + exact_structure_id=None, + bundle_scope=None, + reason="package_catalog", + ) + for index, raw in enumerate(raw_items): + if route.wire_format.encoding == "jsonl": + payload: JSONValue = [ + json.loads(line) for line in raw.decode("utf-8").splitlines() if line.strip() + ] + payload_items = payload if isinstance(payload, list) else [] + else: + payload = json.loads(raw) + payload_items = [payload] if isinstance(payload, (dict, list)) else [] + artifact_results = [validator.validate(item) for item in payload_items] + validation_results.extend(artifact_results) + artifact_validation_error = ( + "; ".join(error for result in artifact_results for error in result.errors) or None + ) + artifact_coverage = construct_coverage(selection.schema, payload_items) + parse_error: str | None = None + artifact_sessions = [] + parser_payload: JSONValue = payload + try: + if provider == "chatgpt" and isinstance(payload, dict): + # Validate and account for the complete envelope, but + # keep the optional native subpayload from selecting a + # second schema-shaped tree during parser dispatch. + parser_payload = { + key: value for key, value in payload.items() if key != "raw_provider_payload" + } + parsed_sessions_for_artifact = parse_payload( + provider, + parser_payload, + f"synthetic-wire-receipt:{provider}:{package.version}:{element_kind}:{index}", + schema_resolution=schema_resolution, + ) + artifact_sessions = require_positive_conversational_evidence( + parsed_sessions_for_artifact, + provider=provider, + source_path=f"synthetic-wire-receipt:{provider}:{package.version}:{element_kind}:{index}", + ) + except Exception as exc: # Keep the witness failure in the receipt. + parse_error = f"{type(exc).__name__}: {exc}" + artifact_evidence = _parser_artifact_evidence( + artifact_sessions, provider, - parser_payload, - f"synthetic-wire-receipt:{provider}:{index}", - schema_resolution=schema_resolution, + payload, + f"synthetic-wire-receipt:{provider}:{package.version}:{element_kind}:{index}", ) - artifact_sessions = require_positive_conversational_evidence( - parsed_sessions_for_artifact, - provider=provider, - source_path=f"synthetic-wire-receipt:{provider}:{index}", + coverage_error: str | None = None + fallback_id = f"synthetic-wire-receipt:{provider}:{package.version}:{element_kind}:{index}" + if not _parser_artifact_has_complete_message_coverage( + artifact_sessions, provider, parser_payload, fallback_id + ) or not _parser_artifact_has_complete_semantic_coverage( + artifact_sessions, provider, parser_payload, fallback_id + ): + artifact_evidence = () + coverage_error = "artifact message coverage is incomplete" + parsed_sessions.extend(artifact_sessions) + artifact_kind: Literal["baseline", "coverage"] = "baseline" if index == 0 else "coverage" + parser_witnesses.append( + WireParserWitness( + index=-1 if index == 0 else index - 1, + exercised_keywords=artifact_coverage.exercised_keywords, + parsed_session_count=len(artifact_sessions), + parsed_message_count=sum(len(session.messages) for session in artifact_sessions), + validation_error=parse_error or artifact_validation_error or coverage_error, + artifact_kind=artifact_kind, + artifact_evidence=artifact_evidence, + ) + ) + if ( + artifact_sessions + and artifact_evidence + and artifact_validation_error is None + and parse_error is None + ): + payloads.extend(payload_items) + if parse_error is not None: + label = "baseline" if index == 0 else f"coverage witness {index - 1}" + parser_errors.append(f"{label} parser: {parse_error}") + schema_valid = bool(validation_results) and all(result.is_valid for result in validation_results) + if not schema_valid: + parser_errors.append( + "; ".join(error for result in validation_results for error in result.errors) + or "selected package schema rejected generated payload" ) - except Exception as exc: # Keep the witness failure in the receipt. - parse_error = f"{type(exc).__name__}: {exc}" - artifact_evidence = _parser_artifact_evidence( - artifact_sessions, + except Exception as exc: # Receipt records route failures instead of hiding them. + validation_error = f"{type(exc).__name__}: {exc}" + + if parser_errors: + validation_error = "; ".join((*parser_errors, *((validation_error,) if validation_error else ()))) + + if selection is not None: + witnessed = construct_coverage(selection.schema, payloads) + nonrepresentable_reasons = _route_nonrepresentable_reasons( provider, - payload, - f"synthetic-wire-receipt:{provider}:{index}", + witnessed.missing_keywords, + package_version=package.version, ) - parsed_sessions.extend(artifact_sessions) - artifact_kind: Literal["baseline", "coverage"] = "baseline" if index == 0 else "coverage" - parser_witnesses.append( - WireParserWitness( - index=-1 if index == 0 else index - 1, - exercised_keywords=artifact_coverage.exercised_keywords, - parsed_session_count=len(artifact_sessions), - parsed_message_count=sum(len(session.messages) for session in artifact_sessions), - validation_error=parse_error or artifact_validation_error, - artifact_kind=artifact_kind, - artifact_evidence=artifact_evidence, - ) + coverage = construct_coverage( + selection.schema, + payloads, + nonrepresentable_keywords=nonrepresentable_reasons, + nonrepresentable_reasons=nonrepresentable_reasons, ) - if ( - artifact_sessions - and artifact_evidence - and artifact_validation_error is None - and parse_error is None - ): - payloads.extend(payload_items) - if parse_error is not None: - label = "baseline" if index == 0 else f"coverage witness {index - 1}" - parser_errors.append(f"{label} parser: {parse_error}") - schema_valid = bool(validation_results) and all(result.is_valid for result in validation_results) - if not schema_valid: - parser_errors.append( - "; ".join(error for result in validation_results for error in result.errors) - or "selected package schema rejected generated payload" + else: + coverage = None + entries.append( + WireSupportEntry( + provider=provider, + status=route.status, + reason=None, + package_version=selection.package_version if selection is not None else package_version, + element_kind=selection.element_kind if selection is not None else element_kind, + schema_valid=schema_valid, + parsed_session_count=len(parsed_sessions), + parsed_message_count=sum(len(session.messages) for session in parsed_sessions), + construct_coverage=coverage, + validation_error=validation_error, + parser_witnesses=tuple(parser_witnesses), ) - except Exception as exc: # Receipt records route failures instead of hiding them. - validation_error = f"{type(exc).__name__}: {exc}" - - if parser_errors: - validation_error = "; ".join(parser_errors) - - if selection is not None: - witnessed = construct_coverage(selection.schema, payloads) - nonrepresentable_reasons = _route_nonrepresentable_reasons(provider, witnessed.missing_keywords) - coverage = construct_coverage( - selection.schema, - payloads, - nonrepresentable_keywords=nonrepresentable_reasons, - nonrepresentable_reasons=nonrepresentable_reasons, ) - else: - coverage = None - entries.append( - WireSupportEntry( - provider=provider, - status=route.status, - reason=None, - package_version=selection.package_version if selection is not None else package_version, - element_kind=selection.element_kind if selection is not None else element_kind, - schema_valid=schema_valid, - parsed_session_count=len(parsed_sessions), - parsed_message_count=sum(len(session.messages) for session in parsed_sessions), - construct_coverage=coverage, - validation_error=validation_error, - parser_witnesses=tuple(parser_witnesses), - ) - ) return WireSupportReceipt( catalog_providers=catalog_providers, entries=tuple(entries), missing_routes=tuple(sorted(missing_routes)), + witness_seed=seed, + catalog_scope="registry-default" if providers is None else "explicit", ) __all__ = [ "ConstructCoverage", + "CATALOG_ELEMENT_UNSUPPORTED_REASON", "PROVIDER_WIRE_FORMATS", "PROVIDER_WIRE_CAPABILITIES", "PROVIDER_WIRE_ROUTES", @@ -1282,9 +1917,13 @@ def build_wire_support_receipt(*, registry: object | None = None, seed: int = 20 "WireParserWitness", "WireRoute", "WireSupportEntry", + "WireSupportEntryKey", "WireSupportReceipt", "UnsupportedSyntheticWireRouteError", "build_wire_support_receipt", + "validate_wire_support_entry_keys", + "wire_support_key", + "wire_support_entry_key", "construct_coverage", "generate_coverage_witnesses", ] diff --git a/polylogue/sources/live/batch.py b/polylogue/sources/live/batch.py index 94e1f5d392..b7ed29f790 100644 --- a/polylogue/sources/live/batch.py +++ b/polylogue/sources/live/batch.py @@ -55,6 +55,7 @@ RAW_FAILURE_LIFECYCLE_EVIDENCE_SUPPORT_STATUS_PAIRS, RawFailureEvidenceKind, ) +from polylogue.core.sources import origin_from_provider from polylogue.logging import get_logger from polylogue.pipeline.ids import session_revision_projection from polylogue.pipeline.ingest_outcomes import ( @@ -352,6 +353,67 @@ def _is_json_stream_decode_error(error: BaseException) -> bool: LiveBatchSyncRunner = Callable[..., Awaitable[Any]] P = ParamSpec("P") T = TypeVar("T") + + +@dataclass(frozen=True, slots=True) +class AppendCapabilityReceipt: + """Production append-route capability for one resolved wire selection.""" + + provider: str + package_version: str + element_kind: str + status: Literal["supported", "unsupported"] + reason: str | None + capability_source: str = "LiveBatchProcessor.append" + + def to_dict(self) -> dict[str, str | None]: + return { + "provider": self.provider, + "package_version": self.package_version, + "element_kind": self.element_kind, + "operation": "append_prefix", + "status": self.status, + "reason": self.reason, + "capability_source": self.capability_source, + } + + +_APPEND_CAPABLE_PROVIDER_VALUES = frozenset({Provider.CODEX.value, Provider.CLAUDE_CODE.value}) + + +def append_capability_receipt( + *, + provider: str, + package_version: str, + element_kind: str, + stable_session_identity: bool, +) -> AppendCapabilityReceipt: + """Resolve append support from the live route's identity contract.""" + if provider not in _APPEND_CAPABLE_PROVIDER_VALUES: + return AppendCapabilityReceipt( + provider=provider, + package_version=package_version, + element_kind=element_kind, + status="unsupported", + reason="live append route supports only Codex and Claude Code JSONL identity contracts", + ) + if not stable_session_identity: + return AppendCapabilityReceipt( + provider=provider, + package_version=package_version, + element_kind=element_kind, + status="unsupported", + reason="append delta requires a stable persisted session identity", + ) + return AppendCapabilityReceipt( + provider=provider, + package_version=package_version, + element_kind=element_kind, + status="supported", + reason=None, + ) + + _ARCHIVE_RUNTIME_TIERS = ",".join(spec.tier.value for spec in ARCHIVE_TIER_SPECS.values()) _ARCHIVE_NATIVE_WRITE_TIERS = "source,index" _FULL_CAPTURE_PREFIX_PROOF_ATTEMPTS = 2 @@ -3575,10 +3637,22 @@ def _append_payload_for_provider( NEW writes going forward, per polylogue-u19l's scope. """ provider = Provider.from_string(canonical_acquisition_provider(source_name, source_name=source_name)) - if provider is Provider.CODEX: - identity = self._existing_provider_session_id(path) - if identity is None: + if provider in {Provider.CODEX, Provider.CLAUDE_CODE}: + identity = self._existing_provider_session_id( + path, + expected_origin=origin_from_provider(provider).value, + ) + capability = append_capability_receipt( + provider=provider.value, + package_version="live", + element_kind="session_record_stream", + stable_session_identity=identity is not None, + ) + if capability.status != "supported": return None + else: + identity = None + if provider is Provider.CODEX: # A Codex append-mode delta is the file's tail bytes only -- the # real `session_meta` header that carries native-session identity # was already consumed by an earlier full/append observation and @@ -3597,21 +3671,68 @@ def _append_payload_for_provider( "line and carried as native_id_hint, not spliced into hashed bytes", ) return payload, identity - if provider is Provider.CLAUDE_CODE and not self._claude_code_tail_matches_existing_identity(path, payload): + if provider is Provider.CLAUDE_CODE and not self._claude_code_tail_matches_existing_identity( + path, payload, existing_id=identity + ): return None return payload, None - def _existing_provider_session_id(self, path: Path) -> str | None: - identity = self._existing_archive_session_native_id(path) + def _existing_provider_session_id(self, path: Path, *, expected_origin: str) -> str | None: + identity = self._existing_archive_session_native_id(path, expected_origin=expected_origin) if identity is not None: return identity + if expected_origin != Origin.CODEX_SESSION.value: + return None codex_identity = self._codex_session_meta_native_id(path) if codex_identity is None: return None + # An index-only identity recovery is valid when the source tier has no + # row for this path. It is not valid when the path is already owned by + # another origin: accepting the Codex id from the index in that case + # would turn a mixed-origin path collision into an append match. + if self._source_path_has_conflicting_origin(path, expected_origin=expected_origin): + return None if self._archive_has_native_session("codex-session", codex_identity): return codex_identity return None + def _source_path_has_conflicting_origin(self, path: Path, *, expected_origin: str) -> bool: + """Reject a raw path whose source or joined indexed origin disagrees.""" + archive_root = Path(getattr(self._polylogue, "archive_root", self._cursor._db_path.parent)) + source_db = archive_root / "source.db" + index_db = ArchiveLocation.resolve(archive_root).active_index_path + if not source_db.exists() or not index_db.exists(): + return False + try: + conn = sqlite3.connect(f"file:{index_db}?mode=ro", uri=True) + try: + conn.execute("ATTACH DATABASE ? AS source_tier", (f"file:{source_db}?mode=ro",)) + row = conn.execute( + """ + SELECT 1 + FROM source_tier.raw_sessions AS r + LEFT JOIN sessions AS s ON s.raw_id = r.raw_id + WHERE r.source_path = ? + AND (r.origin <> ? OR (s.session_id IS NOT NULL AND s.origin <> ?)) + LIMIT 1 + """, + (str(path), expected_origin, expected_origin), + ).fetchone() + conn.execute("DETACH DATABASE source_tier") + finally: + conn.close() + except sqlite3.Error as exc: + # This query protects an append from adopting another session's + # native id. An unavailable ownership view is unsafe to treat as + # unowned, so defer instead of using the global Codex fallback. + logger.warning( + "live.watcher: source-path ownership view unavailable for %s; refusing Codex identity fallback: %s", + path, + exc, + ) + return True + return row is not None + def _codex_session_meta_native_id(self, path: Path) -> str | None: try: with path.open("rb") as handle: @@ -3643,8 +3764,8 @@ def _archive_has_native_session(self, origin: str, native_id: str) -> bool: row = conn.execute( """ SELECT 1 - FROM sessions - WHERE origin = ? AND native_id = ? + FROM sessions AS s + WHERE s.origin = ? AND s.native_id = ? LIMIT 1 """, (origin, native_id), @@ -3655,7 +3776,7 @@ def _archive_has_native_session(self, origin: str, native_id: str) -> bool: return False return row is not None - def _existing_archive_session_native_id(self, path: Path) -> str | None: + def _existing_archive_session_native_id(self, path: Path, *, expected_origin: str) -> str | None: archive_root = Path(getattr(self._polylogue, "archive_root", self._cursor._db_path.parent)) index_db = ArchiveLocation.resolve(archive_root).active_index_path source_db = archive_root / "source.db" @@ -3670,11 +3791,11 @@ def _existing_archive_session_native_id(self, path: Path) -> str | None: SELECT s.native_id FROM sessions AS s JOIN source_tier.raw_sessions AS r ON r.raw_id = s.raw_id - WHERE r.source_path = ? + WHERE s.origin = ? AND r.origin = ? AND r.source_path = ? ORDER BY s.sort_key_ms DESC, s.created_at_ms DESC, s.session_id DESC LIMIT 1 """, - (str(path),), + (expected_origin, expected_origin, str(path)), ).fetchone() conn.execute("DETACH DATABASE source_tier") finally: @@ -3686,8 +3807,9 @@ def _existing_archive_session_native_id(self, path: Path) -> str | None: value = row[0] return value if isinstance(value, str) and value.strip() else None - def _claude_code_tail_matches_existing_identity(self, path: Path, payload: bytes) -> bool: - existing_id = self._existing_provider_session_id(path) + def _claude_code_tail_matches_existing_identity( + self, path: Path, payload: bytes, *, existing_id: str | None + ) -> bool: if existing_id is None: return False session_ids: set[str] = set() @@ -3835,5 +3957,20 @@ def _record_append_cursor(self, plan: _AppendPlan) -> bool: # fmt: off -__all__ = ["LiveBatchMetrics", "LiveBatchProcessor", "_FullIngestResult", "_LARGE_FULL_PARSE_PROGRESS_BYTES", "_MAX_APPEND_PLAN_PAYLOAD_BYTES", "_SMALL_FULL_PARSE_PROGRESS_MAX_BYTES", "_SMALL_FULL_PARSE_PROGRESS_MAX_FILES", "_STREAMING_FULL_INGEST_BYTES", "_full_ingest_worker_count", "_full_parse_progress_groups", "fingerprint_file", "last_complete_newline_from_tail"] +__all__ = [ + "AppendCapabilityReceipt", + "LiveBatchMetrics", + "LiveBatchProcessor", + "_FullIngestResult", + "_LARGE_FULL_PARSE_PROGRESS_BYTES", + "_MAX_APPEND_PLAN_PAYLOAD_BYTES", + "_SMALL_FULL_PARSE_PROGRESS_MAX_BYTES", + "_SMALL_FULL_PARSE_PROGRESS_MAX_FILES", + "_STREAMING_FULL_INGEST_BYTES", + "_full_ingest_worker_count", + "_full_parse_progress_groups", + "append_capability_receipt", + "fingerprint_file", + "last_complete_newline_from_tail", +] # fmt: on diff --git a/polylogue/storage/insights/session/latency_profiles.py b/polylogue/storage/insights/session/latency_profiles.py index d797bed4ea..75b935aa32 100644 --- a/polylogue/storage/insights/session/latency_profiles.py +++ b/polylogue/storage/insights/session/latency_profiles.py @@ -60,12 +60,14 @@ def build_session_latency_profile_record( ) if part ) + source_updated_at = profile.updated_at + source_sort_timestamp = source_updated_at or profile.created_at return SessionLatencyProfileRecord( session_id=SessionId(str(session.id)), materializer_version=SESSION_INSIGHT_MATERIALIZER_VERSION, materialized_at=built_at, - source_updated_at=_iso_datetime(session.updated_at), - source_sort_key=float(session.updated_at.timestamp()) if session.updated_at is not None else None, + source_updated_at=_iso_datetime(source_updated_at), + source_sort_key=float(source_sort_timestamp.timestamp()) if source_sort_timestamp is not None else None, input_high_water_mark=input_high_water_mark, input_high_water_mark_source=input_high_water_mark_source, input_row_count=input_row_count, diff --git a/polylogue/storage/insights/session/profiles.py b/polylogue/storage/insights/session/profiles.py index 8fc3cf6a2e..acf302a5db 100644 --- a/polylogue/storage/insights/session/profiles.py +++ b/polylogue/storage/insights/session/profiles.py @@ -350,13 +350,14 @@ def build_session_profile_record( evidence_search_text = profile_evidence_search_text(profile) inference_search_text = profile_inference_search_text(profile) source_updated_at = profile.updated_at.isoformat() if profile.updated_at else None + source_sort_timestamp = profile.updated_at or profile.created_at return SessionProfileRecord( session_id=SessionId(profile.session_id), logical_session_id=SessionId(resolved_logical_session_id), materializer_version=SESSION_INSIGHT_MATERIALIZER_VERSION, materialized_at=built_at, source_updated_at=source_updated_at, - source_sort_key=profile.updated_at.timestamp() if profile.updated_at else None, + source_sort_key=source_sort_timestamp.timestamp() if source_sort_timestamp else None, input_high_water_mark=source_updated_at, input_high_water_mark_source=classify_profile_hwm_source(profile.updated_at), input_row_count=profile.message_count, diff --git a/polylogue/storage/insights/session/rebuild.py b/polylogue/storage/insights/session/rebuild.py index c3dbaa0e1b..3c93bbf56b 100644 --- a/polylogue/storage/insights/session/rebuild.py +++ b/polylogue/storage/insights/session/rebuild.py @@ -787,6 +787,8 @@ def add_timing(name: str, started_at: float) -> None: profile, latency_facts, materialized_at=materialized_at, + input_high_water_mark=profile_record.input_high_water_mark, + input_high_water_mark_source=profile_record.input_high_water_mark_source, ) add_timing("build_records.latency_profile_record", t0) t0 = time.perf_counter() @@ -1239,7 +1241,14 @@ def _large_session_profile_record_from_row( FallbackReason.NO_USER_TURNS, ), ) - source_sort_key = float(row["sort_key"]) if row["sort_key"] is not None else None + source_sort_timestamp = updated_at or created_at + source_sort_key = ( + float(row["sort_key"]) + if row["sort_key"] is not None + else float(source_sort_timestamp.timestamp()) + if source_sort_timestamp is not None + else None + ) source_updated_at = updated_at.isoformat() if updated_at else None search_text = " \n".join(part for part in (origin, title, row["git_branch"], row["git_repository_url"]) if part) return SessionProfileRecord( @@ -1439,10 +1448,11 @@ def _stamp_bundle_materialization(conn: sqlite3.Connection, bundle: SessionInsig from polylogue.storage.sqlite.archive_tiers.write import apply_insight_materialization profile = bundle.profile_record + latency = bundle.latency_profile_record session_id = str(profile.session_id) materialized_at_ms = _epoch_ms_or_none(profile.materialized_at) or 0 source_updated_at_ms = _epoch_ms_or_none(profile.source_updated_at) - source_sort_key_ms = int(profile.source_sort_key * 1000) if profile.source_sort_key is not None else None + source_sort_key_ms = _source_sort_key_ms(profile.source_sort_key) input_high_water_mark_ms = _epoch_ms_or_none(profile.input_high_water_mark) # polylogue-f2qv.5: re-derive session_model_usage every time a session's # insights are rebuilt (missing-profile backfill, stale-version repair, or @@ -1452,7 +1462,7 @@ def _stamp_bundle_materialization(conn: sqlite3.Connection, bundle: SessionInsig provider_usage_row_count = _refresh_provider_usage_rollup(conn, session_id) for insight_type, materializer_version, input_row_count in ( ("session_profile", profile.materializer_version, profile.input_row_count), - ("latency", profile.materializer_version, bundle.latency_profile_record.input_row_count), + ("latency", latency.materializer_version, latency.input_row_count), ("work_events", SESSION_INSIGHT_MATERIALIZER_VERSION, len(bundle.work_event_records)), ("phases", SESSION_INSIGHT_MATERIALIZER_VERSION, len(bundle.phase_records)), ("runs", SESSION_INSIGHT_MATERIALIZER_VERSION, bundle.run_count), @@ -1461,20 +1471,34 @@ def _stamp_bundle_materialization(conn: sqlite3.Connection, bundle: SessionInsig ("thread", SESSION_INSIGHT_MATERIALIZER_VERSION, 1), ("provider_usage", SESSION_INSIGHT_MATERIALIZER_VERSION, provider_usage_row_count), ): + stamp_source_updated_at_ms = source_updated_at_ms + stamp_source_sort_key_ms = source_sort_key_ms + stamp_input_high_water_mark_ms = input_high_water_mark_ms + stamp_input_high_water_mark_source = profile.input_high_water_mark_source + if insight_type == "latency": + stamp_source_updated_at_ms = _epoch_ms_or_none(latency.source_updated_at) + stamp_source_sort_key_ms = _source_sort_key_ms(latency.source_sort_key) + stamp_input_high_water_mark_ms = _epoch_ms_or_none(latency.input_high_water_mark) + stamp_input_high_water_mark_source = latency.input_high_water_mark_source apply_insight_materialization( conn, insight_type=insight_type, session_id=session_id, materializer_version=materializer_version, materialized_at_ms=materialized_at_ms, - source_updated_at_ms=source_updated_at_ms, - source_sort_key_ms=source_sort_key_ms, - input_high_water_mark_ms=input_high_water_mark_ms, - input_high_water_mark_source=profile.input_high_water_mark_source, + source_updated_at_ms=stamp_source_updated_at_ms, + source_sort_key_ms=stamp_source_sort_key_ms, + input_high_water_mark_ms=stamp_input_high_water_mark_ms, + input_high_water_mark_source=stamp_input_high_water_mark_source, input_row_count=input_row_count, ) +def _source_sort_key_ms(source_sort_key: float | None) -> int | None: + """Recover the canonical integer millisecond key from a float-seconds value.""" + return round(source_sort_key * 1000) if source_sort_key is not None else None + + def _count_record_bundles( bundles: Sequence[SessionInsightRecordBundle], ) -> tuple[int, int, int]: diff --git a/tests/infra/convergence_harness.py b/tests/infra/convergence_harness.py index df2300984b..33258ee1ca 100644 --- a/tests/infra/convergence_harness.py +++ b/tests/infra/convergence_harness.py @@ -20,7 +20,7 @@ from dataclasses import dataclass from datetime import UTC, datetime, timedelta from pathlib import Path -from typing import cast +from typing import TYPE_CHECKING, cast import polylogue.daemon.convergence_stages as convergence_stages from polylogue.archive.message.roles import Role @@ -65,6 +65,9 @@ compose_quarantined_head_arrangement, ) +if TYPE_CHECKING: + from polylogue.maintenance.rebuild_index import RebuildIndexReceipt + SqlValue = str | int | float | bytes | None FactRow = tuple[SqlValue, ...] @@ -170,6 +173,35 @@ def build_converged_archive( return archive +def rebuild_retained_raw_index(archive: ConvergenceArchive | Path) -> RebuildIndexReceipt: + """Run the production source.db-retained reindex route for this archive.""" + from polylogue.maintenance.rebuild_index import RebuildIndexRequest, rebuild_index_from_source_sync + from tests.infra.rebuild_receipt import write_valid_rebuild_receipt + + root = archive.root if isinstance(archive, ConvergenceArchive) else archive + with sqlite3.connect(root / "source.db") as conn: + raw_session_count = int(conn.execute("SELECT COUNT(*) FROM raw_sessions").fetchone()[0]) + receipt_path = write_valid_rebuild_receipt( + root, + root.parent / f"{root.name}-test-schema-inference-receipt.json", + ) + receipt = rebuild_index_from_source_sync( + RebuildIndexRequest( + archive_root=root, + promote=True, + raw_batch_size=max(1, raw_session_count), + schema_inference_receipt_path=receipt_path, + ) + ) + if receipt.status != "replayed" or not receipt.materialized: + raise AssertionError(f"retained-raw production reindex did not materialize a generation: {receipt!r}") + if receipt.selected_raw_count != receipt.raw_session_count or receipt.raw_session_count == 0: + raise AssertionError(f"retained-raw reindex did not select every source raw row: {receipt!r}") + if receipt.operation.get("recovery_state") != "promoted": + raise AssertionError(f"retained-raw reindex did not record promotion recovery state: {receipt!r}") + return receipt + + def initialize_active_archive(root: Path) -> None: """Create all archive tiers for a temporary property-test archive.""" from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root @@ -876,6 +908,7 @@ def _stable_json(value: object, root: Path) -> str: "make_messages_fts_stale", "messages_fts_match_count", "raw_authority_facts", + "rebuild_retained_raw_index", "rich_convergence_pathology", "replay_convergence_archive", "rotated_session_order", diff --git a/tests/infra/inferred_corpus.py b/tests/infra/inferred_corpus.py index dbcf82d30d..2d89d9e8d7 100644 --- a/tests/infra/inferred_corpus.py +++ b/tests/infra/inferred_corpus.py @@ -14,7 +14,7 @@ from collections.abc import Mapping, Sequence from dataclasses import dataclass, replace from pathlib import Path -from typing import Literal, TypeAlias, cast +from typing import Literal, TypeAlias, cast, get_args from polylogue.core.json import JSONDocument from polylogue.core.sources import origin_from_provider @@ -30,16 +30,38 @@ from polylogue.schemas.synthetic import SyntheticCorpus from polylogue.schemas.synthetic.classification import ConstructSupport, classify_schema_constructs from polylogue.schemas.synthetic.models import SchemaRecord, SyntheticSchemaSelection -from polylogue.schemas.synthetic.wire_formats import PROVIDER_WIRE_FORMATS, WireFormat +from polylogue.schemas.synthetic.wire_formats import ( + CATALOG_ELEMENT_UNSUPPORTED_REASON, + PROVIDER_WIRE_FORMATS, + ConstructCoverage, + WireFormat, + WireParserWitness, + WireSupportEntry, + WireSupportReceipt, + build_wire_support_receipt, + validate_wire_support_entry_keys, + wire_support_entry_key, + wire_support_key, +) -INFERRED_CORPUS_MANIFEST_SCHEMA_VERSION = 1 +INFERRED_CORPUS_MANIFEST_SCHEMA_VERSION = 3 UnsupportedCorpusReason: TypeAlias = Literal[ "provider_without_wire_format", + "wire_support_selection_unwitnessed", + "wire_support_receipt_incomplete", + "unsupported_wire_route", "unsupported_element", "missing_schema", "unsupported_json_schema_construct", ] PackageReceipt: TypeAlias = JSONDocument +_WIRE_AUTHORITY_ONLY_REASONS = frozenset( + { + "wire_support_selection_unwitnessed", + "wire_support_receipt_incomplete", + "unsupported_wire_route", + } +) @dataclass(frozen=True, order=True) @@ -124,6 +146,7 @@ class InferredCorpusManifest: entries: tuple[InferredCorpusManifestEntry, ...] package_receipt: PackageReceipt | None = None + wire_support_receipt: JSONDocument | None = None def __post_init__(self) -> None: ordered = tuple(sorted(self.entries, key=lambda entry: entry.key)) @@ -160,6 +183,7 @@ def _payload_without_id(self) -> dict[str, object]: "schema_version": INFERRED_CORPUS_MANIFEST_SCHEMA_VERSION, "receipt_state": self.receipt_state, "package_receipt": self.package_receipt, + "wire_support_receipt": self.wire_support_receipt, "entries": [entry.to_payload() for entry in self.entries], } @@ -177,6 +201,7 @@ def from_payload(cls, payload: Mapping[str, object]) -> InferredCorpusManifest: "schema_version", "receipt_state", "package_receipt", + "wire_support_receipt", "entries", "payload_sha256", } @@ -194,15 +219,19 @@ def from_payload(cls, payload: Mapping[str, object]) -> InferredCorpusManifest: entries = tuple(_manifest_entry_from_payload(item) for item in raw_entries) receipt_state = payload.get("receipt_state") package_receipt = payload.get("package_receipt") + wire_support_receipt = payload.get("wire_support_receipt") if receipt_state not in {"catalog_only", "package_receipt_attached"}: raise ValueError(f"invalid inferred corpus manifest receipt_state: {receipt_state!r}") if receipt_state == "catalog_only" and package_receipt is not None: raise ValueError("catalog_only manifest must not carry a package receipt") if receipt_state == "package_receipt_attached" and not isinstance(package_receipt, dict): raise ValueError("package_receipt_attached manifest requires a JSON object receipt") + if wire_support_receipt is not None and not isinstance(wire_support_receipt, dict): + raise ValueError("wire_support_receipt must be a JSON object when present") manifest = cls( entries=entries, package_receipt=package_receipt if isinstance(package_receipt, dict) else None, + wire_support_receipt=wire_support_receipt if isinstance(wire_support_receipt, dict) else None, ) expected_manifest_id = manifest.manifest_id if payload.get("manifest_id") != expected_manifest_id: @@ -216,6 +245,7 @@ def from_payload(cls, payload: Mapping[str, object]) -> InferredCorpusManifest: "inferred corpus manifest payload integrity mismatch: " f"expected={expected_payload_sha256!r}, actual={payload.get('payload_sha256')!r}" ) + _wire_support_entry_index(manifest) return manifest @@ -368,12 +398,7 @@ def _manifest_entry_from_payload(payload: object) -> InferredCorpusManifestEntry details = raw_unsupported.get("details", []) if set(raw_unsupported) != {"reason", "details"}: raise ValueError("manifest unsupported record fields changed") - valid_reasons = { - "provider_without_wire_format", - "unsupported_element", - "missing_schema", - "unsupported_json_schema_construct", - } + valid_reasons = set(get_args(UnsupportedCorpusReason)) if ( reason not in valid_reasons or not isinstance(details, list) @@ -450,6 +475,7 @@ def build_inferred_corpus_convergence_handoff( ) -> InferredCorpusConvergenceHandoff: """Bind every supported row from memory or persisted disk to convergence.""" + read_validated_campaign_manifest = isinstance(manifest, Path) and campaign_mode persisted_manifest = ( read_inferred_corpus_manifest( manifest, @@ -461,7 +487,7 @@ def build_inferred_corpus_convergence_handoff( if isinstance(manifest, Path) else manifest ) - if campaign_mode: + if campaign_mode and not read_validated_campaign_manifest: _require_inference_handoff(persisted_manifest) if registry is None: raise ValueError("campaign mode requires a live schema registry") @@ -544,7 +570,131 @@ def _catalog_entries( return tuple(result) -def _unsupported_reason( +def _wire_support_entry_index( + manifest: InferredCorpusManifest, +) -> dict[tuple[str, str | None, str | None], WireSupportEntry]: + """Recover the exact wire decision bound into a persisted manifest. + + The manifest stores the canonical receipt payload so a campaign read can + replay the same unsupported-route decision without consulting a fresh + support probe. Treat malformed authority as a hard validation failure; + falling back to an unbound decision would make the persisted claim weaker. + """ + + receipt = manifest.wire_support_receipt + if receipt is None: + return {} + raw_entries = receipt.get("entries") + if not isinstance(raw_entries, list): + raise ValueError("wire_support_receipt entries must be a list") + entries: list[WireSupportEntry] = [] + for raw_entry in raw_entries: + if not isinstance(raw_entry, Mapping): + raise ValueError("wire_support_receipt entries must be objects") + entries.append(_wire_support_entry_from_payload(raw_entry)) + validate_wire_support_entry_keys(entries, boundary="persisted wire support receipt") + return {wire_support_entry_key(entry): entry for entry in entries} + + +def _wire_support_entry_from_payload(payload: Mapping[str, object]) -> WireSupportEntry: + def required_string(field: str) -> str: + value = payload.get(field) + if not isinstance(value, str): + raise ValueError(f"wire_support_receipt entry {field} must be a string") + return value + + def optional_string(field: str) -> str | None: + value = payload.get(field) + if value is not None and not isinstance(value, str): + raise ValueError(f"wire_support_receipt entry {field} must be a string or null") + return value + + def required_int(value: object, field: str) -> int: + if isinstance(value, bool) or not isinstance(value, int): + raise ValueError(f"wire_support_receipt entry {field} must be an integer") + return value + + def string_tuple(value: object, field: str) -> tuple[str, ...]: + if not isinstance(value, list) or not all(isinstance(item, str) for item in value): + raise ValueError(f"wire_support_receipt entry {field} must be a list of strings") + return tuple(value) + + status = payload.get("status") + if status not in {"supported", "unsupported"}: + raise ValueError("wire_support_receipt entry status is invalid") + schema_valid = payload.get("schema_valid") + if schema_valid is not None and not isinstance(schema_valid, bool): + raise ValueError("wire_support_receipt entry schema_valid must be boolean or null") + raw_coverage = payload.get("construct_coverage") + coverage: ConstructCoverage | None = None + if raw_coverage is not None: + if not isinstance(raw_coverage, Mapping): + raise ValueError("wire_support_receipt construct_coverage must be an object or null") + raw_reasons = raw_coverage.get("nonrepresentable_reasons") + if not isinstance(raw_reasons, list) or not all(isinstance(item, Mapping) for item in raw_reasons): + raise ValueError("wire_support_receipt nonrepresentable_reasons must be a list of objects") + reasons: list[tuple[str, str]] = [] + for item in raw_reasons: + keyword = item.get("keyword") + reason = item.get("reason") + if not isinstance(keyword, str) or not isinstance(reason, str): + raise ValueError("wire_support_receipt nonrepresentable reasons require strings") + reasons.append((keyword, reason)) + coverage = ConstructCoverage( + schema_keywords=string_tuple(raw_coverage.get("schema_keywords"), "schema_keywords"), + exercised_keywords=string_tuple(raw_coverage.get("exercised_keywords"), "exercised_keywords"), + missing_keywords=string_tuple(raw_coverage.get("missing_keywords"), "missing_keywords"), + nonrepresentable_keywords=string_tuple( + raw_coverage.get("nonrepresentable_keywords"), "nonrepresentable_keywords" + ), + nonrepresentable_reasons=tuple(reasons), + ) + raw_witnesses = payload.get("parser_witnesses") + if not isinstance(raw_witnesses, list) or not all(isinstance(item, Mapping) for item in raw_witnesses): + raise ValueError("wire_support_receipt parser_witnesses must be a list of objects") + witnesses: list[WireParserWitness] = [] + for raw_witness in raw_witnesses: + artifact_kind = raw_witness.get("artifact_kind") + if artifact_kind not in {"baseline", "coverage"}: + raise ValueError("wire_support_receipt parser witness artifact_kind is invalid") + validation_error = raw_witness.get("validation_error") + if validation_error is not None and not isinstance(validation_error, str): + raise ValueError("wire_support_receipt parser witness validation_error must be string or null") + witnesses.append( + WireParserWitness( + index=required_int(raw_witness.get("index"), "parser_witness.index"), + exercised_keywords=string_tuple( + raw_witness.get("exercised_keywords"), "parser_witness.exercised_keywords" + ), + parsed_session_count=required_int( + raw_witness.get("parsed_session_count"), "parser_witness.parsed_session_count" + ), + parsed_message_count=required_int( + raw_witness.get("parsed_message_count"), "parser_witness.parsed_message_count" + ), + validation_error=validation_error, + artifact_kind=cast(Literal["baseline", "coverage"], artifact_kind), + artifact_evidence=string_tuple( + raw_witness.get("artifact_evidence"), "parser_witness.artifact_evidence" + ), + ) + ) + return WireSupportEntry( + provider=required_string("provider"), + status=status, + reason=optional_string("reason"), + package_version=optional_string("package_version"), + element_kind=optional_string("element_kind"), + schema_valid=schema_valid, + parsed_session_count=required_int(payload.get("parsed_session_count"), "parsed_session_count"), + parsed_message_count=required_int(payload.get("parsed_message_count"), "parsed_message_count"), + construct_coverage=coverage, + validation_error=optional_string("validation_error"), + parser_witnesses=tuple(witnesses), + ) + + +def _schema_unsupported_reason( *, element: SchemaElementManifest, schema: SchemaRecord | None, @@ -563,6 +713,49 @@ def _unsupported_reason( return None +def _unsupported_reason( + *, + element: SchemaElementManifest, + schema: SchemaRecord | None, + wire_format: WireFormat | None, + construct_support: tuple[ConstructSupport, ...], + support_entry: WireSupportEntry | None, + support_receipt_bound: bool, +) -> UnsupportedCorpusRecord | None: + if support_entry is not None: + if support_entry.status == "unsupported": + reason: UnsupportedCorpusReason = ( + "unsupported_element" + if support_entry.reason == CATALOG_ELEMENT_UNSUPPORTED_REASON + else "unsupported_wire_route" + ) + return UnsupportedCorpusRecord( + reason, + (support_entry.reason or "route is explicitly unsupported",), + ) + if not support_entry.healthy: + details = tuple( + detail + for detail in ( + support_entry.validation_error, + *(witness.validation_error for witness in support_entry.parser_witnesses), + ) + if detail + ) + return UnsupportedCorpusRecord("wire_support_receipt_incomplete", details) + elif support_receipt_bound: + return UnsupportedCorpusRecord( + "wire_support_selection_unwitnessed", + (f"no exact parser witness for {element.schema_file!r}",), + ) + return _schema_unsupported_reason( + element=element, + schema=schema, + wire_format=wire_format, + construct_support=construct_support, + ) + + def _compile_entry( *, provider: str, @@ -570,6 +763,8 @@ def _compile_entry( element: SchemaElementManifest, registry: RuntimeSchemaRegistryLike, wire_formats: Mapping[str, WireFormat], + support_entry: WireSupportEntry | None, + support_receipt_bound: bool, ) -> InferredCorpusManifestEntry: key_without_constructs = CorpusManifestKey(provider, package.version, element.element_kind) schema = registry.get_element_schema( @@ -585,6 +780,8 @@ def _compile_entry( schema=schema if isinstance(schema, dict) else None, wire_format=wire_format, construct_support=construct_support, + support_entry=support_entry, + support_receipt_bound=support_receipt_bound, ) if unsupported is not None: return InferredCorpusManifestEntry(key=key, unsupported=unsupported) @@ -661,6 +858,7 @@ def compile_inferred_corpus_manifest( registry: RuntimeSchemaRegistryLike, package_receipt: PackageReceipt | None = None, wire_formats: Mapping[str, WireFormat] | None = None, + wire_support_receipt: WireSupportReceipt | None = None, providers: Sequence[str] | None = None, campaign_mode: bool = False, gate_receipt_path: Path | None = None, @@ -669,8 +867,17 @@ def compile_inferred_corpus_manifest( """Compile every persisted package/version/element into a typed manifest.""" formats = PROVIDER_WIRE_FORMATS if wire_formats is None else wire_formats + support_entries: dict[tuple[str, str | None, str | None], WireSupportEntry] = {} + if wire_support_receipt is not None: + validate_wire_support_entry_keys(wire_support_receipt.entries, boundary="manifest wire support receipt") + support_entries = {wire_support_entry_key(entry): entry for entry in wire_support_receipt.entries} if campaign_mode and package_receipt is None: raise ValueError("campaign mode requires a persisted schema-inference handoff") + if campaign_mode and wire_support_receipt is not None and wire_support_receipt.missing_routes: + raise ValueError( + "campaign mode requires an explicit synthetic wire route for every catalog provider: " + f"missing={list(wire_support_receipt.missing_routes)!r}" + ) entries = tuple( _compile_entry( provider=provider, @@ -678,11 +885,17 @@ def compile_inferred_corpus_manifest( element=element, registry=registry, wire_formats=formats, + support_entry=support_entries.get(wire_support_key(provider, package.version, element.element_kind)), + support_receipt_bound=wire_support_receipt is not None, ) for provider, catalog, package, element in _catalog_entries(registry, providers) ) manifest = InferredCorpusManifest( - entries=tuple(sorted(entries, key=lambda entry: entry.key)), package_receipt=package_receipt + entries=tuple(sorted(entries, key=lambda entry: entry.key)), + package_receipt=package_receipt, + wire_support_receipt=cast(JSONDocument, wire_support_receipt.to_dict()) + if wire_support_receipt is not None + else None, ) assert_inferred_corpus_manifest_complete(manifest, registry, providers=providers) if campaign_mode: @@ -710,6 +923,12 @@ def _validate_inference_handoff( gate_receipt_path=gate_receipt_path, archive_root=archive_root, ) + current_wire_support = _validate_current_wire_support_route(manifest, registry) + if current_wire_support is not None and current_wire_support.missing_routes: + raise ValueError( + "campaign mode requires an explicit synthetic wire route for every catalog provider: " + f"missing={list(current_wire_support.missing_routes)!r}" + ) if not manifest.supported_specs: raise ValueError("campaign mode has no executable synthetic corpus selection") expected_packages = package_hashes_for_registry(cast(SchemaReceiptRegistry, registry), providers) @@ -717,6 +936,7 @@ def _validate_inference_handoff( raise ValueError("schema-inference handoff package/version/element hashes do not match the registry") catalog_entries = _catalog_entries(registry, providers) + support_entries = _wire_support_entry_index(manifest) expected_coverage = {(provider, origin_from_provider(provider).value) for provider, *_rest in catalog_entries} actual_coverage = {(item.provider, item.origin) for item in receipt.coverage_decisions} if actual_coverage != expected_coverage: @@ -730,32 +950,55 @@ def _validate_inference_handoff( entries_by_provider.setdefault(entry.key.provider, []).append(entry) for coverage in receipt.coverage_decisions: provider_entries = entries_by_provider.get(coverage.provider, []) + # The package receipt records schema inference authority. Wire-route + # refusals are independently bound by the serialized WireSupportReceipt + # and must not rewrite a committed schema package decision. + schema_blocking_reasons = tuple( + entry.unsupported.reason + for entry in provider_entries + if entry.unsupported is not None and entry.unsupported.reason not in _WIRE_AUTHORITY_ONLY_REASONS + ) if any(entry.spec is not None for entry in provider_entries): expected_decision = "committed" - elif provider_entries and all( - entry.unsupported is not None and entry.unsupported.reason == "unsupported_json_schema_construct" - for entry in provider_entries - ): + elif not schema_blocking_reasons: + expected_decision = "committed" if coverage.provider in PROVIDER_WIRE_FORMATS else "unsupported" + elif all(reason == "unsupported_json_schema_construct" for reason in schema_blocking_reasons): expected_decision = "nonrepresentable" else: expected_decision = "unsupported" if coverage.decision != expected_decision: raise ValueError("schema-inference handoff coverage decision changed") + expected_unsupported: set[tuple[str, str, str, str, str, tuple[str, ...]]] = set() + entries_by_wire_key = { + wire_support_key(entry.key.provider, entry.key.package_version, entry.key.element_kind): entry + for entry in manifest.entries + } for provider, _catalog, package, element in catalog_entries: - live_entry = next( - ( - candidate - for candidate in manifest.entries - if (candidate.key.provider, candidate.key.package_version, candidate.key.element_kind) - == (provider, package.version, element.element_kind) - ), - None, - ) + live_entry = entries_by_wire_key.get(wire_support_key(provider, package.version, element.element_kind)) if live_entry is None: raise ValueError("schema-inference manifest is missing a live registry entry") live_schema = registry.get_element_schema(provider, version=package.version, element_kind=element.element_kind) live_constructs = _schema_constructs(live_schema) + schema_unsupported = _schema_unsupported_reason( + element=element, + schema=live_schema if isinstance(live_schema, dict) else None, + wire_format=PROVIDER_WIRE_FORMATS.get(provider), + construct_support=live_constructs, + ) + if schema_unsupported is not None: + expected_unsupported.add( + ( + provider, + package.version, + element.element_kind, + "nonrepresentable" + if schema_unsupported.reason == "unsupported_json_schema_construct" + else "unsupported", + schema_unsupported.reason, + schema_unsupported.details, + ) + ) if live_entry.key.construct_support != live_constructs: raise ValueError("schema-inference manifest classifier output changed") live_unsupported = _unsupported_reason( @@ -763,6 +1006,8 @@ def _validate_inference_handoff( schema=live_schema if isinstance(live_schema, dict) else None, wire_format=PROVIDER_WIRE_FORMATS.get(provider), construct_support=live_constructs, + support_entry=support_entries.get(wire_support_key(provider, package.version, element.element_kind)), + support_receipt_bound=manifest.wire_support_receipt is not None, ) if (live_entry.unsupported is None) != (live_unsupported is None): raise ValueError("schema-inference manifest executable support changed") @@ -786,18 +1031,6 @@ def _validate_inference_handoff( if len(witness) != 1 or not witness[0]: raise ValueError("schema-inference manifest selection produced no executable witness") - expected_unsupported = { - ( - entry.key.provider, - entry.key.package_version, - entry.key.element_kind, - "nonrepresentable" if entry.unsupported.reason == "unsupported_json_schema_construct" else "unsupported", - entry.unsupported.reason, - entry.unsupported.details, - ) - for entry in manifest.entries - if entry.unsupported is not None - } actual_unsupported = { ( item.provider, @@ -816,6 +1049,41 @@ def _validate_inference_handoff( ) +def _validate_current_wire_support_route( + manifest: InferredCorpusManifest, + registry: RuntimeSchemaRegistryLike, +) -> WireSupportReceipt | None: + """Re-run the exact persisted wire witnesses through current production code.""" + + persisted = manifest.wire_support_receipt + if persisted is None: + return None + witness_seed = persisted.get("witness_seed") + if isinstance(witness_seed, bool) or not isinstance(witness_seed, int): + raise ValueError("wire_support_receipt witness_seed must be an integer") + raw_providers = persisted.get("catalog_providers") + if not isinstance(raw_providers, list) or not all(isinstance(provider, str) for provider in raw_providers): + raise ValueError("wire_support_receipt catalog_providers must be a list of strings") + catalog_scope = persisted.get("catalog_scope") + if catalog_scope not in {"registry-default", "explicit"}: + raise ValueError("wire_support_receipt catalog_scope must be registry-default or explicit") + current = build_wire_support_receipt( + registry=registry, + seed=witness_seed, + providers=None + if catalog_scope == "registry-default" + else tuple(cast(str, provider) for provider in raw_providers), + ) + rebuilt = current.to_dict() + if rebuilt != persisted: + changed_fields = sorted(key for key in set(rebuilt) | set(persisted) if rebuilt.get(key) != persisted.get(key)) + raise ValueError( + "schema-inference wire-support receipt changed under the current parser or wire-normalizer route: " + f"changed_fields={changed_fields!r}" + ) + return current + + __all__ = [ "ConstructSupport", "CorpusManifestKey", diff --git a/tests/integration/test_live_read_amplification.py b/tests/integration/test_live_read_amplification.py index 36c5ef99e1..c788a95ff8 100644 --- a/tests/integration/test_live_read_amplification.py +++ b/tests/integration/test_live_read_amplification.py @@ -141,7 +141,8 @@ async def fake_full_ingest( def fake_append_ingest(plans: list[Any]) -> _AppendResult: return _AppendResult(succeeded=plans, failed=[], worker_count=1) - def fake_existing_provider_session_id(path: Path) -> str | None: + def fake_existing_provider_session_id(path: Path, *, expected_origin: str) -> str | None: + del expected_origin return existing_ids.get(path) with ( @@ -185,6 +186,70 @@ def _seed_initial_ingest(proc: LiveBatchProcessor, path: Path, *, session_id: st cast(Any, proc)._test_existing_ids[path] = session_id +def test_claude_code_append_plan_consumes_identity_capability_gate( + processor: tuple[LiveBatchProcessor, Path, Path], + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The production planner must consume the identity receipt before tail matching.""" + from polylogue.sources.live import batch as live_batch + + proc, root, _ = processor + path = root / "session-abc.jsonl" + _write_jsonl(path, [_claude_code_record(session_id="abc", uuid="message-0")]) + _seed_initial_ingest(proc, path, session_id="abc") + _append_jsonl(path, [_claude_code_record(session_id="abc", uuid="message-1", role="assistant", text="tail")]) + + seen: list[tuple[str, bool]] = [] + original_receipt = live_batch.append_capability_receipt + + def capture_receipt(**kwargs: object) -> object: + seen.append((str(kwargs["provider"]), bool(kwargs["stable_session_identity"]))) + return original_receipt(**kwargs) # type: ignore[arg-type] + + monkeypatch.setattr(live_batch, "append_capability_receipt", capture_receipt) + monkeypatch.setattr(proc, "_existing_provider_session_id", lambda _path, **_kwargs: None) + monkeypatch.setattr( + proc, + "_claude_code_tail_matches_existing_identity", + lambda *_args: pytest.fail("tail matching ran before the identity capability gate"), + ) + + assert proc._append_plan(path) is None + assert seen == [("claude-code", False)] + + +def test_claude_code_append_reuses_identity_for_tail_matching( + processor: tuple[LiveBatchProcessor, Path, Path], + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The hot append route performs one identity lookup and passes it to tail matching.""" + proc, root, _ = processor + path = root / "session-abc.jsonl" + _write_jsonl(path, [_claude_code_record(session_id="abc", uuid="message-0")]) + _seed_initial_ingest(proc, path, session_id="abc") + _append_jsonl(path, [_claude_code_record(session_id="abc", uuid="message-1", role="assistant", text="tail")]) + + lookup_count = 0 + matched_identity: list[str | None] = [] + original_lookup = proc._existing_provider_session_id + + def count_lookup(candidate: Path, *, expected_origin: str) -> str | None: + nonlocal lookup_count + lookup_count += 1 + return original_lookup(candidate, expected_origin=expected_origin) + + def capture_identity(_path: Path, _payload: bytes, *, existing_id: str | None) -> bool: + matched_identity.append(existing_id) + return True + + monkeypatch.setattr(proc, "_existing_provider_session_id", count_lookup) + monkeypatch.setattr(proc, "_claude_code_tail_matches_existing_identity", capture_identity) + + assert proc._append_plan(path) is not None + assert lookup_count == 1 + assert matched_identity == ["abc"] + + # --------------------------------------------------------------------------- # Scenario 1 — active Claude Code session appended to # --------------------------------------------------------------------------- diff --git a/tests/property/test_inferred_corpus_loop.py b/tests/property/test_inferred_corpus_loop.py index 8690f9bd93..018d6e9f6b 100644 --- a/tests/property/test_inferred_corpus_loop.py +++ b/tests/property/test_inferred_corpus_loop.py @@ -3,21 +3,38 @@ from __future__ import annotations import asyncio +import json +import os import re import shutil import sqlite3 +import subprocess +import sys +from collections.abc import Sequence +from dataclasses import replace from pathlib import Path import pytest from polylogue.config import Source +from polylogue.core.enums import Provider from polylogue.core.outcomes import OutcomeStatus from polylogue.daemon.convergence import DaemonConverger from polylogue.daemon.convergence_stages import make_fts_stage, make_insights_stage +from polylogue.daemon.fts_startup import record_fts_freshness_snapshot_sync from polylogue.maintenance.archive_verification import verify_archive from polylogue.pipeline.services.archive_ingest import parse_sources_archive +from polylogue.scenarios import CorpusSpec from polylogue.schemas.registry import SCHEMA_DIR, SchemaRegistry from polylogue.schemas.synthetic import SyntheticCorpus +from polylogue.schemas.synthetic.models import SyntheticSchemaSelection +from polylogue.schemas.synthetic.wire_formats import build_wire_support_receipt +from polylogue.sources.live.cursor import CursorStore +from polylogue.sources.revision_backfill import backfill_historical_revision_evidence +from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore +from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root +from tests.infra.archive_canonical_snapshot import archive_snapshot, assert_archives_equivalent +from tests.infra.convergence_harness import rebuild_retained_raw_index, set_debt_retry_at from tests.infra.inferred_corpus import ( assert_inferred_corpus_convergence_handoff_complete, build_inferred_corpus_convergence_handoff, @@ -51,12 +68,163 @@ def _assert_fts_match(conn: sqlite3.Connection, token: str) -> None: assert rows, f"FTS MATCH returned no blocks for generated token {token!r}" +def _run_retry_in_fresh_process(index_db: Path) -> int: + """Exercise the production debt drain after a real interpreter restart.""" + repo_root = Path(__file__).resolve().parents[2] + env = os.environ.copy() + existing_pythonpath = env.get("PYTHONPATH") + env["PYTHONPATH"] = str(repo_root) if not existing_pythonpath else f"{repo_root}{os.pathsep}{existing_pythonpath}" + env["POLYLOGUE_ARCHIVE_ROOT"] = str(index_db.parent) + script = ( + "from pathlib import Path\n" + "from polylogue.daemon.cli import _drain_convergence_debt_once\n" + f"print('RETRIED=' + str(_drain_convergence_debt_once(Path({str(index_db)!r}))))\n" + ) + completed = subprocess.run( + [sys.executable, "-c", script], + cwd=repo_root, + env=env, + capture_output=True, + text=True, + timeout=60, + check=False, + ) + assert completed.returncode == 0, ( + f"fresh-process convergence retry failed\nstdout:\n{completed.stdout}\nstderr:\n{completed.stderr}" + ) + for line in reversed(completed.stdout.splitlines()): + if line.startswith("RETRIED="): + return int(line.removeprefix("RETRIED=")) + raise AssertionError(f"fresh-process convergence retry emitted no result marker: {completed.stdout!r}") + + +def _inferred_selection() -> tuple[CorpusSpec, SyntheticSchemaSelection]: + registry = SchemaRegistry(storage_root=SCHEMA_DIR) + manifest = compile_inferred_corpus_manifest( + registry=registry, + wire_support_receipt=build_wire_support_receipt(registry=registry), + ) + handoff = build_inferred_corpus_convergence_handoff(manifest) + for spec, selection in zip(handoff.specs, handoff.selections, strict=True): + if spec.provider == "codex": + return spec, selection + raise AssertionError("the persisted inferred corpus has no Codex selection") + + +def _ingest_and_converge_sources( + archive_root: Path, + sources: Sequence[Source], +) -> tuple[str, ...]: + initialize_active_archive_root(archive_root) + raw_ids: list[str] = [] + with ArchiveStore.open_existing(archive_root, read_only=False) as archive: + for source_index, source in enumerate(sources): + if source.path is None: + raise AssertionError(f"source path required for inferred fixture: {source.name}") + raw_ids.append( + archive.write_raw_payload( + provider=Provider.from_string(source.name), + payload=source.path.read_bytes(), + source_path=str(source.path), + source_index=source_index, + acquired_at_ms=source_index + 1, + ) + ) + backfill = backfill_historical_revision_evidence(archive_root, selected_raw_ids=raw_ids, ingest_workers=1) + assert backfill.scanned == backfill.classified_full > 0 + assert backfill.quarantined == 0 + assert backfill.adoption_deferred == 0 + with sqlite3.connect(archive_root / "index.db") as conn: + session_ids = tuple(str(row[0]) for row in conn.execute("SELECT session_id FROM sessions ORDER BY session_id")) + states, _timings = DaemonConverger( + (make_fts_stage(archive_root / "index.db"), make_insights_stage(archive_root / "index.db")) + ).converge_sessions(session_ids) + assert states and all(state.converged and state.last_error is None for state in states.values()) + with sqlite3.connect(archive_root / "index.db") as conn: + record_fts_freshness_snapshot_sync(conn) + return session_ids + + +def _converge_existing_archive(archive_root: Path) -> None: + """Run post-reindex convergence over the promoted generation.""" + with sqlite3.connect(archive_root / "index.db") as conn: + session_ids = tuple(str(row[0]) for row in conn.execute("SELECT session_id FROM sessions ORDER BY session_id")) + states, _timings = DaemonConverger( + (make_fts_stage(archive_root / "index.db"), make_insights_stage(archive_root / "index.db")) + ).converge_sessions(session_ids) + assert states and all(state.converged and state.last_error is None for state in states.values()) + with sqlite3.connect(archive_root / "index.db") as conn: + record_fts_freshness_snapshot_sync(conn) + + +def _lineage_material() -> tuple[bytes, bytes, str, str]: + spec, selection = _inferred_selection() + parent_spec = replace( + spec, + count=1, + messages_min=3, + messages_max=3, + seed=101, + session_native_ids=("inferred-lineage-parent",), + style="demo-attachments", + ) + child_spec = replace( + spec, + count=1, + messages_min=3, + messages_max=3, + seed=202, + session_native_ids=("inferred-lineage-child",), + style="demo-attachments", + ) + parent_raw = SyntheticCorpus.generate_batch_for_selection(selection, parent_spec).artifacts[0].raw_bytes + child_raw = SyntheticCorpus.generate_batch_for_selection(selection, child_spec).artifacts[0].raw_bytes + parent_records = [json.loads(line) for line in parent_raw.decode().splitlines() if line] + child_records = [json.loads(line) for line in child_raw.decode().splitlines() if line] + child_meta = next(record for record in child_records if record.get("type") == "session_meta") + child_meta.setdefault("payload", {})["forked_from_id"] = "inferred-lineage-parent" + child_records = [ + child_meta, + *(record for record in parent_records if record.get("type") != "session_meta"), + *(record for record in child_records if record.get("type") != "session_meta"), + ] + return ( + ("\n".join(json.dumps(record, sort_keys=True) for record in parent_records) + "\n").encode(), + ("\n".join(json.dumps(record, sort_keys=True) for record in child_records) + "\n").encode(), + "codex-session:inferred-lineage-parent", + "codex-session:inferred-lineage-child", + ) + + +def _build_lineage_archive( + archive_root: Path, + parent_raw: bytes, + child_raw: bytes, +) -> tuple[str, ...]: + source_root = archive_root.parent / "inferred-lineage-material" + source_root.mkdir(parents=True, exist_ok=True) + parent_path = source_root / "parent.jsonl" + child_path = source_root / "child.jsonl" + parent_path.write_bytes(parent_raw) + child_path.write_bytes(child_raw) + return _ingest_and_converge_sources( + archive_root, + ( + Source(name="codex", path=source_root / "parent.jsonl"), + Source(name="codex", path=source_root / "child.jsonl"), + ), + ) + + def test_persisted_catalog_manifest_reaches_real_ingest_and_convergence( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: registry = SchemaRegistry(storage_root=SCHEMA_DIR) - manifest = compile_inferred_corpus_manifest(registry=registry) + manifest = compile_inferred_corpus_manifest( + registry=registry, + wire_support_receipt=build_wire_support_receipt(registry=registry), + ) manifest_path = tmp_path / "manifest.json" write_inferred_corpus_manifest(manifest, manifest_path) persisted = read_inferred_corpus_manifest(manifest_path) @@ -165,7 +333,10 @@ def test_every_supported_inferred_element_reaches_convergence_and_red_twin( """ registry = SchemaRegistry(storage_root=SCHEMA_DIR) - manifest = compile_inferred_corpus_manifest(registry=registry) + manifest = compile_inferred_corpus_manifest( + registry=registry, + wire_support_receipt=build_wire_support_receipt(registry=registry), + ) manifest_path = tmp_path / "manifest.json" write_inferred_corpus_manifest(manifest, manifest_path) persisted = read_inferred_corpus_manifest(manifest_path) @@ -234,3 +405,125 @@ def test_every_supported_inferred_element_reaches_convergence_and_red_twin( red = verify_archive(broken_root, checks=("message-count-projection",)) check = next(item for item in red.checks if item.name == "message-count-projection") assert check.status is OutcomeStatus.ERROR + + +@pytest.mark.frozen_clock_modules("polylogue.storage.sqlite.archive_tiers.revision_governance") +def test_inferred_selection_retained_raw_reindex_matches_canonical_snapshot( + tmp_path: Path, + frozen_clock: object, +) -> None: + spec, selection = _inferred_selection() + source_root = tmp_path / "retained-source" + written = SyntheticCorpus.write_selection_artifacts(selection, spec, source_root, prefix="retained") + archive_root = tmp_path / "archive" + session_ids = _ingest_and_converge_sources( + archive_root, + (Source(name=spec.provider, path=written.files[0]),), + ) + before = archive_snapshot(archive_root, session_ids=session_ids) + + receipt = rebuild_retained_raw_index(archive_root) + _converge_existing_archive(archive_root) + + assert receipt.raw_session_count == receipt.selected_raw_count > 0 + assert archive_snapshot(archive_root, session_ids=session_ids) == before + + +@pytest.mark.uses_real_clock("fresh-process debt recovery crosses a subprocess wall-clock retry boundary") +def test_inferred_selection_debt_recovers_in_a_fresh_process(tmp_path: Path) -> None: + spec, selection = _inferred_selection() + source_root = tmp_path / "recovery-source" + written = SyntheticCorpus.write_selection_artifacts(selection, spec, source_root, prefix="recovery") + archive_root = tmp_path / "archive" + session_ids = _ingest_and_converge_sources( + archive_root, + (Source(name=spec.provider, path=written.files[0]),), + ) + baseline = archive_snapshot(archive_root, session_ids=session_ids) + with sqlite3.connect(archive_root / "index.db") as conn: + conn.execute( + "DELETE FROM session_profiles WHERE session_id IN ({})".format(",".join("?" for _ in session_ids)), + session_ids, + ) + conn.commit() + cursor = CursorStore(archive_root / "index.db") + for session_id in session_ids: + cursor.record_convergence_debt( + stage="insights", + subject_type="session_id", + subject_id=session_id, + error="inferred-corpus convergence interruption", + ) + set_debt_retry_at( + archive_root / "ops.db", + stage="insights", + subject_type="session_id", + subject_id=session_id, + retry_at="1970-01-01T00:00:00+00:00", + ) + + assert _run_retry_in_fresh_process(archive_root / "index.db") == len(session_ids) + assert CursorStore(archive_root / "index.db").list_convergence_debt(limit=100) == [] + assert archive_snapshot(archive_root, session_ids=session_ids) == baseline + + +@pytest.mark.frozen_clock_modules("polylogue.storage.sqlite.archive_tiers.revision_governance") +def test_inferred_lineage_reindex_preserves_composition_and_detects_tail_mutation( + tmp_path: Path, + frozen_clock: object, +) -> None: + parent_raw, child_raw, parent_id, child_id = _lineage_material() + canonical_root = tmp_path / "canonical" + canonical_ids = _build_lineage_archive(canonical_root, parent_raw, child_raw) + + assert set(canonical_ids) == {parent_id, child_id} + with sqlite3.connect(canonical_root / "index.db") as conn: + parent = conn.execute( + "SELECT root_session_id, message_count FROM sessions WHERE session_id = ?", + (parent_id,), + ).fetchone() + child = conn.execute( + "SELECT parent_session_id, root_session_id, message_count FROM sessions WHERE session_id = ?", + (child_id,), + ).fetchone() + link = conn.execute( + "SELECT resolved_dst_session_id, branch_point_message_id, inheritance, status " + "FROM session_links WHERE src_session_id = ?", + (child_id,), + ).fetchone() + assert parent is not None and child is not None and link is not None + assert parent[0] == parent_id + assert child[0] == parent_id and child[1] == parent_id and int(child[2]) == 3 + assert link[0] == parent_id and link[1] is not None and link[2] == "prefix-sharing" and link[3] is None + with ArchiveStore.open_existing(canonical_root) as archive: + composed = archive.read_session(child_id) + assert len(composed.messages) > int(parent[1]) + assert any(message.message_id.startswith(parent_id + ":") for message in composed.messages) + + rebuilt_root = tmp_path / "rebuilt" + _build_lineage_archive(rebuilt_root, parent_raw, child_raw) + rebuild_retained_raw_index(rebuilt_root) + _converge_existing_archive(rebuilt_root) + assert archive_snapshot(rebuilt_root) == archive_snapshot(canonical_root) + + mutated_records = [json.loads(line) for line in child_raw.decode().splitlines() if line] + mutated_message = mutated_records[-1].get("payload", mutated_records[-1]) + assert isinstance(mutated_message, dict) + mutated_message["content"] = [{"type": "output_text", "text": "mutation-sensitive inferred lineage tail"}] + mutated_raw = ("\n".join(json.dumps(record, sort_keys=True) for record in mutated_records) + "\n").encode() + mutated_root = tmp_path / "mutated" + _build_lineage_archive(mutated_root, parent_raw, mutated_raw) + with pytest.raises(AssertionError, match="canonical archive snapshots differ"): + assert_archives_equivalent(canonical_root, mutated_root) + with sqlite3.connect(mutated_root / "index.db") as conn: + assert ( + conn.execute( + "SELECT 1 FROM blocks WHERE session_id = ? AND text = ?", + (child_id, "mutation-sensitive inferred lineage tail"), + ).fetchone() + is not None + ) + assert conn.execute( + "SELECT resolved_dst_session_id, inheritance, status FROM session_links WHERE src_session_id = ?", + (child_id,), + ).fetchone() == (parent_id, "prefix-sharing", None) diff --git a/tests/unit/core/test_synthetic_wire_support.py b/tests/unit/core/test_synthetic_wire_support.py index a7ffbd846d..9b78352bd7 100644 --- a/tests/unit/core/test_synthetic_wire_support.py +++ b/tests/unit/core/test_synthetic_wire_support.py @@ -5,12 +5,13 @@ import json from collections.abc import Mapping from copy import deepcopy +from dataclasses import replace from pathlib import Path import pytest from polylogue.config import Source -from polylogue.core.enums import Provider, Role +from polylogue.core.enums import BlockType, MaterialOrigin, Provider, Role from polylogue.core.json import JSONValue from polylogue.schemas import validator as validator_module from polylogue.schemas.packages import SchemaResolution @@ -23,7 +24,7 @@ from polylogue.schemas.synthetic.wire_formats import UnsupportedSyntheticWireRouteError from polylogue.schemas.validator import SchemaValidator from polylogue.sources import dispatch as dispatch_module -from polylogue.sources.parsers.base_models import ParsedMessage, ParsedSession +from polylogue.sources.parsers.base_models import ParsedContentBlock, ParsedMessage, ParsedSession from polylogue.sources.source_parsing import iter_antigravity_language_server_sessions @@ -33,15 +34,90 @@ def test_every_catalog_provider_has_an_explicit_route_and_receipt_counts() -> No assert set(receipt.catalog_providers) == set(registry.list_providers()) assert not receipt.missing_routes + catalog_entries: list[tuple[str, str, str, bool, str]] = [] + for provider in registry.list_providers(): + catalog = registry.load_package_catalog(provider) + assert catalog is not None + for package in catalog.packages: + for element in package.elements: + catalog_entries.append( + ( + provider, + package.version, + element.element_kind, + element.supported, + wire_formats.PROVIDER_WIRE_ROUTES[provider].status, + ) + ) + assert {(entry.provider, entry.package_version, entry.element_kind) for entry in receipt.entries} == { + (provider, version, element) for provider, version, element, *_rest in catalog_entries + } assert receipt.supported_count == sum( - route.status == "supported" for route in wire_formats.PROVIDER_WIRE_ROUTES.values() + supported and route_status == "supported" + for _provider, _version, _element, supported, route_status in catalog_entries ) assert receipt.unsupported_count == sum( - route.status == "unsupported" for route in wire_formats.PROVIDER_WIRE_ROUTES.values() + not supported or route_status == "unsupported" + for _provider, _version, _element, supported, route_status in catalog_entries ) assert all(entry.reason for entry in receipt.entries if entry.status == "unsupported") +def test_support_receipt_does_not_substitute_a_default_selection() -> None: + registry = SchemaRegistry() + receipt = wire_formats.build_wire_support_receipt(registry=registry) + + assert all(entry.package_version is not None and entry.element_kind is not None for entry in receipt.entries) + assert len(receipt.entries) > len(receipt.catalog_providers) + + +def test_support_receipt_preserves_an_explicitly_empty_provider_selection() -> None: + receipt = wire_formats.build_wire_support_receipt(registry=SchemaRegistry(), providers=()) + + assert receipt.catalog_providers == () + assert receipt.entries == () + assert receipt.missing_routes == () + + +def test_support_receipt_deduplicates_sorted_provider_selection() -> None: + registry = SchemaRegistry() + available = registry.list_providers() + providers = (available[1], available[0], available[1]) + + receipt = wire_formats.build_wire_support_receipt(registry=registry, providers=providers) + + assert receipt.catalog_providers == tuple(sorted(set(providers))) + assert len({(entry.provider, entry.package_version, entry.element_kind) for entry in receipt.entries}) == len( + receipt.entries + ) + + +def test_support_receipt_counts_a_missing_route_before_skipping_unsupported_elements( + monkeypatch: pytest.MonkeyPatch, +) -> None: + registry = SchemaRegistry() + provider = next(name for name in registry.list_providers() if name in wire_formats.PROVIDER_WIRE_ROUTES) + catalog = registry.load_package_catalog(provider) + assert catalog is not None + unsupported_catalog = replace( + catalog, + packages=[ + replace(package, elements=[replace(element, supported=False) for element in package.elements]) + for package in catalog.packages + ], + ) + + monkeypatch.setattr(registry, "load_package_catalog", lambda _provider: unsupported_catalog) + monkeypatch.delitem(wire_formats.PROVIDER_WIRE_ROUTES, provider) + + receipt = wire_formats.build_wire_support_receipt(registry=registry, providers=(provider,)) + + assert receipt.missing_routes == (provider,) + assert receipt.entries + assert all(entry.reason == wire_formats.CATALOG_ELEMENT_UNSUPPORTED_REASON for entry in receipt.entries) + assert not receipt.complete + + def test_supported_routes_validate_selected_schema_and_parser_entry_point() -> None: receipt = wire_formats.build_wire_support_receipt(registry=SchemaRegistry()) @@ -50,13 +126,21 @@ def test_supported_routes_validate_selected_schema_and_parser_entry_point() -> N assert all(entry.schema_valid is True for entry in supported) assert all(entry.parsed_session_count > 0 for entry in supported) assert all(entry.parsed_message_count > 0 for entry in supported) - assert all(entry.construct_coverage is not None and entry.construct_coverage.complete for entry in supported) + complete_supported = [ + entry for entry in supported if not (entry.provider == "chatgpt" and entry.package_version == "v1") + ] + assert all( + entry.construct_coverage is not None and entry.construct_coverage.complete for entry in complete_supported + ) assert all( any(witness.artifact_kind == "baseline" and witness.healthy for witness in entry.parser_witnesses) for entry in supported ) assert all(all(witness.artifact_evidence for witness in entry.parser_witnesses) for entry in supported) - assert receipt.complete + assert not receipt.complete + chatgpt_v1 = next(entry for entry in supported if entry.provider == "chatgpt" and entry.package_version == "v1") + assert chatgpt_v1.construct_coverage is not None + assert not chatgpt_v1.construct_coverage.complete def test_parser_witness_loss_is_not_masked_by_aggregate_parsed_counts(monkeypatch: pytest.MonkeyPatch) -> None: @@ -95,7 +179,459 @@ def drop_first_coverage_witness( assert not receipt.complete -@pytest.mark.parametrize("returned_session", ["empty", "unrelated", "metadata"]) +def test_parser_witness_partial_output_is_not_accepted_as_complete(monkeypatch: pytest.MonkeyPatch) -> None: + original_parse_payload = dispatch_module.parse_payload + + def return_only_first_message( + provider: str, + payload: object, + fallback_id: str, + _depth: int = 0, + *, + schema_resolution: SchemaResolution | None = None, + source_path: str | None = None, + ) -> list[ParsedSession]: + sessions = original_parse_payload( + provider, + payload, + fallback_id, + _depth, + schema_resolution=schema_resolution, + source_path=source_path, + ) + if provider == "chatgpt" and fallback_id.endswith(":0"): + return [session.model_copy(update={"messages": session.messages[:1]}) for session in sessions] + return sessions + + monkeypatch.setattr(dispatch_module, "parse_payload", return_only_first_message) + receipt = wire_formats.build_wire_support_receipt(registry=SchemaRegistry(), providers=("chatgpt",)) + + entry = next(item for item in receipt.entries if item.package_version == "v1") + baseline = next(item for item in entry.parser_witnesses if item.artifact_kind == "baseline") + assert baseline.parsed_message_count == 1 + assert baseline.artifact_evidence == () + assert baseline.validation_error == "artifact message coverage is incomplete" + assert not baseline.healthy + assert not entry.healthy + assert not receipt.complete + + +@pytest.mark.parametrize("provider", sorted(wire_formats.PROVIDER_WIRE_FORMATS)) +def test_parser_witness_content_loss_is_not_accepted_with_preserved_ids_for_every_supported_route( + monkeypatch: pytest.MonkeyPatch, + provider: str, +) -> None: + original_parse_payload = dispatch_module.parse_payload + + def replace_all_but_one_message_body( + parsed_provider: str, + payload: object, + fallback_id: str, + _depth: int = 0, + *, + schema_resolution: SchemaResolution | None = None, + source_path: str | None = None, + ) -> list[ParsedSession]: + sessions = original_parse_payload( + parsed_provider, + payload, + fallback_id, + _depth, + schema_resolution=schema_resolution, + source_path=source_path, + ) + if parsed_provider == provider and fallback_id.endswith(":0"): + stripped_sessions: list[ParsedSession] = [] + for session in sessions: + retained_body = False + messages: list[ParsedMessage] = [] + for message in session.messages: + has_body = any( + isinstance(text, str) and text.strip() + for text in (message.text, *(block.text for block in message.blocks)) + ) + if has_body and not retained_body: + retained_body = True + messages.append(message) + else: + messages.append(message.model_copy(update={"text": "", "blocks": []})) + stripped_sessions.append(session.model_copy(update={"messages": messages})) + return stripped_sessions + return sessions + + monkeypatch.setattr(dispatch_module, "parse_payload", replace_all_but_one_message_body) + receipt = wire_formats.build_wire_support_receipt(registry=SchemaRegistry(), providers=(provider,)) + + supported_entries = [entry for entry in receipt.entries if entry.status == "supported"] + assert supported_entries + for entry in supported_entries: + baseline = next(item for item in entry.parser_witnesses if item.artifact_kind == "baseline") + assert baseline.parsed_message_count > 0 + assert not baseline.artifact_evidence + assert baseline.validation_error == "artifact message coverage is incomplete" + assert not baseline.healthy + assert not entry.healthy + assert not receipt.complete + + +def test_parser_witness_segment_loss_is_not_accepted_with_preserved_message_identity( + monkeypatch: pytest.MonkeyPatch, +) -> None: + original_parse_payload = dispatch_module.parse_payload + inserted_segment = False + dropped_segment = False + segment = "parser-witness-segment-that-must-survive" + + def drop_one_text_segment( + provider: str, + payload: object, + fallback_id: str, + _depth: int = 0, + *, + schema_resolution: SchemaResolution | None = None, + source_path: str | None = None, + ) -> list[ParsedSession]: + nonlocal inserted_segment, dropped_segment + if provider == "codex" and fallback_id.endswith(":0") and isinstance(payload, list): + for record in payload: + if not isinstance(record, dict) or not isinstance(record.get("content"), list): + continue + record["content"].append({"type": "input_text", "text": segment}) + inserted_segment = True + break + sessions = original_parse_payload( + provider, + payload, + fallback_id, + _depth, + schema_resolution=schema_resolution, + source_path=source_path, + ) + if provider != "codex" or not fallback_id.endswith(":0"): + return sessions + corrupted_sessions: list[ParsedSession] = [] + for parsed_session in sessions: + corrupted_messages: list[ParsedMessage] = [] + for message in parsed_session.messages: + blocks = [] + for block in message.blocks: + if block.text == segment: + dropped_segment = True + blocks.append(block.model_copy(update={"text": ""})) + else: + blocks.append(block) + corrupted_messages.append( + message.model_copy(update={"text": (message.text or "").replace(segment, ""), "blocks": blocks}) + ) + corrupted_sessions.append(parsed_session.model_copy(update={"messages": corrupted_messages})) + return corrupted_sessions + + monkeypatch.setattr(dispatch_module, "parse_payload", drop_one_text_segment) + receipt = wire_formats.build_wire_support_receipt(registry=SchemaRegistry(), providers=("codex",)) + + assert inserted_segment + assert dropped_segment + assert not receipt.complete + assert any( + witness.validation_error == "artifact message coverage is incomplete" + for entry in receipt.entries + for witness in entry.parser_witnesses + if witness.artifact_kind == "baseline" + ) + + +@pytest.mark.parametrize("provider", sorted(wire_formats.PROVIDER_WIRE_FORMATS)) +@pytest.mark.parametrize("artifact_index", (0, 1)) +def test_parser_witness_authoredness_loss_is_not_accepted_for_every_supported_route( + monkeypatch: pytest.MonkeyPatch, + provider: str, + artifact_index: int, +) -> None: + original_parse_payload = dispatch_module.parse_payload + mutated_messages = 0 + + def corrupt_authoredness( + parsed_provider: str, + payload: object, + fallback_id: str, + _depth: int = 0, + *, + schema_resolution: SchemaResolution | None = None, + source_path: str | None = None, + ) -> list[ParsedSession]: + nonlocal mutated_messages + sessions = original_parse_payload( + parsed_provider, + payload, + fallback_id, + _depth, + schema_resolution=schema_resolution, + source_path=source_path, + ) + if parsed_provider != provider or not fallback_id.endswith(f":{artifact_index}"): + return sessions + corrupted_sessions: list[ParsedSession] = [] + for session in sessions: + corrupted_messages: list[ParsedMessage] = [] + for message in session.messages: + mutated_messages += 1 + corrupted_messages.append( + message.model_copy( + update={ + "role": Role.USER if message.role is Role.ASSISTANT else Role.ASSISTANT, + "material_origin": ( + MaterialOrigin.HUMAN_AUTHORED + if message.material_origin is MaterialOrigin.ASSISTANT_AUTHORED + else MaterialOrigin.ASSISTANT_AUTHORED + ), + } + ) + ) + corrupted_sessions.append(session.model_copy(update={"messages": corrupted_messages})) + return corrupted_sessions + + monkeypatch.setattr(dispatch_module, "parse_payload", corrupt_authoredness) + receipt = wire_formats.build_wire_support_receipt(registry=SchemaRegistry(), providers=(provider,)) + + assert mutated_messages > 0 + supported_entries = [entry for entry in receipt.entries if entry.status == "supported"] + assert supported_entries + assert any( + witness.validation_error == "artifact message coverage is incomplete" + for entry in supported_entries + for witness in entry.parser_witnesses + if witness.index == (-1 if artifact_index == 0 else artifact_index - 1) + ) + assert not receipt.complete + + +@pytest.mark.parametrize("provider", ("claude-code", "codex")) +@pytest.mark.parametrize("artifact_index", (0, 1)) +def test_parser_witness_tool_identity_loss_is_not_accepted( + monkeypatch: pytest.MonkeyPatch, + provider: str, + artifact_index: int, +) -> None: + original_parse_payload = dispatch_module.parse_payload + mutated_blocks = 0 + + def drop_tool_identity( + parsed_provider: str, + payload: object, + fallback_id: str, + _depth: int = 0, + *, + schema_resolution: SchemaResolution | None = None, + source_path: str | None = None, + ) -> list[ParsedSession]: + nonlocal mutated_blocks + sessions = original_parse_payload( + parsed_provider, + payload, + fallback_id, + _depth, + schema_resolution=schema_resolution, + source_path=source_path, + ) + if parsed_provider != provider or not fallback_id.endswith(f":{artifact_index}"): + return sessions + corrupted_sessions: list[ParsedSession] = [] + for session in sessions: + corrupted_messages: list[ParsedMessage] = [] + for message in session.messages: + corrupted_blocks: list[ParsedContentBlock] = [] + for block in message.blocks: + if block.type in {BlockType.TOOL_USE, BlockType.TOOL_RESULT} and block.tool_id: + mutated_blocks += 1 + corrupted_blocks.append(block.model_copy(update={"tool_id": None})) + else: + corrupted_blocks.append(block) + corrupted_messages.append(message.model_copy(update={"blocks": corrupted_blocks})) + corrupted_sessions.append(session.model_copy(update={"messages": corrupted_messages})) + return corrupted_sessions + + monkeypatch.setattr(dispatch_module, "parse_payload", drop_tool_identity) + receipt = wire_formats.build_wire_support_receipt(registry=SchemaRegistry(), providers=(provider,)) + + assert mutated_blocks > 0 + assert not receipt.complete + assert any( + witness.validation_error == "artifact message coverage is incomplete" + for entry in receipt.entries + for witness in entry.parser_witnesses + ) + + +@pytest.mark.parametrize( + ("provider", "artifact_index"), + (("claude-code", 0), ("claude-code", 2), ("codex", 0), ("codex", 1)), +) +def test_parser_witness_structured_tool_outcome_loss_is_not_accepted( + monkeypatch: pytest.MonkeyPatch, + provider: str, + artifact_index: int, +) -> None: + original_parse_payload = dispatch_module.parse_payload + mutated_outcomes = 0 + + def corrupt_structured_tool_outcome( + parsed_provider: str, + payload: object, + fallback_id: str, + _depth: int = 0, + *, + schema_resolution: SchemaResolution | None = None, + source_path: str | None = None, + ) -> list[ParsedSession]: + nonlocal mutated_outcomes + if parsed_provider == provider and fallback_id.endswith(f":{artifact_index}") and isinstance(payload, list): + injected_outcome = False + for record in payload: + if not isinstance(record, dict): + continue + container = record.get("message") if provider == "claude-code" else record + content = container.get("content") if isinstance(container, dict) else None + if not isinstance(content, list): + continue + for item in content: + if isinstance(item, dict) and item.get("type") == "tool_result": + item["is_error"] = False + injected_outcome = True + break + if injected_outcome: + break + if not injected_outcome: + for record in payload: + if not isinstance(record, dict): + continue + container = record.get("message") if provider == "claude-code" else record + content = container.get("content") if isinstance(container, dict) else None + if not isinstance(content, list): + continue + tool_use = next( + ( + item + for item in content + if isinstance(item, dict) + and item.get("type") == "tool_use" + and isinstance(item.get("id"), str) + ), + None, + ) + if isinstance(tool_use, dict): + content.append( + { + "type": "tool_result", + "tool_use_id": tool_use["id"], + "content": "synthetic structured outcome", + "is_error": False, + } + ) + break + sessions = original_parse_payload( + parsed_provider, + payload, + fallback_id, + _depth, + schema_resolution=schema_resolution, + source_path=source_path, + ) + if parsed_provider != provider or not fallback_id.endswith(f":{artifact_index}"): + return sessions + corrupted_sessions: list[ParsedSession] = [] + for session in sessions: + corrupted_messages: list[ParsedMessage] = [] + for message in session.messages: + corrupted_blocks: list[ParsedContentBlock] = [] + for block in message.blocks: + if block.type is BlockType.TOOL_RESULT and ( + block.is_error is not None or block.exit_code is not None + ): + mutated_outcomes += 1 + corrupted_blocks.append( + block.model_copy( + update={ + "is_error": (not block.is_error) if block.is_error is not None else True, + "exit_code": (block.exit_code or 0) + 1 if block.exit_code is not None else None, + } + ) + ) + else: + corrupted_blocks.append(block) + corrupted_messages.append(message.model_copy(update={"blocks": corrupted_blocks})) + corrupted_sessions.append(session.model_copy(update={"messages": corrupted_messages})) + return corrupted_sessions + + monkeypatch.setattr(dispatch_module, "parse_payload", corrupt_structured_tool_outcome) + receipt = wire_formats.build_wire_support_receipt(registry=SchemaRegistry(), providers=(provider,)) + + assert mutated_outcomes > 0 + assert not receipt.complete + assert any( + witness.validation_error == "artifact message coverage is incomplete" + for entry in receipt.entries + for witness in entry.parser_witnesses + ) + + +@pytest.mark.parametrize("provider", sorted(wire_formats.PROVIDER_WIRE_FORMATS)) +def test_parser_witness_rejects_messages_rehomed_to_another_raw_identity( + monkeypatch: pytest.MonkeyPatch, + provider: str, +) -> None: + original_parse_payload = dispatch_module.parse_payload + rehomed_sessions = 0 + + def rehome_to_message_id( + parsed_provider: str, + payload: object, + fallback_id: str, + _depth: int = 0, + *, + schema_resolution: SchemaResolution | None = None, + source_path: str | None = None, + ) -> list[ParsedSession]: + nonlocal rehomed_sessions + sessions = original_parse_payload( + parsed_provider, + payload, + fallback_id, + _depth, + schema_resolution=schema_resolution, + source_path=source_path, + ) + if parsed_provider != provider or not fallback_id.endswith(":0"): + return sessions + corrupted_sessions: list[ParsedSession] = [] + for session in sessions: + foreign_raw_identity = next( + ( + message.provider_message_id or message.text + for message in session.messages + if message.provider_message_id or (isinstance(message.text, str) and message.text) + ), + None, + ) + if foreign_raw_identity is None or foreign_raw_identity == session.provider_session_id: + corrupted_sessions.append(session) + continue + rehomed_sessions += 1 + corrupted_sessions.append(session.model_copy(update={"provider_session_id": foreign_raw_identity})) + return corrupted_sessions + + monkeypatch.setattr(dispatch_module, "parse_payload", rehome_to_message_id) + receipt = wire_formats.build_wire_support_receipt(registry=SchemaRegistry(), providers=(provider,)) + + assert rehomed_sessions > 0 + assert not receipt.complete + assert any( + witness.artifact_kind == "baseline" and witness.validation_error == "artifact message coverage is incomplete" + for entry in receipt.entries + for witness in entry.parser_witnesses + ) + + +@pytest.mark.parametrize("returned_session", ["empty", "unrelated", "metadata", "id_only"]) def test_parser_witness_requires_meaningful_evidence_from_its_own_artifact( monkeypatch: pytest.MonkeyPatch, returned_session: str, @@ -131,6 +667,27 @@ def return_non_evidence_for_first_coverage_witness( ], ) ] + if returned_session == "id_only": + payload_record = payload if isinstance(payload, Mapping) else {} + metadata_id = str(payload_record.get("id", "metadata-session")) + return [ + ParsedSession( + source_name=Provider.CHATGPT, + provider_session_id=metadata_id, + messages=[ + ParsedMessage( + provider_message_id=metadata_id, + role=Role.ASSISTANT, + blocks=[ + ParsedContentBlock( + type=BlockType.TEXT, + text="invented content absent from this artifact", + ) + ], + ) + ], + ) + ] return [ ParsedSession( source_name=Provider.CHATGPT, @@ -498,6 +1055,52 @@ def test_claude_code_route_only_waives_unrepresentable_nested_content() -> None: ) +def test_chatgpt_v1_media_waiver_does_not_hide_parser_relevant_omissions() -> None: + registry = SchemaRegistry() + selection = select_synthetic_schema("chatgpt", version="v1", registry_factory=lambda: registry) + corpus = SyntheticCorpus.from_selection(selection) + payload = corpus.generate_batch(count=1, messages_per_session=range(4, 5), seed=20260805).raw_items[0] + payloads: tuple[JSONValue, ...] = (json.loads(payload),) + witnessed = wire_formats.construct_coverage(selection.schema, payloads) + parser_relevant = next( + keyword + for keyword in witnessed.missing_keywords + if keyword.startswith("type:null@") and ".properties.message.anyOf[0]" in keyword + ) + reasons = wire_formats._route_nonrepresentable_reasons( + "chatgpt", + witnessed.missing_keywords, + package_version="v1", + ) + media = next( + keyword + for keyword in witnessed.missing_keywords + if ".properties.content.properties.parts.items[*].anyOf[1]" in keyword + ) + + assert media in reasons + assert parser_relevant not in reasons + final = wire_formats.construct_coverage( + selection.schema, + payloads, + nonrepresentable_keywords=reasons, + nonrepresentable_reasons=reasons, + ) + assert parser_relevant in final.missing_keywords + assert not final.complete + + receipt = wire_formats.build_wire_support_receipt(registry=registry) + receipt_entry = next( + entry + for entry in receipt.entries + if (entry.provider, entry.package_version, entry.element_kind) == ("chatgpt", "v1", selection.element_kind) + ) + assert receipt_entry.construct_coverage is not None + assert parser_relevant in receipt_entry.construct_coverage.missing_keywords + assert not receipt_entry.healthy + assert not receipt.complete + + def test_unmatched_union_does_not_count_as_exercised() -> None: schema: SchemaRecord = { "oneOf": [{"type": "integer"}, {"type": "string"}], diff --git a/tests/unit/daemon/test_convergence_stages.py b/tests/unit/daemon/test_convergence_stages.py index 8fa4efcc39..b2db41ecda 100644 --- a/tests/unit/daemon/test_convergence_stages.py +++ b/tests/unit/daemon/test_convergence_stages.py @@ -7,6 +7,7 @@ import time from collections.abc import Iterator from contextlib import contextmanager +from datetime import UTC, datetime from pathlib import Path from types import SimpleNamespace from typing import cast @@ -33,7 +34,9 @@ ) from polylogue.sources.parsers.base import ParsedContentBlock, ParsedMessage, ParsedSession from polylogue.storage.insights.session import storage as session_storage +from polylogue.storage.insights.session.repair_assessment import session_insight_status_ready from polylogue.storage.insights.session.runtime import SessionInsightCounts +from polylogue.storage.insights.session.status import session_insight_status_sync from polylogue.storage.runtime import SESSION_INSIGHT_MATERIALIZER_VERSION from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root, initialize_archive_tier @@ -1678,7 +1681,7 @@ def test_insights_stage_scopes_session_debt_to_stale_profiles(tmp_path: Path) -> def test_archive_insights_execute_ids_preserves_millisecond_sort_key(tmp_path: Path) -> None: db_path = tmp_path / "index.db" session_id = "codex-session:conv-ms" - source_sort_key_ms = 1_779_606_000_953 + source_sort_key_ms = 1_097_440_214_212 with open_connection(db_path) as conn: _seed_index_session(conn, session_id="conv-ms", text="Message with millisecond sort key") conn.execute( @@ -1697,9 +1700,94 @@ def test_archive_insights_execute_ids_preserves_millisecond_sort_key(tmp_path: P "SELECT source_sort_key FROM session_profiles WHERE session_id = ?", (session_id,), ).fetchone() + latency_materialization = conn.execute( + """ + SELECT source_sort_key_ms + FROM insight_materialization + WHERE session_id = ? AND insight_type = 'latency' + """, + (session_id,), + ).fetchone() assert profile is not None assert profile["source_sort_key"] == pytest.approx(source_sort_key_ms / 1000.0) + assert latency_materialization is not None + assert latency_materialization["source_sort_key_ms"] == source_sort_key_ms + assert stages._archive_stale_session_profile_ids(conn, [session_id]) == [] + + +def test_archive_insights_execute_ids_propagates_provider_high_water_mark(tmp_path: Path) -> None: + db_path = tmp_path / "index.db" + session_id = "codex-session:conv-provider-hwm" + provider_hwm_ms = 1_779_606_000_953 + with open_connection(db_path) as conn: + _seed_index_session(conn, session_id="conv-provider-hwm", text="Provider timestamp session") + conn.execute( + "UPDATE sessions SET updated_at_ms = ? WHERE session_id = ?", + (provider_hwm_ms, session_id), + ) + conn.commit() + + assert stages._archive_insights_execute_ids(conn, [session_id]) + + latency = conn.execute( + """ + SELECT input_high_water_mark, input_high_water_mark_source + FROM session_latency_profiles + WHERE session_id = ? + """, + (session_id,), + ).fetchone() + materialization = conn.execute( + """ + SELECT input_high_water_mark_ms, input_high_water_mark_source + FROM insight_materialization + WHERE session_id = ? AND insight_type = 'latency' + """, + (session_id,), + ).fetchone() + + assert latency is not None + assert latency["input_high_water_mark"] == datetime.fromtimestamp(provider_hwm_ms / 1000, tz=UTC).isoformat() + assert latency["input_high_water_mark_source"] == "provider_ts" + assert materialization is not None + assert materialization["input_high_water_mark_ms"] == provider_hwm_ms + assert materialization["input_high_water_mark_source"] == "provider_ts" + + +def test_archive_insights_created_without_updated_stays_ready_after_materialization(tmp_path: Path) -> None: + db_path = tmp_path / "index.db" + session_id = "codex-session:conv-created-only" + created_at_ms = 1_779_606_000_953 + with open_connection(db_path) as conn: + _seed_index_session(conn, session_id="conv-created-only", text="Created-only session") + conn.execute( + "UPDATE sessions SET created_at_ms = ?, updated_at_ms = NULL WHERE session_id = ?", + (created_at_ms, session_id), + ) + conn.commit() + + assert stages._archive_insights_execute_ids(conn, [session_id]) + + latency = conn.execute( + "SELECT source_updated_at, source_sort_key FROM session_latency_profiles WHERE session_id = ?", + (session_id,), + ).fetchone() + assert latency is not None + assert latency["source_updated_at"] is None + assert latency["source_sort_key"] == pytest.approx(created_at_ms / 1000.0) + materialization = conn.execute( + """ + SELECT source_updated_at_ms, source_sort_key_ms + FROM insight_materialization + WHERE session_id = ? AND insight_type = 'latency' + """, + (session_id,), + ).fetchone() + assert materialization is not None + assert materialization["source_updated_at_ms"] is None + assert materialization["source_sort_key_ms"] == created_at_ms assert stages._archive_stale_session_profile_ids(conn, [session_id]) == [] + assert session_insight_status_ready(session_insight_status_sync(conn)) def test_archive_insights_execute_ids_deduplicates_session_ids(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: diff --git a/tests/unit/schemas/test_inferred_corpus_manifest.py b/tests/unit/schemas/test_inferred_corpus_manifest.py index 94086a4488..24ad12a1bf 100644 --- a/tests/unit/schemas/test_inferred_corpus_manifest.py +++ b/tests/unit/schemas/test_inferred_corpus_manifest.py @@ -1,13 +1,14 @@ from __future__ import annotations import json +from collections.abc import Sequence from dataclasses import replace from pathlib import Path -from typing import cast +from typing import Any, cast import pytest -from polylogue.core.json import JSONValue +from polylogue.core.json import JSONDocument, JSONValue from polylogue.maintenance.schema_inference_gate import ( run_schema_inference_gate, schema_inference_gate_receipt_digest, @@ -18,7 +19,15 @@ ) from polylogue.schemas.registry import SCHEMA_DIR, SchemaRegistry from polylogue.schemas.synthetic.models import SchemaRecord -from polylogue.schemas.synthetic.wire_formats import PROVIDER_WIRE_FORMATS +from polylogue.schemas.synthetic.wire_formats import ( + PROVIDER_WIRE_FORMATS, + PROVIDER_WIRE_ROUTES, + WireSupportEntry, + WireSupportReceipt, + build_wire_support_receipt, +) +from polylogue.sources.parsers.base_models import ParsedSession +from tests.infra import inferred_corpus as inferred_corpus_module from tests.infra.inferred_corpus import ( CorpusManifestKey, InferredCorpusManifest, @@ -65,9 +74,10 @@ def __init__(self, base: SchemaRegistry) -> None: self.base = base self.catalog_overrides: dict[str, object] = {} self.schema_overrides: dict[tuple[str, str, str], object] = {} + self.provider_order: list[str] | None = None def list_providers(self) -> list[str]: - return self.base.list_providers() + return self.provider_order if self.provider_order is not None else self.base.list_providers() def load_package_catalog(self, provider: str) -> object: return self.catalog_overrides.get(provider, self.base.load_package_catalog(provider)) @@ -110,6 +120,389 @@ def test_persisted_manifest_round_trip_validates_identity_and_integrity(tmp_path assert read_inferred_corpus_manifest(path) == manifest +def test_manifest_can_bind_every_selection_to_the_exact_wire_support_receipt() -> None: + registry = _registry() + support = build_wire_support_receipt(registry=registry) + + manifest = compile_inferred_corpus_manifest(registry=registry, wire_support_receipt=support) + + assert manifest.entries + assert all(entry.unsupported is None for entry in manifest.entries if entry.spec is not None) + assert {(entry.key.provider, entry.key.package_version, entry.key.element_kind) for entry in manifest.entries} == { + (entry.provider, entry.package_version, entry.element_kind) for entry in support.entries + } + + +def test_wire_support_receipt_is_canonical_across_catalog_reordering() -> None: + registry = _registry() + reordered = _RegistryProxy(registry) + reordered.provider_order = list(reversed(registry.list_providers())) + for provider in registry.list_providers(): + catalog = registry.load_package_catalog(provider) + assert catalog is not None + reordered.catalog_overrides[provider] = replace( + catalog, + packages=[ + replace(package, elements=list(reversed(package.elements))) for package in reversed(catalog.packages) + ], + ) + + assert ( + build_wire_support_receipt(registry=registry).to_dict() + == build_wire_support_receipt(registry=reordered).to_dict() + ) + + +def test_wire_support_receipt_rejects_conflicting_duplicate_identity_at_all_boundaries() -> None: + registry = _registry() + receipt = build_wire_support_receipt(registry=registry, providers=("codex",)) + original = receipt.entries[0] + conflicting = replace(original, reason="conflicting duplicate") + + with pytest.raises(ValueError, match="duplicate wire support entry key"): + WireSupportReceipt( + catalog_providers=receipt.catalog_providers, + entries=(original, conflicting), + missing_routes=receipt.missing_routes, + witness_seed=receipt.witness_seed, + ) + + malformed = object.__new__(WireSupportReceipt) + object.__setattr__(malformed, "catalog_providers", receipt.catalog_providers) + object.__setattr__(malformed, "entries", (original, conflicting)) + object.__setattr__(malformed, "missing_routes", receipt.missing_routes) + object.__setattr__(malformed, "witness_seed", receipt.witness_seed) + with pytest.raises(ValueError, match="duplicate wire support entry key"): + compile_inferred_corpus_manifest(registry=registry, wire_support_receipt=malformed) + + manifest = compile_inferred_corpus_manifest(registry=registry, wire_support_receipt=receipt) + assert manifest.wire_support_receipt is not None + persisted_receipt = cast(dict[str, object], dict(manifest.wire_support_receipt)) + persisted_entries = list(cast(list[dict[str, object]], persisted_receipt["entries"])) + persisted_entries.append(dict(persisted_entries[0], reason="conflicting duplicate")) + persisted_receipt["entries"] = persisted_entries + persisted_manifest = replace(manifest, wire_support_receipt=cast(JSONDocument, persisted_receipt)) + + with pytest.raises(ValueError, match="duplicate wire support entry key"): + inferred_corpus_module._wire_support_entry_index(persisted_manifest) + with pytest.raises(ValueError, match="duplicate wire support entry key"): + InferredCorpusManifest.from_payload(persisted_manifest.to_payload()) + + +def test_all_provider_campaign_round_trip_preserves_unsupported_wire_authority(tmp_path: Path) -> None: + registry = _registry() + archive_root, gate_receipt_path, gate_digest = _authoritative_gate(tmp_path) + package_receipts = [ + build_schema_inference_receipt(cast(Any, registry), provider=provider, gate_receipt_digest=gate_digest) + for provider in registry.list_providers() + ] + package_receipt = package_receipts[0] + for other in package_receipts[1:]: + package_receipt = package_receipt.merged_with(other) + wire_support = build_wire_support_receipt(registry=registry) + + manifest = compile_inferred_corpus_manifest( + registry=cast(Any, registry), + package_receipt=package_receipt.to_payload(), + wire_support_receipt=wire_support, + campaign_mode=True, + gate_receipt_path=gate_receipt_path, + archive_root=archive_root, + ) + + antigravity_entries = [entry for entry in manifest.entries if entry.key.provider == "antigravity"] + assert antigravity_entries + assert all(entry.unsupported is not None for entry in antigravity_entries) + assert all( + entry.unsupported.reason == "unsupported_wire_route" for entry in antigravity_entries if entry.unsupported + ) + + path = tmp_path / "all-provider-campaign.json" + write_inferred_corpus_manifest(manifest, path) + restored = read_inferred_corpus_manifest( + path, + campaign_mode=True, + registry=registry, + gate_receipt_path=gate_receipt_path, + archive_root=archive_root, + ) + assert restored == manifest + assert restored.wire_support_receipt == wire_support.to_dict() + + +def test_default_scope_campaign_rejects_a_new_provider_during_receipt_revalidation(tmp_path: Path) -> None: + base_registry = _registry() + registry = _RegistryProxy(base_registry) + registry.provider_order = ["codex"] + archive_root, gate_receipt_path, gate_digest = _authoritative_gate(tmp_path) + providers = tuple(registry.list_providers()) + package_receipts = [ + build_schema_inference_receipt(cast(Any, registry), provider=provider, gate_receipt_digest=gate_digest) + for provider in providers + ] + package_receipt = package_receipts[0] + for other in package_receipts[1:]: + package_receipt = package_receipt.merged_with(other) + wire_support = build_wire_support_receipt(registry=registry) + assert wire_support.catalog_scope == "registry-default" + manifest = compile_inferred_corpus_manifest( + registry=cast(Any, registry), + package_receipt=package_receipt.to_payload(), + wire_support_receipt=wire_support, + campaign_mode=True, + gate_receipt_path=gate_receipt_path, + archive_root=archive_root, + ) + path = tmp_path / "all-provider-campaign.json" + write_inferred_corpus_manifest(manifest, path) + + registry.provider_order = ["codex", "new-unrouted-provider"] + + with pytest.raises(ValueError, match=r"wire-support receipt changed.*catalog_providers.*missing_routes"): + read_inferred_corpus_manifest( + path, + campaign_mode=True, + registry=cast(Any, registry), + gate_receipt_path=gate_receipt_path, + archive_root=archive_root, + ) + + +def test_explicit_scope_campaign_does_not_re_census_unselected_provider(tmp_path: Path) -> None: + base_registry = _registry() + registry = _RegistryProxy(base_registry) + archive_root, gate_receipt_path, gate_digest = _authoritative_gate(tmp_path) + package_receipt = build_schema_inference_receipt( + cast(Any, registry), provider="codex", gate_receipt_digest=gate_digest + ) + wire_support = build_wire_support_receipt(registry=registry, providers=("codex",)) + assert wire_support.catalog_scope == "explicit" + manifest = compile_inferred_corpus_manifest( + registry=cast(Any, registry), + providers=("codex",), + package_receipt=package_receipt.to_payload(), + wire_support_receipt=wire_support, + campaign_mode=True, + gate_receipt_path=gate_receipt_path, + archive_root=archive_root, + ) + path = tmp_path / "codex-campaign.json" + write_inferred_corpus_manifest(manifest, path) + + registry.provider_order = [*base_registry.list_providers(), "new-unrouted-provider"] + + assert ( + read_inferred_corpus_manifest( + path, + campaign_mode=True, + registry=cast(Any, registry), + gate_receipt_path=gate_receipt_path, + archive_root=archive_root, + ) + == manifest + ) + + +def test_campaign_rejects_a_bound_receipt_with_a_missing_catalog_route( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + registry = _registry() + archive_root, gate_receipt_path, gate_digest = _authoritative_gate(tmp_path) + providers = ("claude-ai", "codex") + package_receipt = build_schema_inference_receipt( + registry, + provider=providers[0], + gate_receipt_digest=gate_digest, + ).merged_with( + build_schema_inference_receipt( + registry, + provider=providers[1], + gate_receipt_digest=gate_digest, + ) + ) + monkeypatch.delitem(PROVIDER_WIRE_ROUTES, "codex") + wire_support = build_wire_support_receipt(registry=registry, providers=providers) + + assert wire_support.missing_routes == ("codex",) + assert any(entry.status == "supported" for entry in wire_support.entries) + with pytest.raises(ValueError, match="explicit synthetic wire route"): + compile_inferred_corpus_manifest( + registry=registry, + providers=providers, + package_receipt=package_receipt.to_payload(), + wire_support_receipt=wire_support, + campaign_mode=True, + gate_receipt_path=gate_receipt_path, + archive_root=archive_root, + ) + + +def test_campaign_indexes_persisted_wire_support_entries_once( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + registry = _registry() + archive_root, gate_receipt_path, gate_digest = _authoritative_gate(tmp_path) + package_receipt = build_schema_inference_receipt( + registry, + provider="codex", + gate_receipt_digest=gate_digest, + ) + wire_support = build_wire_support_receipt(registry=registry, providers=("codex",)) + index_calls = 0 + payload_calls = 0 + original_index = inferred_corpus_module._wire_support_entry_index + original_payload = inferred_corpus_module._wire_support_entry_from_payload + + def count_index( + manifest: InferredCorpusManifest, + ) -> dict[tuple[str, str | None, str | None], WireSupportEntry]: + nonlocal index_calls + index_calls += 1 + return original_index(manifest) + + def count_payload(payload: object) -> object: + nonlocal payload_calls + payload_calls += 1 + return original_payload(cast(dict[str, object], payload)) + + monkeypatch.setattr(inferred_corpus_module, "_wire_support_entry_index", count_index) + monkeypatch.setattr(inferred_corpus_module, "_wire_support_entry_from_payload", count_payload) + + compile_inferred_corpus_manifest( + registry=registry, + providers=("codex",), + package_receipt=package_receipt.to_payload(), + wire_support_receipt=wire_support, + campaign_mode=True, + gate_receipt_path=gate_receipt_path, + archive_root=archive_root, + ) + + assert index_calls == 1 + assert payload_calls == len(wire_support.entries) + + +def test_campaign_read_rejects_wire_route_drift(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + registry = _registry() + archive_root, gate_receipt_path, gate_digest = _authoritative_gate(tmp_path) + package_receipt = build_schema_inference_receipt( + registry, + provider="codex", + gate_receipt_digest=gate_digest, + ) + wire_support = build_wire_support_receipt(registry=registry, providers=("codex",)) + manifest = compile_inferred_corpus_manifest( + registry=registry, + package_receipt=package_receipt.to_payload(), + wire_support_receipt=wire_support, + providers=("codex",), + campaign_mode=True, + gate_receipt_path=gate_receipt_path, + archive_root=archive_root, + ) + path = tmp_path / "campaign.json" + write_inferred_corpus_manifest(manifest, path) + + from polylogue.sources import dispatch as dispatch_module + + original_parse_payload = dispatch_module.parse_payload + + def drifted_parse_payload(*args: object, **kwargs: object) -> list[ParsedSession]: + if args and args[0] == "codex": + raise ValueError("simulated parser drift") + return original_parse_payload(*args, **kwargs) # type: ignore[arg-type] + + monkeypatch.setattr(dispatch_module, "parse_payload", drifted_parse_payload) + with pytest.raises(ValueError, match=r"wire-support receipt changed.*changed_fields=.*entries"): + read_inferred_corpus_manifest( + path, + campaign_mode=True, + registry=registry, + gate_receipt_path=gate_receipt_path, + archive_root=archive_root, + ) + + +def test_path_campaign_handoff_replays_current_wire_route_once( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + registry = _registry() + archive_root, gate_receipt_path, gate_digest = _authoritative_gate(tmp_path) + package_receipt = build_schema_inference_receipt( + registry, + provider="codex", + gate_receipt_digest=gate_digest, + ) + wire_support = build_wire_support_receipt(registry=registry, providers=("codex",)) + manifest = compile_inferred_corpus_manifest( + registry=registry, + package_receipt=package_receipt.to_payload(), + wire_support_receipt=wire_support, + providers=("codex",), + campaign_mode=True, + gate_receipt_path=gate_receipt_path, + archive_root=archive_root, + ) + path = tmp_path / "campaign.json" + write_inferred_corpus_manifest(manifest, path) + + original_build_wire_support_receipt = build_wire_support_receipt + replay_count = 0 + + def count_wire_replays( + *, + registry: object | None = None, + seed: int = 20260805, + providers: Sequence[str] | None = None, + ) -> WireSupportReceipt: + nonlocal replay_count + replay_count += 1 + return original_build_wire_support_receipt(registry=registry, seed=seed, providers=providers) + + monkeypatch.setattr("tests.infra.inferred_corpus.build_wire_support_receipt", count_wire_replays) + + handoff = build_inferred_corpus_convergence_handoff( + path, + campaign_mode=True, + registry=registry, + gate_receipt_path=gate_receipt_path, + archive_root=archive_root, + ) + + assert handoff.specs == manifest.supported_specs + assert replay_count == 1 + + +def test_manifest_refuses_a_selection_missing_from_bound_wire_support_receipt() -> None: + registry = _registry() + support = build_wire_support_receipt(registry=registry) + missing = next(entry for entry in support.entries if entry.status == "supported") + reduced_support = replace( + support, + entries=tuple( + entry + for entry in support.entries + if (entry.provider, entry.package_version, entry.element_kind) + != (missing.provider, missing.package_version, missing.element_kind) + ), + ) + + manifest = compile_inferred_corpus_manifest(registry=registry, wire_support_receipt=reduced_support) + + target = next( + entry + for entry in manifest.entries + if (entry.key.provider, entry.key.package_version, entry.key.element_kind) + == (missing.provider, missing.package_version, missing.element_kind) + ) + assert target.spec is None + assert target.unsupported is not None + assert target.unsupported.reason == "wire_support_selection_unwitnessed" + + def test_campaign_read_revalidates_live_schema_and_classifier(tmp_path: Path) -> None: registry = _registry() provider = "codex" @@ -509,6 +902,28 @@ def test_missing_element_schema_becomes_explicit_unsupported_record() -> None: assert target.unsupported.reason == "missing_schema" +def test_missing_element_schema_precedes_missing_wire_format() -> None: + registry = _registry() + proxy = _RegistryProxy(registry) + target_provider = registry.list_providers()[0] + catalog = registry.load_package_catalog(target_provider) + assert catalog is not None + target_package = catalog.packages[0] + target_element = target_package.elements[0] + proxy.schema_overrides[(target_provider, target_package.version, target_element.element_kind)] = None + + manifest = compile_inferred_corpus_manifest(registry=proxy, wire_formats={}) # type: ignore[arg-type] + target = next( + entry + for entry in manifest.entries + if (entry.key.provider, entry.key.package_version, entry.key.element_kind) + == (target_provider, target_package.version, target_element.element_kind) + ) + + assert target.unsupported is not None + assert target.unsupported.reason == "missing_schema" + + def test_catalog_element_marked_unsupported_is_retained_as_a_typed_record() -> None: registry = _registry() proxy = _RegistryProxy(registry) diff --git a/tests/unit/sources/test_live_append_cursor_resynthesis.py b/tests/unit/sources/test_live_append_cursor_resynthesis.py index a25e7d2de8..2ab983e80b 100644 --- a/tests/unit/sources/test_live_append_cursor_resynthesis.py +++ b/tests/unit/sources/test_live_append_cursor_resynthesis.py @@ -90,6 +90,14 @@ def test_append_plan_resynthesizes_lost_cursor_from_durable_full_head(tmp_path: appended = _codex_message("grown-after-reset") source.write_bytes(baseline + appended) _seed_native_session(tmp_path, session_id=session_id) + with sqlite3.connect(tmp_path / "source.db") as conn: + assert ( + conn.execute( + "SELECT COUNT(*) FROM raw_sessions WHERE raw_id = ?", + ("unrelated-raw-id",), + ).fetchone()[0] + == 0 + ) cursor = CursorStore(tmp_path / "ops.db") assert cursor.get_record(source) is None # the ops.db cursor is genuinely gone diff --git a/tests/unit/sources/test_live_batch_support.py b/tests/unit/sources/test_live_batch_support.py index 6b35a1c84e..513df715ce 100644 --- a/tests/unit/sources/test_live_batch_support.py +++ b/tests/unit/sources/test_live_batch_support.py @@ -28,7 +28,12 @@ from polylogue.sources.dispatch import parse_payload from polylogue.sources.live import LiveWatcher, WatchSource from polylogue.sources.live.append_ingest import ingest_append_plans -from polylogue.sources.live.batch import _MAX_APPEND_PLAN_PAYLOAD_BYTES, LiveBatchProcessor, _ArchiveFullWriteResult +from polylogue.sources.live.batch import ( + _MAX_APPEND_PLAN_PAYLOAD_BYTES, + LiveBatchProcessor, + _ArchiveFullWriteResult, + append_capability_receipt, +) from polylogue.sources.live.batch_support import ( _BROWSER_CAPTURE_PREFIX_PROBE_BYTES, _DEFER_APPEND, @@ -50,6 +55,47 @@ from polylogue.storage.raw_failure_lifecycle import read_raw_failure_lifecycle from polylogue.storage.sqlite.archive_tiers import archive as archive_tier_module from polylogue.storage.sqlite.archive_tiers import revision_governance as archive_revision_governance + + +@pytest.mark.parametrize( + ("provider", "stable_session_identity", "status"), + [ + ("codex", False, "unsupported"), + ("codex", True, "supported"), + ("claude-code", False, "unsupported"), + ("claude-code", True, "supported"), + ("chatgpt", True, "unsupported"), + ], +) +def test_append_capability_receipt_is_keyed_to_live_identity_contract( + provider: str, + stable_session_identity: bool, + status: str, +) -> None: + receipt = append_capability_receipt( + provider=provider, + package_version="v1", + element_kind="session_record_stream", + stable_session_identity=stable_session_identity, + ) + + assert receipt.status == status + payload = receipt.to_dict() + assert (payload["provider"], payload["package_version"], payload["element_kind"]) == ( + provider, + "v1", + "session_record_stream", + ) + assert payload["capability_source"] == "LiveBatchProcessor.append" + assert payload["operation"] == "append_prefix" + if provider not in {"codex", "claude-code"}: + assert payload["reason"] == "live append route supports only Codex and Claude Code JSONL identity contracts" + elif not stable_session_identity: + assert payload["reason"] == "append delta requires a stable persisted session identity" + else: + assert payload["reason"] is None + + from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore from polylogue.storage.sqlite.archive_tiers.bootstrap import ( ARCHIVE_TIER_SPECS, @@ -2370,6 +2416,165 @@ def test_codex_append_plan_reads_archive_file_set_session_identity(tmp_path: Pat assert conn.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name='raw_sessions'").fetchone() is None +@pytest.mark.parametrize( + ("index_origin", "source_origin"), + [ + ("codex-session", "claude-code-session"), + ("claude-code-session", "codex-session"), + ], +) +def test_codex_append_identity_rejects_mixed_origins_at_same_path( + tmp_path: Path, + index_origin: str, + source_origin: str, +) -> None: + from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database + from polylogue.storage.sqlite.archive_tiers.source_write import write_source_raw_session + from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier + + root = tmp_path / "sessions" + root.mkdir() + path = root / "shared.jsonl" + payload = b'{"type":"session_meta","payload":{"id":"codex-id"}}\n' + path.write_bytes(payload) + index_db = tmp_path / "index.db" + source_db = tmp_path / "source.db" + initialize_archive_database(index_db, ArchiveTier.INDEX) + initialize_archive_database(source_db, ArchiveTier.SOURCE) + with sqlite3.connect(source_db) as conn: + raw_id = write_source_raw_session( + conn, + origin=source_origin, + source_path=str(path), + source_index=0, + payload=payload, + acquired_at_ms=1_770_000_000_000, + ) + with sqlite3.connect(index_db) as conn: + conn.execute( + "INSERT INTO sessions (native_id, origin, raw_id, title, content_hash) VALUES (?, ?, ?, ?, ?)", + ("codex-id", index_origin, raw_id, "mixed origin", bytes(32)), + ) + conn.commit() + + processor = LiveBatchProcessor( + cast(Any, SimpleNamespace(archive_root=tmp_path, backend=SimpleNamespace(db_path=index_db))), + (WatchSource(name="codex", root=root),), + cursor=CursorStore(index_db), + parser_fingerprint="test-parser", + ) + + assert processor._append_payload_for_provider(path, "codex", b'{"type":"event_msg"}\n') is None + + +def test_codex_append_identity_rejects_mismatched_index_owner_before_global_fallback(tmp_path: Path) -> None: + from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database + from polylogue.storage.sqlite.archive_tiers.source_write import write_source_raw_session + from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier + + root = tmp_path / "sessions" + root.mkdir() + path = root / "shared.jsonl" + payload = b'{"type":"session_meta","payload":{"id":"codex-id"}}\n' + path.write_bytes(payload) + index_db = tmp_path / "index.db" + source_db = tmp_path / "source.db" + initialize_archive_database(index_db, ArchiveTier.INDEX) + initialize_archive_database(source_db, ArchiveTier.SOURCE) + with sqlite3.connect(source_db) as conn: + wrong_owner_raw_id = write_source_raw_session( + conn, + origin="codex-session", + source_path=str(path), + source_index=0, + payload=payload, + acquired_at_ms=1_770_000_000_000, + ) + unrelated_codex_raw_id = write_source_raw_session( + conn, + origin="codex-session", + source_path=str(root / "other.jsonl"), + source_index=0, + payload=payload, + acquired_at_ms=1_770_000_000_001, + ) + with sqlite3.connect(index_db) as conn: + conn.execute( + "INSERT INTO sessions (native_id, origin, raw_id, title, content_hash) VALUES (?, ?, ?, ?, ?)", + ("codex-id", "claude-code-session", wrong_owner_raw_id, "wrong owner", bytes(32)), + ) + conn.execute( + "INSERT INTO sessions (native_id, origin, raw_id, title, content_hash) VALUES (?, ?, ?, ?, ?)", + ("codex-id", "codex-session", unrelated_codex_raw_id, "unrelated fallback", bytes(32)), + ) + conn.commit() + + processor = LiveBatchProcessor( + cast(Any, SimpleNamespace(archive_root=tmp_path, backend=SimpleNamespace(db_path=index_db))), + (WatchSource(name="codex", root=root),), + cursor=CursorStore(index_db), + parser_fingerprint="test-parser", + ) + + assert processor._append_payload_for_provider(path, "codex", b'{"type":"event_msg"}\n') is None + + +def test_codex_append_identity_rejects_global_fallback_when_ownership_query_errors( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, +) -> None: + from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database + from polylogue.storage.sqlite.archive_tiers.source_write import write_source_raw_session + from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier + + root = tmp_path / "sessions" + root.mkdir() + path = root / "candidate.jsonl" + payload = b'{"type":"session_meta","payload":{"id":"codex-id"}}\n' + path.write_bytes(payload) + index_db = tmp_path / "index.db" + source_db = tmp_path / "source.db" + initialize_archive_database(index_db, ArchiveTier.INDEX) + initialize_archive_database(source_db, ArchiveTier.SOURCE) + with sqlite3.connect(source_db) as conn: + unrelated_raw_id = write_source_raw_session( + conn, + origin="codex-session", + source_path=str(root / "unrelated.jsonl"), + source_index=0, + payload=payload, + acquired_at_ms=1_770_000_000_000, + ) + with sqlite3.connect(index_db) as conn: + conn.execute( + "INSERT INTO sessions (native_id, origin, raw_id, title, content_hash) VALUES (?, ?, ?, ?, ?)", + ("codex-id", "codex-session", unrelated_raw_id, "unrelated fallback", bytes(32)), + ) + conn.commit() + + processor = LiveBatchProcessor( + cast(Any, SimpleNamespace(archive_root=tmp_path, backend=SimpleNamespace(db_path=index_db))), + (WatchSource(name="codex", root=root),), + cursor=CursorStore(index_db), + parser_fingerprint="test-parser", + ) + + assert processor._existing_provider_session_id(path, expected_origin="codex-session") == "codex-id" + + def unavailable_ownership_view(*_args: object, **_kwargs: object) -> sqlite3.Connection: + raise sqlite3.OperationalError("source tier unavailable") + + # The global index fallback must be viable so this assertion proves that an + # unavailable ownership view, rather than another sqlite failure, rejects + # the append. + monkeypatch.setattr(processor, "_archive_has_native_session", lambda *_args, **_kwargs: True) + monkeypatch.setattr(sqlite3, "connect", unavailable_ownership_view) + + assert processor._append_payload_for_provider(path, "codex", b'{"type":"event_msg"}\n') is None + assert "source-path ownership view unavailable" in caplog.text + + def test_latest_raw_fingerprint_ignores_archive_source_row_with_missing_blob(tmp_path: Path) -> None: from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier @@ -5755,6 +5960,7 @@ def session(native_id: str, *texts: str) -> ParsedSession: # retry-candidate query (storage/repair.py) nothing stable to match # once the message text drifts -- exactly what happened to a real # production session that hit this guard under #2718's original + # wording. # The structured evidence row, not the diagnostic wording, is the # retry authorization. with sqlite3.connect(tmp_path / "source.db") as conn: diff --git a/tests/unit/storage/test_session_insight_refresh.py b/tests/unit/storage/test_session_insight_refresh.py index 515a087abd..358bd865b5 100644 --- a/tests/unit/storage/test_session_insight_refresh.py +++ b/tests/unit/storage/test_session_insight_refresh.py @@ -89,6 +89,48 @@ def test_rebuild_session_insights_preserves_null_thread_sort_key(tmp_path: Path) assert row["source_sort_key_ms"] is None +def test_latency_materialization_uses_latency_record_fallback_sort_key( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + db_path = _current_index_db(tmp_path, "latency-fallback-stamp") + session_id = _sid("latency-fallback", "codex-session") + created_at_ms = 1_700_000_000_123 + with open_connection(db_path) as conn: + store_records( + session=make_session("latency-fallback", source_name="codex", title="Created-only"), + messages=[make_message("latency-fallback:msg-1", "latency-fallback", text="hello")], + attachments=[], + conn=conn, + ) + conn.execute( + "UPDATE sessions SET created_at_ms = ?, updated_at_ms = NULL WHERE session_id = ?", + (created_at_ms, session_id), + ) + conn.commit() + + original_builder = rebuild_mod.__dict__["build_session_profile_record"] + + def profile_without_sort_key(*args: object, **kwargs: object) -> object: + record = original_builder(*args, **kwargs) + return record.model_copy(update={"source_sort_key": None}) + + monkeypatch.setitem(rebuild_mod.__dict__, "build_session_profile_record", profile_without_sort_key) + rebuild_session_insights_sync(conn, session_ids=[session_id]) + + materialization = conn.execute( + """ + SELECT source_sort_key_ms + FROM insight_materialization + WHERE session_id = ? AND insight_type = 'latency' + """, + (session_id,), + ).fetchone() + + assert materialization is not None + assert materialization[0] == created_at_ms + + @pytest.mark.asyncio async def test_apply_session_insight_session_updates_async_batches_hydrated_sessions( tmp_path: Path, @@ -1204,10 +1246,11 @@ def test_large_session_rebuild_uses_bounded_degraded_profile( conn.execute( """ UPDATE sessions - SET message_count = ?, word_count = ?, tool_use_count = ?, thinking_count = ? + SET message_count = ?, word_count = ?, tool_use_count = ?, thinking_count = ?, + created_at_ms = ?, updated_at_ms = NULL WHERE session_id = ? """, - (50, 1234, 7, 3, session_id), + (50, 1234, 7, 3, 1_700_000_000_000, session_id), ) conn.commit() @@ -1225,6 +1268,12 @@ def fail_full_load(_conn: sqlite3.Connection, _session_ids: object) -> object: (session_id,), ).fetchone() assert profile is not None + latency = conn.execute( + "SELECT source_updated_at, source_sort_key, input_high_water_mark FROM session_latency_profiles " + "WHERE session_id = ?", + (session_id,), + ).fetchone() + assert latency is not None work_events_row = conn.execute( "SELECT COUNT(*) FROM session_work_events WHERE session_id = ?", (session_id,), @@ -1247,6 +1296,12 @@ def fail_full_load(_conn: sqlite3.Connection, _session_ids: object) -> object: assert profile["message_count"] == 50 assert profile["word_count"] == 1234 assert profile["tool_use_count"] == 7 + assert profile["source_updated_at"] is None + assert profile["source_sort_key"] == pytest.approx(1_700_000_000.0) + assert profile["input_high_water_mark"] is None + assert latency["source_updated_at"] is None + assert latency["source_sort_key"] == pytest.approx(1_700_000_000.0) + assert latency["input_high_water_mark"] is None assert "large_session_bounded" in profile["inference_payload_json"] assert "large_session_bounded" in profile["enrichment_payload_json"] assert work_events == 0