From f48e547d3d6bd39775d346bf4b755aa650f3a2fc Mon Sep 17 00:00:00 2001 From: Sinity Date: Wed, 5 Aug 2026 09:38:15 +0200 Subject: [PATCH 1/2] feat(reads): wire intended evidence modules Problem Three modules deleted in PR #3390 were unreachable despite current architecture and project intent still describing their capabilities. The holdout guard also lacked a production enforcement seam. What changed Restore and wire the holdout policy into DurableRefResolver before retained result-set members are read, expose private Fable packet regeneration through the async archive facade, and resolve content-hash block anchors through Polylogue.resolve_ref. Add real-route tests and restore the empty delegation predicate path needed by cold Fable regeneration. Update API parity and generated documentation for the new facade operation. Disposition - storage/sqlite/holdout_cohorts.py: restore and wire. Bead polylogue-rxdo.9.4, PR #2888, and current planner architecture establish this as an adopted analysis-integrity policy. The production MCP reference query route now fails closed for exploratory reads of marked holdouts. - insights/fable_packet.py: restore and wire. Bead polylogue-212.9.1 and PR #2814 establish the private descriptive packet as intended capability. The async archive facade supplies the missing operator-facing production route. - storage/block_anchor.py: restore and wire. Bead polylogue-svfj and PR #2588 establish content-hash citation anchors as intended capability. Polylogue.resolve_ref now provides the existing public read surface; deferred lineage relocation states remain unchanged. Alternatives rejected Deleting all three based only on their former lack of callers would discard capabilities still claimed by the architecture and tracking history. Archive migration and lifecycle work was intentionally excluded. Compatibility/migration No schema or migration changes. The holdout guard is default-deny for exploratory retained-result reads; confirmation callers may explicitly declare confirmation through DurableRefResolver. The Fable route remains fail-closed when structural or annotation evidence is insufficient. Co-Authored-By: Claude --- docs/generated/api-operation-parity.json | 5 + docs/library-api.md | 1 + docs/plans/layering-surface-baseline.json | 5 + polylogue/api/archive.py | 77 +++++ polylogue/api/operation_parity.py | 1 + polylogue/archive/query/evaluator.py | 14 + polylogue/insights/fable_packet.py | 313 ++++++++++++++++++ polylogue/storage/block_anchor.py | 231 +++++++++++++ .../storage/sqlite/archive_tiers/archive.py | 3 +- polylogue/storage/sqlite/holdout_cohorts.py | 259 +++++++++++++++ tests/unit/api/test_facade_contracts.py | 25 ++ tests/unit/archive/query/test_evaluator.py | 40 +++ tests/unit/insights/test_fable_packet.py | 75 +++++ .../unit/mcp/test_reference_query_pipeline.py | 51 ++- tests/unit/storage/test_block_anchor.py | 301 +++++++++++++++++ tests/unit/storage/test_holdout_cohorts.py | 282 ++++++++++++++++ 16 files changed, 1681 insertions(+), 2 deletions(-) create mode 100644 polylogue/insights/fable_packet.py create mode 100644 polylogue/storage/block_anchor.py create mode 100644 polylogue/storage/sqlite/holdout_cohorts.py create mode 100644 tests/unit/insights/test_fable_packet.py create mode 100644 tests/unit/storage/test_block_anchor.py create mode 100644 tests/unit/storage/test_holdout_cohorts.py 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..c7e6868c4a 100644 --- a/tests/unit/api/test_facade_contracts.py +++ b/tests/unit/api/test_facade_contracts.py @@ -50,6 +50,7 @@ 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 @@ -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,17 @@ 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() + + @pytest.mark.parametrize( "method_name", sorted(MUTATION_BY_ID_RAISES_METHODS), @@ -2706,6 +2719,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 From 22e1ee02f94be5eeb1b1b5ab733f6c9d05b70ead Mon Sep 17 00:00:00 2001 From: test Date: Wed, 5 Aug 2026 09:51:15 +0200 Subject: [PATCH 2/2] test(api): exercise Fable facade regeneration Problem The initial Fable facade test covered only an empty archive and therefore did not prove that the restored adapter could compose canonical delegations with durable annotation evidence. What changed Add a real archive-backed facade test that writes a delegation through the production archive writer, initializes the durable built-in schema, persists an active evidence-backed annotation, and asserts a complete packet with the selected delegation and distribution. Adversarial review The independent review identified the empty-only coverage as a legitimate gap. This test closes it. Removing the delegation query, durable schema lookup, or assertion query changes the complete result to not_supported or an empty population. Compatibility/migration Test-only change. No production schema, migration, or lifecycle behavior changed. Co-Authored-By: Claude --- tests/unit/api/test_facade_contracts.py | 63 ++++++++++++++++++++++++- 1 file changed, 62 insertions(+), 1 deletion(-) diff --git a/tests/unit/api/test_facade_contracts.py b/tests/unit/api/test_facade_contracts.py index c7e6868c4a..b7a61fc926 100644 --- a/tests/unit/api/test_facade_contracts.py +++ b/tests/unit/api/test_facade_contracts.py @@ -53,7 +53,7 @@ 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 @@ -1860,6 +1860,67 @@ async def test_regenerate_private_fable_packet_is_an_archive_backed_facade_route 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),