diff --git a/docs/generated/api-operation-parity.json b/docs/generated/api-operation-parity.json index 7795956559..db3a0dc8c5 100644 --- a/docs/generated/api-operation-parity.json +++ b/docs/generated/api-operation-parity.json @@ -738,6 +738,11 @@ "async": true, "binding": "Polylogue.export_insight_bundle", "signature": "(self, request: 'InsightExportBundleRequest') -> 'InsightExportBundleResult'" + }, + { + "async": true, + "binding": "Polylogue.regenerate_private_fable_packet", + "signature": "(self, *, seed: 'str', requested_size: 'int', schema_id: 'str' = 'delegation.discourse', schema_version: 'int' = 1, exact_template_cap: 'int' = 1) -> 'FableDelegationPacket'" } ], "route_class": "index-read", diff --git a/docs/library-api.md b/docs/library-api.md index 5a5d995e69..bb6e708f36 100644 --- a/docs/library-api.md +++ b/docs/library-api.md @@ -541,6 +541,7 @@ Route/tier class: `index-read`. CLI: `analyze`, `read`. MCP: `query`, `get`, `st | `Polylogue.pathology_report` | `async (self, spec: 'SessionQuerySpec | None' = None, *, limit: 'int | None' = None) -> 'PathologyReport'` | | `Polylogue.portfolio_bundle` | `async (self, spec: 'SessionQuerySpec | None' = None, *, limit: 'int | None' = None, top_n: 'int' = 10) -> 'PortfolioBundle'` | | `Polylogue.export_insight_bundle` | `async (self, request: 'InsightExportBundleRequest') -> 'InsightExportBundleResult'` | +| `Polylogue.regenerate_private_fable_packet` | `async (self, *, seed: 'str', requested_size: 'int', schema_id: 'str' = 'delegation.discourse', schema_version: 'int' = 1, exact_template_cap: 'int' = 1) -> 'FableDelegationPacket'` | ### Context and evidence diff --git a/docs/plans/layering-surface-baseline.json b/docs/plans/layering-surface-baseline.json index 7db1b91adc..d1e67f7e52 100644 --- a/docs/plans/layering-surface-baseline.json +++ b/docs/plans/layering-surface-baseline.json @@ -24,6 +24,11 @@ "file": "polylogue/api/archive.py", "import": "polylogue.sources.parsers.hermes_lifecycle" }, + { + "target": "polylogue/api", + "file": "polylogue/api/archive.py", + "import": "polylogue.storage.block_anchor" + }, { "target": "polylogue/api", "file": "polylogue/api/archive.py", diff --git a/polylogue/api/archive.py b/polylogue/api/archive.py index b758d8998c..4d32dcbbd4 100644 --- a/polylogue/api/archive.py +++ b/polylogue/api/archive.py @@ -107,6 +107,7 @@ from polylogue.core.protocols import ProgressCallback from polylogue.insights.audit import InsightRigorAuditQuery, InsightRigorAuditReport from polylogue.insights.export_bundles import InsightExportBundleRequest, InsightExportBundleResult + from polylogue.insights.fable_packet import FableDelegationPacket from polylogue.insights.hermes_integration_health import HermesIntegrationHealth from polylogue.insights.judgment.types import ComparativeJudgment from polylogue.insights.pathology import PathologyReport @@ -3858,6 +3859,7 @@ async def export_otel( async def resolve_ref(self, ref: str) -> PublicRefResolutionPayload: """Resolve one public object/evidence ref into a bounded read payload.""" + from polylogue.storage.block_anchor import InvalidBlockAnchorError, parse_block_anchor, resolve_block_anchor from polylogue.surfaces.payloads import PublicRefResolutionPayload invalid_unicode_ref = _invalid_unicode_ref_payload(ref) @@ -3875,6 +3877,47 @@ async def resolve_ref(self, ref: str) -> PublicRefResolutionPayload: return cast(PublicRefResolutionPayload, bounded_batch_ref) if batch_candidate.kind != "annotation-batch": return cast(PublicRefResolutionPayload, bounded_batch_ref) + try: + block_anchor = parse_block_anchor(ref) + except InvalidBlockAnchorError: + block_anchor = None + if block_anchor is not None: + archive_root = _active_archive_root(self.config) + + def read_anchor(archive: ArchiveStore) -> PublicRefResolutionPayload: + resolution = resolve_block_anchor(archive._conn, block_anchor) + resolved = resolution.state in {"ok", "drifted_position", "drifted_message"} + object_refs = ( + (f"message:{resolution.resolved_message_id}",) if resolution.resolved_message_id is not None else () + ) + return PublicRefResolutionPayload( + ref=ref, + kind="block", + resolved=resolved, + payload_kind="block-anchor", + payload={ + "state": resolution.state, + "anchor": resolution.anchor.to_text(), + "resolved_message_id": resolution.resolved_message_id, + "resolved_position": resolution.resolved_position, + "candidates": [ + {"message_id": message_id, "position": position} + for message_id, position in resolution.candidates + ], + "detail": resolution.detail, + }, + object_refs=object_refs, + caveats=() if resolved else (resolution.detail or f"block anchor state: {resolution.state}",), + ) + + return await run_archive_read( + archive_root, + operation="archive.resolve_block_anchor", + arguments={"ref": ref}, + work=read_anchor, + projection="block-anchor-resolution", + stable_order="canonical", + ) try: parsed = parse_public_ref(ref) except ValueError as exc: @@ -5491,6 +5534,40 @@ async def insight_rigor_audit( workload_class="scan", ) + async def regenerate_private_fable_packet( + self, + *, + seed: str, + requested_size: int, + schema_id: str = "delegation.discourse", + schema_version: int = 1, + exact_template_cap: int = 1, + ) -> FableDelegationPacket: + """Cold-regenerate the private descriptive Fable packet from the archive.""" + from polylogue.insights.fable_packet import regenerate_private_fable_packet + + return await run_archive_read( + _active_archive_root(self.config), + operation="insights.fable_packet.regenerate", + arguments={ + "seed": seed, + "requested_size": requested_size, + "schema_id": schema_id, + "schema_version": schema_version, + "exact_template_cap": exact_template_cap, + }, + work=lambda archive: regenerate_private_fable_packet( + archive, + seed=seed, + requested_size=requested_size, + schema_id=schema_id, + schema_version=schema_version, + exact_template_cap=exact_template_cap, + ), + projection="fable-delegation-packet", + workload_class="scan", + ) + async def get_messages_paginated( self, session_id: str, diff --git a/polylogue/api/operation_parity.py b/polylogue/api/operation_parity.py index e7c3731f5c..f0b6819b77 100644 --- a/polylogue/api/operation_parity.py +++ b/polylogue/api/operation_parity.py @@ -265,6 +265,7 @@ def _surface(*names: str) -> SurfaceBinding: "Polylogue.pathology_report", "Polylogue.portfolio_bundle", "Polylogue.export_insight_bundle", + "Polylogue.regenerate_private_fable_packet", ), _surface("analyze", "read"), _surface("query", "get", "status", "explain"), diff --git a/polylogue/archive/query/evaluator.py b/polylogue/archive/query/evaluator.py index ff76bad30a..1656c9547c 100644 --- a/polylogue/archive/query/evaluator.py +++ b/polylogue/archive/query/evaluator.py @@ -15,6 +15,10 @@ from polylogue.archive.query.expression import RefOperand, RelationGrain, ResolvedRefOperand from polylogue.core.query_identity import require_supported_definition_protocol_version from polylogue.core.refs import ObjectRef +from polylogue.storage.sqlite.holdout_cohorts import ( + HoldoutAccessError, + require_non_holdout_access, +) from polylogue.storage.sqlite.query_objects import ( EvaluationReceipt, QueryObject, @@ -72,11 +76,13 @@ def __init__( *, owner_query_hash: str | None = None, created_at_ms: int = 0, + declared_confirmation: bool = False, ) -> None: self._conn = conn self._evaluator = evaluator self._owner_query_hash = owner_query_hash self._created_at_ms = created_at_ms + self._declared_confirmation = declared_confirmation def resolve_ref_operand(self, operand: RefOperand) -> ResolvedRefOperand: reference = operand.reference @@ -121,6 +127,14 @@ def _retained_result( manifest = get_result_set(self._conn, result_set_id) if manifest is None: raise RetainedRelationUnavailableError(f"retained result-set:{result_set_id} is unavailable") + try: + require_non_holdout_access( + self._conn, + result_set_id, + declared_confirmation=self._declared_confirmation, + ) + except HoldoutAccessError as exc: + raise RetainedRelationUnavailableError(str(exc)) from exc if extra_lineage: run = get_retained_query_run(self._conn, extra_lineage[0].object_id) if run is None or manifest.query_hash != run.query_hash: diff --git a/polylogue/insights/fable_packet.py b/polylogue/insights/fable_packet.py new file mode 100644 index 0000000000..e190a9280b --- /dev/null +++ b/polylogue/insights/fable_packet.py @@ -0,0 +1,313 @@ +"""Private, descriptive delegation packet compilation for the Fable campaign. + +The compiler is intentionally evidence-first and emits ``not_supported`` when +the supplied structural/annotation material cannot support a private descriptive +packet. It produces no comparative, utility, routing-quality, or sentiment +claim. +""" + +from __future__ import annotations + +from collections import Counter, defaultdict +from collections.abc import Sequence +from dataclasses import dataclass +from hashlib import sha256 +from typing import Literal + +from polylogue.archive.query.predicate import QueryBoolPredicate, QueryFieldPredicate, QueryFieldRef +from polylogue.core.refs import delegation_edge_object_id +from polylogue.insights.cohorts import CohortCandidate, CohortManifest, CohortSpec, compile_cohort_manifest +from polylogue.storage.sqlite.archive_tiers.archive import ArchiveDelegationQueryRow, ArchiveStore + +PacketStatus = Literal["complete", "not_supported"] + + +@dataclass(frozen=True) +class DelegationPacketRow: + """Bounded structural evidence needed by the descriptive packet.""" + + delegation_ref: str + evidence_basis: Literal["action", "edge"] + mapping_state: str + instruction_sha256: str | None + + +@dataclass(frozen=True) +class DelegationPacketLabel: + """One accepted or candidate descriptive annotation with evidence spans.""" + + delegation_ref: str + field: str + value: str | None + batch_id: str + accepted: bool + applicable: bool | None + confidence: float | None + evidence_refs: tuple[str, ...] + + +@dataclass(frozen=True) +class DescriptiveDistribution: + """One accepted-label distribution with explicit denominator/missingness.""" + + field: str + value: str + count: int + proportion: float + denominator_n: int + missing_n: int + + +@dataclass(frozen=True) +class FableDelegationPacket: + """A private descriptive packet or a concrete fail-closed explanation.""" + + status: PacketStatus + manifest_id: str + population_count: int + action_observed_count: int + edge_only_count: int + unresolved_count: int + selected_refs: tuple[str, ...] + annotation_schema_id: str | None + annotation_batches: tuple[str, ...] + distributions: tuple[DescriptiveDistribution, ...] + disagreement_count: int + specimen_refs: tuple[str, ...] + counterexample_refs: tuple[str, ...] + limits: tuple[str, ...] + not_supported_reasons: tuple[str, ...] = () + + +def _unsupported( + manifest: CohortManifest, + rows: Sequence[DelegationPacketRow], + reasons: Sequence[str], +) -> FableDelegationPacket: + return FableDelegationPacket( + status="not_supported", + manifest_id=manifest.manifest_id, + population_count=len(rows), + action_observed_count=sum(row.evidence_basis == "action" for row in rows), + edge_only_count=sum(row.evidence_basis == "edge" for row in rows), + unresolved_count=sum(row.mapping_state == "unresolved" for row in rows), + selected_refs=manifest.selected_refs, + annotation_schema_id=None, + annotation_batches=(), + distributions=(), + disagreement_count=0, + specimen_refs=(), + counterexample_refs=(), + limits=("private_descriptive_only",), + not_supported_reasons=tuple(sorted(set(reasons))), + ) + + +def compile_private_fable_packet( + *, + manifest: CohortManifest, + rows: Sequence[DelegationPacketRow], + annotation_schema_id: str | None, + labels: Sequence[DelegationPacketLabel], +) -> FableDelegationPacket: + """Compile a private descriptive packet or fail closed with named gaps. + + Accepted labels must target sampled, action-observed rows and retain at + least one evidence ref. Edge-only and unresolved rows are coverage facts, + never rhetorical evidence. Distributions are per field over applicable + accepted labels, retaining their denominator and missing label count. + """ + + by_ref = {row.delegation_ref: row for row in rows} + reasons: list[str] = [] + if annotation_schema_id is None: + reasons.append("missing_annotation_schema") + if not manifest.selected_refs: + reasons.append("empty_deterministic_sample") + missing_sample_refs = sorted(set(manifest.selected_refs) - by_ref.keys()) + if missing_sample_refs: + reasons.append("selected_refs_missing_from_structural_population") + action_rows = {ref: row for ref, row in by_ref.items() if row.evidence_basis == "action"} + if not action_rows: + reasons.append("no_action_observed_delegation_attempts") + + accepted = [label for label in labels if label.accepted] + if not accepted: + reasons.append("no_accepted_labels") + for label in accepted: + if label.delegation_ref not in action_rows: + reasons.append("accepted_label_not_action_observed") + if label.delegation_ref not in manifest.selected_refs: + reasons.append("accepted_label_outside_deterministic_sample") + if not label.evidence_refs: + reasons.append("accepted_label_missing_evidence") + if reasons: + return _unsupported(manifest, rows, reasons) + + labels_by_field: dict[str, list[DelegationPacketLabel]] = defaultdict(list) + for label in accepted: + labels_by_field[label.field].append(label) + distributions: list[DescriptiveDistribution] = [] + disagreement_count = 0 + specimen_refs: set[str] = set() + counterexample_refs: set[str] = set() + for field, field_labels in sorted(labels_by_field.items()): + counterexample_refs.update(label.delegation_ref for label in field_labels if label.applicable is False) + applicable = [label for label in field_labels if label.applicable is not False] + denominator = len(applicable) + missing = sum(label.value is None for label in applicable) + values = Counter(label.value for label in applicable if label.value is not None) + for value, count in sorted(values.items()): + assert value is not None + distributions.append( + DescriptiveDistribution( + field=field, + value=value, + count=count, + proportion=count / denominator if denominator else 0.0, + denominator_n=denominator, + missing_n=missing, + ) + ) + labels_by_ref: dict[str, set[str]] = defaultdict(set) + for label in applicable: + if label.value is not None: + labels_by_ref[label.delegation_ref].add(label.value) + specimen_refs.add(label.delegation_ref) + disagreement_count += sum(len(values) > 1 for values in labels_by_ref.values()) + + return FableDelegationPacket( + status="complete", + manifest_id=manifest.manifest_id, + population_count=len(rows), + action_observed_count=len(action_rows), + edge_only_count=sum(row.evidence_basis == "edge" for row in rows), + unresolved_count=sum(row.mapping_state == "unresolved" for row in rows), + selected_refs=manifest.selected_refs, + annotation_schema_id=annotation_schema_id, + annotation_batches=tuple(sorted({label.batch_id for label in accepted})), + distributions=tuple(distributions), + disagreement_count=disagreement_count, + specimen_refs=tuple(sorted(specimen_refs)), + counterexample_refs=tuple(sorted(counterexample_refs)), + limits=( + "private_descriptive_only", + "no_comparative_authoritarianism_success_utility_or_routing_quality_claims", + "edge_only_and_unresolved_rows_excluded_from_rhetoric_denominators", + ), + ) + + +def _delegation_ref(row: ArchiveDelegationQueryRow) -> str: + instruction_block_id = row.instruction_tool_use_block_id + if instruction_block_id is not None: + return f"delegation:{instruction_block_id}" + if row.child_session_id is None: + raise ValueError("edge-only delegation rows require a child session id") + return f"delegation:{delegation_edge_object_id(row.parent_session_id, row.child_session_id)}" + + +def regenerate_private_fable_packet( + archive: ArchiveStore, + *, + seed: str, + requested_size: int, + schema_id: str = "delegation.discourse", + schema_version: int = 1, + exact_template_cap: int = 1, +) -> FableDelegationPacket: + """Cold-regenerate a private packet from canonical archive evidence. + + This is intentionally a read-only composition of the canonical delegation + relation and the durable annotation substrate. Missing schema, batches, or + active labels flows into the compiler's explicit ``not_supported`` result. + """ + + all_rows = archive.query_delegations(QueryBoolPredicate("and", ()), limit=100_000) + packet_rows = tuple( + DelegationPacketRow( + delegation_ref=_delegation_ref(row), + evidence_basis="action" if row.instruction_tool_use_block_id is not None else "edge", + mapping_state=row.mapping_state, + instruction_sha256=( + sha256(row.instruction_payload.encode("utf-8")).hexdigest() + if row.instruction_payload is not None + else None + ), + ) + for row in all_rows + ) + cursor = f"index:{archive.index_db_path.stat().st_mtime_ns}" + manifest = compile_cohort_manifest( + CohortSpec( + population_query="delegations where basis:action", + archive_cursor=cursor, + seed=seed, + requested_size=requested_size, + strata=("origin", "dispatch_model"), + exact_template_cap=exact_template_cap, + ), + tuple( + CohortCandidate( + object_ref=_delegation_ref(row), + dimensions={"origin": row.parent_origin, "dispatch_model": row.dispatch_turn_model}, + template_key=( + sha256(row.instruction_payload.encode("utf-8")).hexdigest() if row.instruction_payload else None + ), + exclusion_reason=None if row.instruction_tool_use_block_id is not None else "edge_only", + ) + for row in all_rows + ), + ) + try: + schema = archive.get_annotation_schema(schema_id, schema_version) + except KeyError: + schema = None + assertions = archive.query_assertions( + QueryFieldPredicate( + field="kind", + values=("annotation",), + field_ref=QueryFieldRef(scope="unit", name="kind", source_name="assertions", unit="assertion"), + ), + limit=100_000, + ) + labels: list[DelegationPacketLabel] = [] + qualified_schema_id = f"{schema_id}@v{schema_version}" + for assertion in assertions: + value = assertion.value + if assertion.status != "active" or not isinstance(value, dict) or value.get("_schema") != qualified_schema_id: + continue + batch_id = assertion.scope_ref.removeprefix("annotation-batch:") if assertion.scope_ref else "unbatched" + applicable_value = value.get("applicable") + confidence_value = value.get("confidence") + for field, field_value in value.items(): + if field.startswith("_") or field in {"applicable", "confidence", "abstain"}: + continue + labels.append( + DelegationPacketLabel( + delegation_ref=assertion.target_ref, + field=field, + value=field_value if isinstance(field_value, str) else None, + batch_id=batch_id, + accepted=True, + applicable=applicable_value if isinstance(applicable_value, bool) else None, + confidence=float(confidence_value) if isinstance(confidence_value, (int, float)) else None, + evidence_refs=assertion.evidence_refs, + ) + ) + return compile_private_fable_packet( + manifest=manifest, + rows=packet_rows, + annotation_schema_id=schema.schema.qualified_id if schema is not None else None, + labels=labels, + ) + + +__all__ = [ + "DelegationPacketLabel", + "DelegationPacketRow", + "DescriptiveDistribution", + "FableDelegationPacket", + "compile_private_fable_packet", + "regenerate_private_fable_packet", +] diff --git a/polylogue/storage/block_anchor.py b/polylogue/storage/block_anchor.py new file mode 100644 index 0000000000..a184a7433f --- /dev/null +++ b/polylogue/storage/block_anchor.py @@ -0,0 +1,231 @@ +"""Block content-hash citation anchors (svfj) — resolve a stored citation +against the current archive, never guessing. + +``blocks.content_hash`` (index.py) hashes a block's canonical EVIDENCE only — +type, text, tool_name, canonical tool_input, semantic/media/language, +is_error, exit_code — deliberately excluding session_id/message_id/position/ +tool_id. That is what lets a citation anchor survive fork-position replay, +re-ingest renumbering, and provider tool-id regeneration: the identity +components can shift, but the evidence they point at is still findable by +its hash. + +The textual anchor form is ``::::block@sha256:``. +Session/message ids are themselves colon-bearing (``codex-session:abc``), so +the outer separator is the double colon, never single. + +The resolver returns a TYPED state, never a silent best-guess pick: + +- ``ok`` — the hash resolves in the named message at the expected position + (or no position hint was given). +- ``drifted_position`` — the hash resolves in the named message, but at a + different position than the caller's hint. +- ``drifted_message`` — the hash is not in the named message, but resolves + to exactly one other message in the same session. +- ``ambiguous`` — more than one block carries this hash within the resolved + scope (e.g. the same prompt text repeated N times); candidates are listed, + never picked for the caller. +- ``hash_mismatch`` — the named message/position exists, but its current + content_hash differs from the anchor's. A hard fail: never auto-rewrite + the anchor or guess which content it "really" meant. +- ``missing`` — neither the message nor any block with this hash resolves + anywhere in the session. +- ``relocated_lineage`` and ``quarantined`` are reserved states in the type + but NOT YET PRODUCED by this resolver — they require the lineage- + composition read path (searching the fork/resume neighborhood, preferring + prefix-sharing inheritance over spawned-fresh) and the topology-edge + quarantine model respectively. A message that has moved to a composed + parent-lineage session currently resolves as ``missing``, not a guess. + Filed as a follow-up rather than implemented here without that grounding. +""" + +from __future__ import annotations + +import sqlite3 +from dataclasses import dataclass, field +from typing import Literal + +BlockAnchorState = Literal[ + "ok", + "drifted_position", + "drifted_message", + "relocated_lineage", + "ambiguous", + "missing", + "quarantined", + "hash_mismatch", +] + +_ANCHOR_SEPARATOR = "::" +_BLOCK_PREFIX = "block@sha256:" + + +@dataclass(frozen=True) +class BlockAnchor: + """Parsed textual citation anchor.""" + + session_id: str + message_id: str + content_hash_hex: str + + def to_text(self) -> str: + return format_block_anchor(self.session_id, self.message_id, self.content_hash_hex) + + +@dataclass(frozen=True) +class BlockAnchorResolution: + """Result of resolving a :class:`BlockAnchor` against the current archive.""" + + state: BlockAnchorState + anchor: BlockAnchor + resolved_message_id: str | None = None + resolved_position: int | None = None + candidates: tuple[tuple[str, int], ...] = field(default_factory=tuple) + detail: str = "" + + +def format_block_anchor(session_id: str, message_id: str, content_hash_hex: str) -> str: + """Build the canonical textual anchor form for a block.""" + + return f"{session_id}{_ANCHOR_SEPARATOR}{message_id}{_ANCHOR_SEPARATOR}{_BLOCK_PREFIX}{content_hash_hex}" + + +class InvalidBlockAnchorError(ValueError): + """Raised when a textual anchor does not parse as ``::::block@sha256:``.""" + + +def parse_block_anchor(anchor_text: str) -> BlockAnchor: + """Parse the canonical textual anchor form. + + Raises :class:`InvalidBlockAnchorError` on malformed input rather than + guessing a partial parse — a citation anchor is meant to be exact. + """ + + parts = anchor_text.split(_ANCHOR_SEPARATOR) + if len(parts) != 3: + raise InvalidBlockAnchorError(f"expected ::::block@sha256:, got {anchor_text!r}") + session_id, message_id, block_part = parts + if not session_id or not message_id: + raise InvalidBlockAnchorError(f"empty session_id/message_id in anchor {anchor_text!r}") + if not block_part.startswith(_BLOCK_PREFIX): + raise InvalidBlockAnchorError(f"expected {_BLOCK_PREFIX!r} prefix, got {anchor_text!r}") + content_hash_hex = block_part[len(_BLOCK_PREFIX) :] + if len(content_hash_hex) != 64 or not all(c in "0123456789abcdef" for c in content_hash_hex): + raise InvalidBlockAnchorError(f"expected a 64-char lowercase hex sha256 digest, got {anchor_text!r}") + return BlockAnchor(session_id=session_id, message_id=message_id, content_hash_hex=content_hash_hex) + + +def resolve_block_anchor( + conn: sqlite3.Connection, + anchor: BlockAnchor, + *, + position_hint: int | None = None, +) -> BlockAnchorResolution: + """Resolve a citation anchor against the current archive (read-only). + + ``conn`` must have ``row_factory = sqlite3.Row`` (or plain tuple access + with matching column order — this function selects by name via + ``sqlite3.Row``, so a plain-tuple connection will raise). + """ + + content_hash = bytes.fromhex(anchor.content_hash_hex) + + message_row = conn.execute( + "SELECT message_id, session_id FROM messages WHERE message_id = ?", + (anchor.message_id,), + ).fetchone() + + if message_row is not None and message_row["session_id"] == anchor.session_id: + in_message = conn.execute( + "SELECT position FROM blocks WHERE message_id = ? AND content_hash = ? ORDER BY position", + (anchor.message_id, content_hash), + ).fetchall() + if len(in_message) > 1: + return BlockAnchorResolution( + state="ambiguous", + anchor=anchor, + resolved_message_id=anchor.message_id, + candidates=tuple((anchor.message_id, int(row["position"])) for row in in_message), + detail=f"{len(in_message)} blocks in this message share the anchor's content_hash", + ) + if len(in_message) == 1: + position = int(in_message[0]["position"]) + state: BlockAnchorState = "ok" if position_hint is None or position_hint == position else "drifted_position" + return BlockAnchorResolution( + state=state, + anchor=anchor, + resolved_message_id=anchor.message_id, + resolved_position=position, + ) + + # No block in the named message carries this hash. If the named + # position still exists but with different content, that is a hard + # hash_mismatch -- never guess a rewrite. + if position_hint is not None: + mismatch_row = conn.execute( + "SELECT content_hash FROM blocks WHERE message_id = ? AND position = ?", + (anchor.message_id, position_hint), + ).fetchone() + if mismatch_row is not None and mismatch_row["content_hash"] != content_hash: + return BlockAnchorResolution( + state="hash_mismatch", + anchor=anchor, + resolved_message_id=anchor.message_id, + resolved_position=position_hint, + detail="content_hash at the hinted position differs from the anchor -- never auto-rewritten", + ) + + # Look for the hash elsewhere in the same session (message drift). + in_session = conn.execute( + """ + SELECT b.message_id, b.position + FROM blocks b + JOIN messages m ON m.message_id = b.message_id + WHERE m.session_id = ? AND b.content_hash = ? + ORDER BY b.message_id, b.position + """, + (anchor.session_id, content_hash), + ).fetchall() + if len(in_session) > 1: + return BlockAnchorResolution( + state="ambiguous", + anchor=anchor, + candidates=tuple((str(row["message_id"]), int(row["position"])) for row in in_session), + detail=f"{len(in_session)} blocks across the session share the anchor's content_hash", + ) + if len(in_session) == 1: + return BlockAnchorResolution( + state="drifted_message", + anchor=anchor, + resolved_message_id=str(in_session[0]["message_id"]), + resolved_position=int(in_session[0]["position"]), + ) + + return BlockAnchorResolution( + state="missing", + anchor=anchor, + detail="no block with this content_hash resolves in the named session", + ) + + # The named message_id no longer exists (or belongs to a different + # session than the anchor claims). Resolving across a fork/resume + # lineage neighborhood is not yet implemented here (relocated_lineage) -- + # report the honest, conservative state rather than guess. + return BlockAnchorResolution( + state="missing", + anchor=anchor, + detail=( + "message_id not found in the named session; lineage-neighborhood search " + "(relocated_lineage) is not yet implemented, see polylogue-svfj follow-up" + ), + ) + + +__all__ = [ + "BlockAnchor", + "BlockAnchorResolution", + "BlockAnchorState", + "InvalidBlockAnchorError", + "format_block_anchor", + "parse_block_anchor", + "resolve_block_anchor", +] diff --git a/polylogue/storage/sqlite/archive_tiers/archive.py b/polylogue/storage/sqlite/archive_tiers/archive.py index d14cb084c1..f25a162a8d 100644 --- a/polylogue/storage/sqlite/archive_tiers/archive.py +++ b/polylogue/storage/sqlite/archive_tiers/archive.py @@ -7999,6 +7999,7 @@ def query_delegations( normalized_offset = max(int(offset), 0) order_direction = _query_unit_order_direction(sort_direction) clause, params = _structural_predicate_clause("delegation", "d", predicate, session_alias="s") + where_clause = f"WHERE {clause}" if clause else "" session_clause = "" session_params: list[object] = [] if session_filters: @@ -8008,7 +8009,7 @@ def query_delegations( SELECT d.* FROM delegations d JOIN sessions s ON s.session_id = d.parent_session_id - WHERE {clause} + {where_clause} {session_clause} ORDER BY d.parent_session_id {order_direction}, COALESCE(d.instruction_tool_use_block_id, d.child_session_id) {order_direction} diff --git a/polylogue/storage/sqlite/holdout_cohorts.py b/polylogue/storage/sqlite/holdout_cohorts.py new file mode 100644 index 0000000000..5a076af1bc --- /dev/null +++ b/polylogue/storage/sqlite/holdout_cohorts.py @@ -0,0 +1,259 @@ +"""Holdout cohorts: persistence-class policy + planner enforcement (rxdo.9.4). + +Holdout is an access POLICY layered on an existing promoted +:class:`~polylogue.storage.sqlite.query_objects.ResultSetManifest`, not a +second cohort/result relation type -- a result set keeps whatever +``persistence_class`` it was created with (typically ``cohort``), and +:func:`mark_holdout` additionally records that its members are reserved for +confirmation and must not be read by exploratory analysis. This gives demo +and evaluation claims a provable "held on untouched data" leg: exploratory +queries are refused read access by default (:func:`require_non_holdout_access`), +a declared confirmation run is explicitly allowed and leaves a visible +access receipt, and an undeclared/accidental access is recorded as +contamination that can never be retroactively cleared -- once contaminated, +the relation can no longer back an "untouched holdout" claim. + +The production reference planner calls +:func:`require_non_holdout_access` before resolving an exploratory query's +``from result-set:`` operand. The guard therefore protects the real MCP +query route, while direct confirmation tooling can opt into the declared +confirmation mode and record its receipt. + +Reset/excision durability: no excision mechanism exists for ``result_sets`` +in this tree yet (see ``polylogue-layg`` for the separate source.db blob +excision cluster, which does not touch this table). The floor this module +ships is the migration's ``ON DELETE RESTRICT`` FK from +``result_set_holdout_policies`` to ``result_sets`` -- a raw ``DELETE`` of a +holdout-marked result set raises ``sqlite3.IntegrityError`` rather than +silently dropping the policy (see +``test_deleting_a_holdout_marked_result_set_is_blocked_by_the_durable_fk``). +A future excision/reset mechanism must route through an explicit +unmark-then-delete step (or a policy-aware cascade) rather than a raw +DELETE, or it will hit this same constraint; that integration is not +designed or implemented here. +""" + +from __future__ import annotations + +import json +import sqlite3 +from collections.abc import Mapping +from dataclasses import dataclass + +from polylogue.storage.sqlite.query_objects import get_result_set + + +class HoldoutAccessError(RuntimeError): + """Raised when an exploratory read is attempted against a holdout relation.""" + + +@dataclass(frozen=True, slots=True) +class HoldoutPolicy: + result_set_id: str + frame: str + selection_definition: Mapping[str, object] + intended_confirmation_use: str + authority: str + created_epoch: str + + +@dataclass(frozen=True, slots=True) +class HoldoutAccessReceipt: + receipt_id: str + result_set_id: str + accessor_ref: str + declared_confirmation: bool + contamination: bool + reason: str | None + accessed_at_ms: int + + +def mark_holdout( + conn: sqlite3.Connection, + *, + result_set_id: str, + frame: str, + selection_definition: Mapping[str, object], + intended_confirmation_use: str, + authority: str, + created_epoch: str, + created_at_ms: int, +) -> HoldoutPolicy: + """Mark an existing promoted result set as a holdout relation. + + Idempotent when re-marking with byte-identical policy fields; raises if + the result set does not exist or a *different* policy is already bound + (a holdout's declared frame/selection/authority is fixed at creation, + not silently mutable). + """ + if get_result_set(conn, result_set_id) is None: + raise KeyError(f"result-set:{result_set_id}") + policy = HoldoutPolicy( + result_set_id=result_set_id, + frame=frame, + selection_definition=dict(selection_definition), + intended_confirmation_use=intended_confirmation_use, + authority=authority, + created_epoch=created_epoch, + ) + existing = get_holdout_policy(conn, result_set_id) + if existing is not None: + if existing != policy: + raise ValueError(f"result-set:{result_set_id} is already a holdout with a different declared policy") + return existing + conn.execute( + """ + INSERT INTO result_set_holdout_policies ( + result_set_id, frame, selection_definition_json, intended_confirmation_use, + authority, created_epoch, created_at_ms + ) VALUES (?, ?, ?, ?, ?, ?, ?) + """, + ( + result_set_id, + frame, + _json(policy.selection_definition), + intended_confirmation_use, + authority, + created_epoch, + created_at_ms, + ), + ) + return policy + + +def get_holdout_policy(conn: sqlite3.Connection, result_set_id: str) -> HoldoutPolicy | None: + row = conn.execute( + """ + SELECT result_set_id, frame, selection_definition_json, intended_confirmation_use, + authority, created_epoch + FROM result_set_holdout_policies WHERE result_set_id = ? + """, + (result_set_id,), + ).fetchone() + if row is None: + return None + return HoldoutPolicy( + result_set_id=str(row[0]), + frame=str(row[1]), + selection_definition=json.loads(str(row[2])), + intended_confirmation_use=str(row[3]), + authority=str(row[4]), + created_epoch=str(row[5]), + ) + + +def is_holdout(conn: sqlite3.Connection, result_set_id: str) -> bool: + return get_holdout_policy(conn, result_set_id) is not None + + +def record_holdout_access( + conn: sqlite3.Connection, + *, + receipt_id: str, + result_set_id: str, + accessor_ref: str, + declared_confirmation: bool, + accessed_at_ms: int, + reason: str | None = None, +) -> HoldoutAccessReceipt: + """Record one read of a holdout relation's members, declared or not. + + ``declared_confirmation=False`` records contamination -- this is + permanent and cannot be cleared by a later declared access. Raises + :class:`KeyError` if ``result_set_id`` was never marked as a holdout + (there is nothing to declare confirmation *against*). + """ + if get_holdout_policy(conn, result_set_id) is None: + raise KeyError(f"result-set:{result_set_id} is not a holdout") + contamination = not declared_confirmation + receipt = HoldoutAccessReceipt( + receipt_id=receipt_id, + result_set_id=result_set_id, + accessor_ref=accessor_ref, + declared_confirmation=declared_confirmation, + contamination=contamination, + reason=reason, + accessed_at_ms=accessed_at_ms, + ) + conn.execute( + """ + INSERT INTO holdout_access_receipts ( + receipt_id, result_set_id, accessor_ref, declared_confirmation, + contamination, reason, accessed_at_ms + ) VALUES (?, ?, ?, ?, ?, ?, ?) + """, + ( + receipt_id, + result_set_id, + accessor_ref, + int(declared_confirmation), + int(contamination), + reason, + accessed_at_ms, + ), + ) + return receipt + + +def list_holdout_access_receipts(conn: sqlite3.Connection, result_set_id: str) -> tuple[HoldoutAccessReceipt, ...]: + rows = conn.execute( + """ + SELECT receipt_id, result_set_id, accessor_ref, declared_confirmation, contamination, reason, accessed_at_ms + FROM holdout_access_receipts WHERE result_set_id = ? ORDER BY accessed_at_ms + """, + (result_set_id,), + ).fetchall() + return tuple( + HoldoutAccessReceipt( + receipt_id=str(row[0]), + result_set_id=str(row[1]), + accessor_ref=str(row[2]), + declared_confirmation=bool(row[3]), + contamination=bool(row[4]), + reason=row[5] if row[5] is None else str(row[5]), + accessed_at_ms=int(row[6]), + ) + for row in rows + ) + + +def has_holdout_contamination(conn: sqlite3.Connection, result_set_id: str) -> bool: + """Whether any recorded access to this holdout was undeclared -- permanent once true.""" + row = conn.execute( + "SELECT 1 FROM holdout_access_receipts WHERE result_set_id = ? AND contamination = 1 LIMIT 1", + (result_set_id,), + ).fetchone() + return row is not None + + +def require_non_holdout_access(conn: sqlite3.Connection, result_set_id: str, *, declared_confirmation: bool) -> None: + """Planner-level guard: refuse an exploratory read of a holdout relation. + + Raises :class:`HoldoutAccessError` when ``result_set_id`` is a holdout + and the caller has not declared this as a confirmation access. Not a + holdout relation at all is always fine (no-op). + """ + if is_holdout(conn, result_set_id) and not declared_confirmation: + raise HoldoutAccessError( + f"result-set:{result_set_id} is a holdout relation; exploratory queries cannot read its " + "members. Pass declared_confirmation=True (and record the access) for an authorized " + "confirmation run." + ) + + +def _json(value: object) -> str: + return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + + +__all__ = [ + "HoldoutAccessError", + "HoldoutAccessReceipt", + "HoldoutPolicy", + "get_holdout_policy", + "has_holdout_contamination", + "is_holdout", + "list_holdout_access_receipts", + "mark_holdout", + "record_holdout_access", + "require_non_holdout_access", +] diff --git a/tests/unit/api/test_facade_contracts.py b/tests/unit/api/test_facade_contracts.py index e037967e15..b7a61fc926 100644 --- a/tests/unit/api/test_facade_contracts.py +++ b/tests/unit/api/test_facade_contracts.py @@ -50,9 +50,10 @@ delegation_subtree_object_id, ) from polylogue.sources.parsers.base import ParsedContentBlock, ParsedMessage, ParsedSession +from polylogue.storage.block_anchor import format_block_anchor from polylogue.storage.runtime.store_constants import SESSION_INSIGHT_MATERIALIZER_VERSION from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore -from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_tier +from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database, initialize_archive_tier from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier from polylogue.storage.sqlite.archive_tiers.user_write import upsert_assertion from tests.infra.frozen_clock import FrozenClock @@ -267,6 +268,7 @@ "join_typed_annotations", "neighbor_candidate_payloads", "resolve_ref", + "regenerate_private_fable_packet", "export_otel", "session_correlation_payload", # Pathology and portfolio read methods added in recent PRs. @@ -1847,6 +1849,78 @@ def test_session_not_found_is_typed_polylogue_error() -> None: assert SessionNotFoundError.http_status_code == 404 +async def test_regenerate_private_fable_packet_is_an_archive_backed_facade_route(tmp_path: Path) -> None: + """The public facade reaches cold regeneration and preserves fail-closed status.""" + archive = _archive(tmp_path) + try: + packet = await archive.regenerate_private_fable_packet(seed="facade-test", requested_size=1) + assert packet.status == "not_supported" + assert "empty_deterministic_sample" in packet.not_supported_reasons + finally: + await archive.close() + + +async def test_regenerate_private_fable_packet_reads_real_delegations_and_labels(tmp_path: Path) -> None: + """The facade adapter composes canonical delegation rows with durable labels. + + Anti-vacuity: this seeds the same delegation writer used by the archive + route, persists the built-in schema and one active evidence-backed label, + then asserts a complete packet. Removing either the delegation query, + durable schema read, or assertion query changes the result to + ``not_supported`` or an empty population. + """ + archive = _archive(tmp_path) + try: + with ArchiveStore(archive.config.archive_root) as archive_db: + parent_session_id = archive_db.write_parsed( + _delegation_parent_session(provider_session_id="fable-facade-parent-v1", with_dispatch=True) + ) + archive_db.write_parsed( + ParsedSession( + source_name=Provider.CLAUDE_CODE, + provider_session_id="fable-facade-child-v1", + title="Fable facade child fixture", + messages=[ParsedMessage(provider_message_id="c1", role=Role.ASSISTANT, text="on it")], + parent_session_provider_id="fable-facade-parent-v1", + branch_type=BranchType.SUBAGENT, + ) + ) + + initialize_archive_database(archive.config.archive_root / "user.db", ArchiveTier.USER) + instruction_block_id = f"{parent_session_id}:dispatch:0" + with sqlite3.connect(archive.config.archive_root / "user.db") as conn: + upsert_assertion( + conn, + assertion_id="fable-facade-label-v1", + target_ref=f"delegation:{instruction_block_id}", + kind=AssertionKind.ANNOTATION, + scope_ref="annotation-batch:fable-facade-batch-v1", + value={ + "_schema": "delegation.discourse@v1", + "directive_mode": "imperative", + "applicable": True, + "confidence": 0.95, + }, + evidence_refs=(f"block:{instruction_block_id}",), + status=AssertionStatus.ACTIVE, + author_kind="user", + now_ms=1, + ) + + packet = await archive.regenerate_private_fable_packet(seed="facade-test", requested_size=1) + assert packet.status == "complete" + assert packet.population_count == 1 + assert packet.action_observed_count == 1 + assert packet.annotation_schema_id == "delegation.discourse@v1" + assert packet.selected_refs == (f"delegation:{instruction_block_id}",) + assert packet.annotation_batches == ("fable-facade-batch-v1",) + assert {(item.field, item.value, item.count) for item in packet.distributions} == { + ("directive_mode", "imperative", 1) + } + finally: + await archive.close() + + @pytest.mark.parametrize( "method_name", sorted(MUTATION_BY_ID_RAISES_METHODS), @@ -2706,6 +2780,18 @@ async def test_resolve_ref_returns_bounded_session_message_block_and_runtime_pay assert evidence_block_payload.payload_kind == "block" assert evidence_block_payload.evidence_refs == (f"{session_id}::{message_id}::0",) + with ArchiveStore.open_existing(archive.config.archive_root) as archive_db: + content_hash = archive_db._conn.execute( + "SELECT content_hash FROM blocks WHERE message_id = ? AND position = 0", (message_id,) + ).fetchone()[0] + anchor_ref = format_block_anchor(session_id, message_id, bytes(content_hash).hex()) + anchor_payload = await archive.resolve_ref(anchor_ref) + assert anchor_payload.resolved is True + assert anchor_payload.payload_kind == "block-anchor" + assert anchor_payload.payload is not None + assert anchor_payload.payload["state"] == "ok" + assert anchor_payload.payload["anchor"] == anchor_ref + runtime_payload = await archive.resolve_ref(f"context-snapshot:{session_id}:session_start") assert runtime_payload.resolved is True assert runtime_payload.payload_kind == "context-snapshot" diff --git a/tests/unit/archive/query/test_evaluator.py b/tests/unit/archive/query/test_evaluator.py index 10cb9bc8d4..2b4bb9015b 100644 --- a/tests/unit/archive/query/test_evaluator.py +++ b/tests/unit/archive/query/test_evaluator.py @@ -14,6 +14,7 @@ from polylogue.core.refs import ObjectRef from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_tier from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier +from polylogue.storage.sqlite.holdout_cohorts import mark_holdout from polylogue.storage.sqlite.query_objects import ( EvaluationReceipt, QueryObject, @@ -248,3 +249,42 @@ def test_retained_sampled_result_set_fails_closed_as_a_set_operand() -> None: RefOperand(ObjectRef(kind="result-set", object_id=result.result_set_id)), DurableRefResolver(conn, _Evaluator()), ) + + +def test_retained_holdout_result_set_fails_closed_before_members_are_read() -> None: + conn = _conn() + query = put_query( + conn, + {"field": "title", "value": "holdout"}, + grain="session", + lane="dialogue", + rank_policy="mixed", + created_at_ms=1, + ) + result = put_result_set( + conn, + result_set_id="holdout-retained", + query_hash=query.query_hash, + grain="session", + corpus_epoch="index:g1", + member_refs=("session:secret",), + exactness="exact", + persistence_class="cohort", + created_at_ms=2, + ) + mark_holdout( + conn, + result_set_id=result.result_set_id, + frame="test-frame", + selection_definition={"seed": "test"}, + intended_confirmation_use="confirmation test", + authority="test", + created_epoch="index:g1", + created_at_ms=3, + ) + + with pytest.raises(RetainedRelationUnavailableError, match="exploratory queries cannot read"): + resolve_ref_operand( + RefOperand(ObjectRef(kind="result-set", object_id=result.result_set_id)), + DurableRefResolver(conn, _Evaluator()), + ) diff --git a/tests/unit/insights/test_fable_packet.py b/tests/unit/insights/test_fable_packet.py new file mode 100644 index 0000000000..87f939fe0d --- /dev/null +++ b/tests/unit/insights/test_fable_packet.py @@ -0,0 +1,75 @@ +from __future__ import annotations + +from polylogue.insights.cohorts import CohortCandidate, CohortManifest, CohortSpec, compile_cohort_manifest +from polylogue.insights.fable_packet import ( + DelegationPacketLabel, + DelegationPacketRow, + compile_private_fable_packet, +) + + +def _manifest() -> CohortManifest: + return compile_cohort_manifest( + CohortSpec("delegations where mapping_state:resolved", "index:24:fable", "seed", 2), + [ + CohortCandidate("delegation:one"), + CohortCandidate("delegation:two"), + ], + ) + + +def test_private_packet_reports_coverage_labels_distributions_and_limits() -> None: + manifest = _manifest() + packet = compile_private_fable_packet( + manifest=manifest, + rows=[ + DelegationPacketRow("delegation:one", "action", "resolved", "a" * 64), + DelegationPacketRow("delegation:two", "action", "unresolved", "b" * 64), + DelegationPacketRow("delegation:edge", "edge", "edge_only", None), + ], + annotation_schema_id="delegation-discourse/v1", + labels=[ + DelegationPacketLabel( + "delegation:one", "directive_mode", "direct", "batch-a", True, True, 0.9, ("block:a",) + ), + DelegationPacketLabel( + "delegation:two", "directive_mode", "direct", "batch-b", True, True, 0.8, ("block:b",) + ), + DelegationPacketLabel( + "delegation:two", "directive_mode", "collaborative", "batch-c", True, True, 0.7, ("block:b",) + ), + DelegationPacketLabel("delegation:one", "rationale", None, "batch-a", True, True, 0.9, ("block:a",)), + DelegationPacketLabel("delegation:two", "checkpoint", "none", "batch-b", True, False, 0.8, ("block:b",)), + ], + ) + + assert packet.status == "complete" + assert packet.action_observed_count == 2 + assert packet.edge_only_count == 1 + assert packet.unresolved_count == 1 + assert packet.annotation_batches == ("batch-a", "batch-b", "batch-c") + assert packet.disagreement_count == 1 + assert packet.counterexample_refs == ("delegation:two",) + assert { + (item.field, item.value, item.count, item.denominator_n, item.missing_n) for item in packet.distributions + } == { + ("directive_mode", "collaborative", 1, 3, 0), + ("directive_mode", "direct", 2, 3, 0), + } + assert "no_comparative_authoritarianism_success_utility_or_routing_quality_claims" in packet.limits + + +def test_private_packet_fails_closed_when_accepted_labels_lack_evidence() -> None: + manifest = _manifest() + packet = compile_private_fable_packet( + manifest=manifest, + rows=[DelegationPacketRow("delegation:one", "action", "resolved", "a" * 64)], + annotation_schema_id="delegation-discourse/v1", + labels=[DelegationPacketLabel("delegation:one", "directive_mode", "direct", "batch-a", True, True, 1.0, ())], + ) + + assert packet.status == "not_supported" + assert packet.not_supported_reasons == ( + "accepted_label_missing_evidence", + "selected_refs_missing_from_structural_population", + ) diff --git a/tests/unit/mcp/test_reference_query_pipeline.py b/tests/unit/mcp/test_reference_query_pipeline.py index 9acd4f47d7..daa6f15b4e 100644 --- a/tests/unit/mcp/test_reference_query_pipeline.py +++ b/tests/unit/mcp/test_reference_query_pipeline.py @@ -26,7 +26,8 @@ from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier -from polylogue.storage.sqlite.query_objects import QueryObject, put_query +from polylogue.storage.sqlite.holdout_cohorts import mark_holdout +from polylogue.storage.sqlite.query_objects import QueryObject, put_query, put_result_set from tests.infra.mcp import MCPServerUnderTest, invoke_surface from tests.unit.mcp.test_contract_evidence import _seeded_runtime_services @@ -126,3 +127,51 @@ def test_mcp_query_from_reference_with_stages_returns_typed_not_implemented( body = json.loads(result) assert body.get("code") == "not_implemented", body + + +def test_mcp_query_refuses_an_exploratory_read_of_a_holdout_result_set( + mcp_server: MCPServerUnderTest, tmp_path: Path +) -> None: + """The real MCP planner route cannot bypass the holdout guard. + + Anti-vacuity: this seeds a durable result-set and policy, then invokes the + production query handler. Removing the ``require_non_holdout_access`` call + from ``DurableRefResolver`` changes the response from a typed refusal to a + successful member read, so this test protects the actual enforcement seam. + """ + archive_root = tmp_path / "archive" + _seed_archive(archive_root) + with sqlite3.connect(archive_root / "user.db") as conn: + query = _origin_query(conn, origin="codex-session") + result = put_result_set( + conn, + result_set_id="rs-holdout-route", + query_hash=query.query_hash, + grain="session", + corpus_epoch="index:test", + member_refs=("session:codex-session:codex-1",), + exactness="exact", + persistence_class="cohort", + created_at_ms=2, + ) + mark_holdout( + conn, + result_set_id=result.result_set_id, + frame="test-frame", + selection_definition={"origin": "codex-session"}, + intended_confirmation_use="route enforcement test", + authority="test", + created_epoch="index:test", + created_at_ms=3, + ) + conn.commit() + + with _seeded_runtime_services(archive_root): + response = invoke_surface( + mcp_server._tool_manager._tools["query"].fn, + expression="from result-set:rs-holdout-route", + ) + + body = json.loads(response) + assert body["code"] == "invalid_argument", body + assert "holdout relation" in body.get("message", ""), body diff --git a/tests/unit/storage/test_block_anchor.py b/tests/unit/storage/test_block_anchor.py new file mode 100644 index 0000000000..f067dfce4f --- /dev/null +++ b/tests/unit/storage/test_block_anchor.py @@ -0,0 +1,301 @@ +"""Block content-hash citation anchor tests (svfj).""" + +from __future__ import annotations + +import sqlite3 +from pathlib import Path + +import pytest + +from polylogue.archive.message.roles import Role +from polylogue.core.enums import BlockType, MaterialOrigin, Provider +from polylogue.sources.parsers.base import ParsedContentBlock, ParsedMessage, ParsedSession +from polylogue.storage.block_anchor import ( + BlockAnchor, + InvalidBlockAnchorError, + format_block_anchor, + parse_block_anchor, + resolve_block_anchor, +) +from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_tier +from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier +from polylogue.storage.sqlite.archive_tiers.write import write_parsed_session_to_archive + + +def _connect(path: Path) -> sqlite3.Connection: + conn = sqlite3.connect(path) + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA foreign_keys = ON") + initialize_archive_tier(conn, ArchiveTier.INDEX) + return conn + + +def _anchor_for(conn: sqlite3.Connection, session_id: str, native_message_id: str, position: int) -> BlockAnchor: + row = conn.execute( + """ + SELECT m.message_id, b.content_hash + FROM blocks b + JOIN messages m ON m.message_id = b.message_id + WHERE m.session_id = ? AND m.native_id = ? AND b.position = ? + """, + (session_id, native_message_id, position), + ).fetchone() + assert row is not None + return BlockAnchor( + session_id=session_id, + message_id=str(row["message_id"]), + content_hash_hex=bytes(row["content_hash"]).hex(), + ) + + +def test_format_and_parse_anchor_round_trip() -> None: + anchor = BlockAnchor(session_id="codex-session:abc", message_id="codex-session:abc:m1:0", content_hash_hex="a" * 64) + text = anchor.to_text() + assert text == format_block_anchor(anchor.session_id, anchor.message_id, anchor.content_hash_hex) + assert parse_block_anchor(text) == anchor + + +@pytest.mark.parametrize( + "bad_anchor", + [ + "only-one-part", + "session::message-without-block-part", + "session::message::not-a-block-prefix:deadbeef", + "session::message::block@sha256:tooshort", + "session::message::block@sha256:" + ("g" * 64), # not hex + ], +) +def test_parse_block_anchor_rejects_malformed_input(bad_anchor: str) -> None: + with pytest.raises(InvalidBlockAnchorError): + parse_block_anchor(bad_anchor) + + +def test_resolve_block_anchor_ok_when_unchanged(tmp_path: Path) -> None: + conn = _connect(tmp_path / "index.db") + try: + session = ParsedSession( + source_name=Provider.CODEX, + provider_session_id="anchor-ok", + messages=[ + ParsedMessage( + provider_message_id="m1", + role=Role.ASSISTANT, + material_origin=MaterialOrigin.ASSISTANT_AUTHORED, + blocks=[ParsedContentBlock(type=BlockType.TEXT, text="stable evidence")], + ) + ], + ) + session_id = write_parsed_session_to_archive(conn, session) + anchor = _anchor_for(conn, session_id, "m1", 0) + + resolution = resolve_block_anchor(conn, anchor, position_hint=0) + assert resolution.state == "ok" + assert resolution.resolved_message_id == anchor.message_id + assert resolution.resolved_position == 0 + + # No position hint at all is also ok -- the anchor itself carries no position. + resolution_no_hint = resolve_block_anchor(conn, anchor) + assert resolution_no_hint.state == "ok" + finally: + conn.close() + + +def test_resolve_block_anchor_drifted_position_after_reorder(tmp_path: Path) -> None: + conn = _connect(tmp_path / "index.db") + try: + session = ParsedSession( + source_name=Provider.CODEX, + provider_session_id="anchor-drift-position", + messages=[ + ParsedMessage( + provider_message_id="m1", + role=Role.ASSISTANT, + material_origin=MaterialOrigin.ASSISTANT_AUTHORED, + blocks=[ + ParsedContentBlock(type=BlockType.TEXT, text="first"), + ParsedContentBlock(type=BlockType.TEXT, text="second"), + ], + ) + ], + ) + session_id = write_parsed_session_to_archive(conn, session) + anchor = _anchor_for(conn, session_id, "m1", 0) # anchors "first" at position 0 + + reordered = session.model_copy( + update={ + "messages": [ + ParsedMessage( + provider_message_id="m1", + role=Role.ASSISTANT, + material_origin=MaterialOrigin.ASSISTANT_AUTHORED, + blocks=[ + ParsedContentBlock(type=BlockType.TEXT, text="second"), + ParsedContentBlock(type=BlockType.TEXT, text="first"), + ], + ) + ] + } + ) + write_parsed_session_to_archive(conn, reordered) + + resolution = resolve_block_anchor(conn, anchor, position_hint=0) + assert resolution.state == "drifted_position" + assert resolution.resolved_position == 1 + + # Without a position hint, a moved-but-findable block still resolves ok. + resolution_no_hint = resolve_block_anchor(conn, anchor) + assert resolution_no_hint.state == "ok" + finally: + conn.close() + + +def test_resolve_block_anchor_ambiguous_on_duplicate_evidence(tmp_path: Path) -> None: + conn = _connect(tmp_path / "index.db") + try: + session = ParsedSession( + source_name=Provider.CODEX, + provider_session_id="anchor-ambiguous", + messages=[ + ParsedMessage( + provider_message_id="m1", + role=Role.ASSISTANT, + material_origin=MaterialOrigin.ASSISTANT_AUTHORED, + blocks=[ + ParsedContentBlock(type=BlockType.TEXT, text="same text twice"), + ParsedContentBlock(type=BlockType.TEXT, text="same text twice"), + ], + ) + ], + ) + session_id = write_parsed_session_to_archive(conn, session) + anchor = _anchor_for(conn, session_id, "m1", 0) + + resolution = resolve_block_anchor(conn, anchor) + assert resolution.state == "ambiguous" + assert len(resolution.candidates) == 2 + assert {position for _, position in resolution.candidates} == {0, 1} + finally: + conn.close() + + +def test_resolve_block_anchor_drifted_message_when_content_moves_within_session(tmp_path: Path) -> None: + conn = _connect(tmp_path / "index.db") + try: + session = ParsedSession( + source_name=Provider.CODEX, + provider_session_id="anchor-drift-message", + messages=[ + ParsedMessage( + provider_message_id="m1", + role=Role.ASSISTANT, + material_origin=MaterialOrigin.ASSISTANT_AUTHORED, + blocks=[ParsedContentBlock(type=BlockType.TEXT, text="moved evidence")], + ), + ParsedMessage( + provider_message_id="m2", + role=Role.ASSISTANT, + material_origin=MaterialOrigin.ASSISTANT_AUTHORED, + blocks=[ParsedContentBlock(type=BlockType.TEXT, text="unrelated")], + ), + ], + ) + session_id = write_parsed_session_to_archive(conn, session) + anchor = _anchor_for(conn, session_id, "m1", 0) + + # Re-ingest with the SAME evidence now attached to m2 instead of m1 + # (simulating a provider renumbering messages) -- m1 no longer + # carries any block with this hash, but the session still does. + moved = session.model_copy( + update={ + "messages": [ + ParsedMessage( + provider_message_id="m1", + role=Role.ASSISTANT, + material_origin=MaterialOrigin.ASSISTANT_AUTHORED, + blocks=[ParsedContentBlock(type=BlockType.TEXT, text="replacement content")], + ), + ParsedMessage( + provider_message_id="m2", + role=Role.ASSISTANT, + material_origin=MaterialOrigin.ASSISTANT_AUTHORED, + blocks=[ParsedContentBlock(type=BlockType.TEXT, text="moved evidence")], + ), + ] + } + ) + write_parsed_session_to_archive(conn, moved) + + resolution = resolve_block_anchor(conn, anchor) + assert resolution.state == "drifted_message" + assert resolution.resolved_message_id != anchor.message_id + assert resolution.resolved_position == 0 + finally: + conn.close() + + +def test_resolve_block_anchor_hash_mismatch_never_guesses(tmp_path: Path) -> None: + conn = _connect(tmp_path / "index.db") + try: + session = ParsedSession( + source_name=Provider.CODEX, + provider_session_id="anchor-hash-mismatch", + messages=[ + ParsedMessage( + provider_message_id="m1", + role=Role.ASSISTANT, + material_origin=MaterialOrigin.ASSISTANT_AUTHORED, + blocks=[ParsedContentBlock(type=BlockType.TEXT, text="original content")], + ) + ], + ) + session_id = write_parsed_session_to_archive(conn, session) + anchor = _anchor_for(conn, session_id, "m1", 0) + + rewritten = session.model_copy( + update={ + "messages": [ + ParsedMessage( + provider_message_id="m1", + role=Role.ASSISTANT, + material_origin=MaterialOrigin.ASSISTANT_AUTHORED, + blocks=[ParsedContentBlock(type=BlockType.TEXT, text="rewritten content, same position")], + ) + ] + } + ) + write_parsed_session_to_archive(conn, rewritten) + + resolution = resolve_block_anchor(conn, anchor, position_hint=0) + assert resolution.state == "hash_mismatch" + assert resolution.resolved_message_id == anchor.message_id + assert resolution.resolved_position == 0 + finally: + conn.close() + + +def test_resolve_block_anchor_missing_when_nothing_matches(tmp_path: Path) -> None: + conn = _connect(tmp_path / "index.db") + try: + session = ParsedSession( + source_name=Provider.CODEX, + provider_session_id="anchor-missing", + messages=[ + ParsedMessage( + provider_message_id="m1", + role=Role.ASSISTANT, + material_origin=MaterialOrigin.ASSISTANT_AUTHORED, + blocks=[ParsedContentBlock(type=BlockType.TEXT, text="evidence")], + ) + ], + ) + session_id = write_parsed_session_to_archive(conn, session) + anchor = _anchor_for(conn, session_id, "m1", 0) + + # Delete the session entirely -- the anchor now resolves nowhere. + conn.execute("DELETE FROM sessions WHERE session_id = ?", (session_id,)) + conn.commit() + + resolution = resolve_block_anchor(conn, anchor) + assert resolution.state == "missing" + finally: + conn.close() diff --git a/tests/unit/storage/test_holdout_cohorts.py b/tests/unit/storage/test_holdout_cohorts.py new file mode 100644 index 0000000000..44238daa48 --- /dev/null +++ b/tests/unit/storage/test_holdout_cohorts.py @@ -0,0 +1,282 @@ +from __future__ import annotations + +import sqlite3 + +import pytest + +from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_tier +from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier +from polylogue.storage.sqlite.holdout_cohorts import ( + HoldoutAccessError, + get_holdout_policy, + has_holdout_contamination, + is_holdout, + list_holdout_access_receipts, + mark_holdout, + record_holdout_access, + require_non_holdout_access, +) +from polylogue.storage.sqlite.query_objects import get_result_set, put_query, put_result_set + + +def _conn() -> sqlite3.Connection: + conn = sqlite3.connect(":memory:") + conn.execute("PRAGMA foreign_keys = ON") + initialize_archive_tier(conn, ArchiveTier.USER) + return conn + + +def _seeded_result_set(conn: sqlite3.Connection, *, result_set_id: str = "rs-holdout") -> str: + query = put_query( + conn, + {"field": "origin", "value": "codex-session"}, + grain="session", + lane="dialogue", + rank_policy="mixed", + created_at_ms=1, + ) + put_result_set( + conn, + result_set_id=result_set_id, + query_hash=query.query_hash, + grain="session", + corpus_epoch="epoch-1", + member_refs=("session:a", "session:b"), + exactness="exact", + persistence_class="cohort", + created_at_ms=2, + ) + return result_set_id + + +def test_not_marked_result_set_is_not_a_holdout_and_reads_are_unrestricted() -> None: + conn = _conn() + result_set_id = _seeded_result_set(conn) + + assert is_holdout(conn, result_set_id) is False + require_non_holdout_access(conn, result_set_id, declared_confirmation=False) # no raise + + +def test_mark_holdout_requires_an_existing_result_set() -> None: + conn = _conn() + + with pytest.raises(KeyError): + mark_holdout( + conn, + result_set_id="does-not-exist", + frame="frame-a", + selection_definition={"kind": "seeded"}, + intended_confirmation_use="precision@k evaluation", + authority="operator", + created_epoch="epoch-1", + created_at_ms=10, + ) + + +def test_exploratory_query_cannot_read_a_seeded_holdout() -> None: + conn = _conn() + result_set_id = _seeded_result_set(conn) + mark_holdout( + conn, + result_set_id=result_set_id, + frame="frame-a", + selection_definition={"kind": "seeded"}, + intended_confirmation_use="precision@k evaluation", + authority="operator", + created_epoch="epoch-1", + created_at_ms=10, + ) + + assert is_holdout(conn, result_set_id) is True + with pytest.raises(HoldoutAccessError, match="exploratory queries cannot read"): + require_non_holdout_access(conn, result_set_id, declared_confirmation=False) + + +def test_declared_confirmation_run_can_access_with_a_visible_receipt() -> None: + conn = _conn() + result_set_id = _seeded_result_set(conn) + mark_holdout( + conn, + result_set_id=result_set_id, + frame="frame-a", + selection_definition={"kind": "seeded"}, + intended_confirmation_use="precision@k evaluation", + authority="operator", + created_epoch="epoch-1", + created_at_ms=10, + ) + + require_non_holdout_access(conn, result_set_id, declared_confirmation=True) # no raise + receipt = record_holdout_access( + conn, + receipt_id="receipt-1", + result_set_id=result_set_id, + accessor_ref="agent:confirmation-run", + declared_confirmation=True, + accessed_at_ms=20, + ) + + assert receipt.contamination is False + receipts = list_holdout_access_receipts(conn, result_set_id) + assert len(receipts) == 1 + assert receipts[0].receipt_id == "receipt-1" + assert has_holdout_contamination(conn, result_set_id) is False + + +def test_undeclared_access_marks_permanent_contamination() -> None: + conn = _conn() + result_set_id = _seeded_result_set(conn) + mark_holdout( + conn, + result_set_id=result_set_id, + frame="frame-a", + selection_definition={"kind": "seeded"}, + intended_confirmation_use="precision@k evaluation", + authority="operator", + created_epoch="epoch-1", + created_at_ms=10, + ) + + receipt = record_holdout_access( + conn, + receipt_id="receipt-accident", + result_set_id=result_set_id, + accessor_ref="agent:exploratory-slip", + declared_confirmation=False, + accessed_at_ms=30, + reason="unauthorized exploratory read", + ) + + assert receipt.contamination is True + assert has_holdout_contamination(conn, result_set_id) is True + + # A later declared confirmation access does not retroactively clear it. + record_holdout_access( + conn, + receipt_id="receipt-confirmation", + result_set_id=result_set_id, + accessor_ref="agent:confirmation-run", + declared_confirmation=True, + accessed_at_ms=40, + ) + assert has_holdout_contamination(conn, result_set_id) is True + assert len(list_holdout_access_receipts(conn, result_set_id)) == 2 + + +def test_record_access_requires_the_result_set_to_be_a_holdout() -> None: + conn = _conn() + result_set_id = _seeded_result_set(conn) + + with pytest.raises(KeyError): + record_holdout_access( + conn, + receipt_id="receipt-1", + result_set_id=result_set_id, + accessor_ref="agent:x", + declared_confirmation=True, + accessed_at_ms=1, + ) + + +def test_remarking_with_identical_policy_is_idempotent() -> None: + conn = _conn() + result_set_id = _seeded_result_set(conn) + kwargs: dict[str, object] = { + "result_set_id": result_set_id, + "frame": "frame-a", + "selection_definition": {"kind": "seeded"}, + "intended_confirmation_use": "precision@k evaluation", + "authority": "operator", + "created_epoch": "epoch-1", + "created_at_ms": 10, + } + + first = mark_holdout(conn, **kwargs) # type: ignore[arg-type] + second = mark_holdout(conn, **kwargs) # type: ignore[arg-type] + + assert first == second + + +def test_remarking_with_a_different_policy_raises() -> None: + conn = _conn() + result_set_id = _seeded_result_set(conn) + mark_holdout( + conn, + result_set_id=result_set_id, + frame="frame-a", + selection_definition={"kind": "seeded"}, + intended_confirmation_use="precision@k evaluation", + authority="operator", + created_epoch="epoch-1", + created_at_ms=10, + ) + + with pytest.raises(ValueError, match="different declared policy"): + mark_holdout( + conn, + result_set_id=result_set_id, + frame="frame-b", + selection_definition={"kind": "seeded"}, + intended_confirmation_use="precision@k evaluation", + authority="operator", + created_epoch="epoch-1", + created_at_ms=10, + ) + + +def test_deleting_a_holdout_marked_result_set_is_blocked_by_the_durable_fk() -> None: + """rxdo.9.4 AC: 'reset/excision preserve the declared durability + semantics.' No excision mechanism exists for result_sets yet (see + migration 009's ON DELETE RESTRICT), so the durable floor this PR ships + is DB-level: a holdout-marked result set cannot be deleted out from + under its policy row. A future excision mechanism must route through an + explicit unmark-then-delete step rather than a raw DELETE, or it will + hit this same IntegrityError.""" + + conn = _conn() + result_set_id = _seeded_result_set(conn) + mark_holdout( + conn, + result_set_id=result_set_id, + frame="frame-a", + selection_definition={"kind": "seeded"}, + intended_confirmation_use="precision@k evaluation", + authority="operator", + created_epoch="epoch-1", + created_at_ms=10, + ) + + with pytest.raises(sqlite3.IntegrityError): + conn.execute("DELETE FROM result_sets WHERE result_set_id = ?", (result_set_id,)) + + # Contrast: a non-holdout result set has no such protection -- the + # restriction is specific to the holdout policy, not a blanket + # disallow-all-deletes on result_sets. + other_id = _seeded_result_set(conn, result_set_id="rs-no-holdout") + conn.execute("DELETE FROM result_sets WHERE result_set_id = ?", (other_id,)) + assert get_result_set(conn, other_id) is None + + +def test_cohort_and_holdout_relation_identities_remain_distinct_while_sharing_the_manifest() -> None: + """rxdo.9.4 design: holdout is a policy layered on the existing + result_sets manifest, not a second relation type -- the cohort keeps its + own persistence_class identity even after being marked as a holdout.""" + + conn = _conn() + result_set_id = _seeded_result_set(conn, result_set_id="rs-cohort-holdout") + + mark_holdout( + conn, + result_set_id=result_set_id, + frame="frame-a", + selection_definition={"kind": "seeded"}, + intended_confirmation_use="precision@k evaluation", + authority="operator", + created_epoch="epoch-1", + created_at_ms=10, + ) + + row = conn.execute("SELECT persistence_class FROM result_sets WHERE result_set_id = ?", (result_set_id,)).fetchone() + assert row is not None + assert row[0] == "cohort" + assert get_holdout_policy(conn, result_set_id) is not None