diff --git a/docs/plans/topology-target.yaml b/docs/plans/topology-target.yaml index bbc435aa80..dae5b8c364 100644 --- a/docs/plans/topology-target.yaml +++ b/docs/plans/topology-target.yaml @@ -41,7 +41,7 @@ files: owner: stable cross_cut: { api: async } - path: polylogue/api/archive.py - loc: 5813 + loc: 5891 target: polylogue/api/archive.py owner: stable cross_cut: { api: async } @@ -382,10 +382,15 @@ files: owner: archive-query reason: archive-domain query semantics - path: polylogue/archive/query/predicate.py - loc: 245 + loc: 389 target: polylogue/archive/query/predicate.py owner: archive-query reason: archive-domain query semantics + - path: polylogue/archive/query/production_evaluator.py + loc: 244 + target: polylogue/archive/query/production_evaluator.py + owner: archive-query + reason: archive-domain query semantics - path: polylogue/archive/query/retrieval.py loc: 49 target: polylogue/archive/query/retrieval.py @@ -1415,7 +1420,7 @@ files: target: polylogue/daemon/convergence_debt_status.py owner: stable - path: polylogue/daemon/convergence_stages.py - loc: 1675 + loc: 1677 target: polylogue/daemon/convergence_stages.py owner: stable - path: polylogue/daemon/convergence_standing_queries.py @@ -1745,6 +1750,10 @@ files: loc: 216 target: polylogue/insights/feedback.py owner: stable + - path: polylogue/insights/improvement_loops.py + loc: 215 + target: polylogue/insights/improvement_loops.py + owner: stable - path: polylogue/insights/otlp_correlation.py loc: 509 target: polylogue/insights/otlp_correlation.py @@ -3740,6 +3749,10 @@ files: loc: 289 target: polylogue/storage/sqlite/connection_profile.py owner: stable + - path: polylogue/storage/sqlite/finding_provenance.py + loc: 134 + target: polylogue/storage/sqlite/finding_provenance.py + owner: stable - path: polylogue/storage/sqlite/lifecycle.py loc: 250 target: polylogue/storage/sqlite/lifecycle.py @@ -4022,7 +4035,7 @@ files: target: polylogue/surfaces/chronicle.py owner: stable - path: polylogue/surfaces/payloads.py - loc: 3569 + loc: 3609 target: polylogue/surfaces/payloads.py owner: stable - path: polylogue/surfaces/projection_spec.py diff --git a/docs/topology-status.md b/docs/topology-status.md index a72d1a0e53..6df5d46388 100644 --- a/docs/topology-status.md +++ b/docs/topology-status.md @@ -19,7 +19,7 @@ Generated by `devtools render topology-status`. Reads `docs/plans/topology-targe | archive-phase | — | 2 | 2 | 0 | 0 | | archive-projection | — | 5 | 5 | 0 | 0 | | archive-provider | — | 2 | 2 | 0 | 0 | -| archive-query | archive query semantics | 27 | 27 | 0 | 0 | +| archive-query | archive query semantics | 28 | 28 | 0 | 0 | | archive-raw-payload | — | 5 | 5 | 0 | 0 | | archive-semantic | — | 12 | 12 | 0 | 0 | | archive-session | — | 18 | 18 | 0 | 0 | @@ -28,12 +28,12 @@ Generated by `devtools render topology-status`. Reads `docs/plans/topology-targe ### Summary -- **Stable** (no move scoped): 817 +- **Stable** (no move scoped): 819 - **Kernel** (polylogue/ root): 10 - **Primitives** (storage-root): 18 - **TBD** (cell needs explicit assignment): 6 -- **Total declared**: 974 -- **Realized polylogue/**/*.py**: 974 files declared +- **Total declared**: 977 +- **Realized polylogue/**/*.py**: 977 files declared ### TBD cells (require explicit routing) diff --git a/polylogue/api/archive.py b/polylogue/api/archive.py index f7305cae2a..a5937f81df 100644 --- a/polylogue/api/archive.py +++ b/polylogue/api/archive.py @@ -358,9 +358,7 @@ def _invalid_unicode_ref_payload(ref: str) -> Any | None: #: (polylogue-rxdo analysis-provenance epic). ``resolve_ref`` returns a typed #: ``PendingObjectRefPayload`` (reason=substrate-pending) for these instead of #: attempting a lookup against tables that do not exist yet. -_PENDING_OBJECT_REF_KINDS: frozenset[str] = frozenset( - {"query", "query-run", "result-set", "finding", "cohort", "analysis"} -) +_PENDING_OBJECT_REF_KINDS: frozenset[str] = frozenset({"query", "query-run", "result-set", "cohort", "analysis"}) def _pending_ref_payload(ref: str, normalized_ref: str, kind: str) -> Any: @@ -3020,6 +3018,8 @@ async def resolve_ref(self, ref: str) -> PublicRefResolutionPayload: return self._resolve_block_object_ref(archive, ref, normalized_ref, object_ref, evidence_ref) if object_ref.kind == "assertion": return self._resolve_assertion_object_ref(archive_root, ref, normalized_ref, object_ref) + if object_ref.kind == "finding": + return self._resolve_finding_object_ref(archive_root, ref, normalized_ref, object_ref) if object_ref.kind == "annotation-batch": return self._resolve_annotation_batch_object_ref(archive, ref, normalized_ref, object_ref) if object_ref.kind == "delegation": @@ -3243,6 +3243,84 @@ def _resolve_assertion_object_ref( actions=(_resolution_action("list assertion target", f"polylogue find {payload.target_ref} then read"),), ) + def _resolve_finding_object_ref( + self, + archive_root: Path, + ref: str, + normalized_ref: str, + object_ref: ObjectRef, + ) -> PublicRefResolutionPayload: + from polylogue.storage.sqlite.finding_provenance import compute_finding_provenance + from polylogue.surfaces.payloads import ( + FindingEvidenceRefState, + FindingProvenancePayload, + PublicRefResolutionPayload, + model_json_document, + ) + + user_db = archive_root / "user.db" + if not user_db.exists(): + return cast( + PublicRefResolutionPayload, + _unresolved_ref_payload(ref, "finding not found", normalized_ref=normalized_ref, kind="finding"), + ) + with closing(sqlite3.connect(user_db)) as conn: + conn.row_factory = sqlite3.Row + provenance = compute_finding_provenance(conn, object_ref.object_id) + if provenance is None: + return cast( + PublicRefResolutionPayload, + _unresolved_ref_payload(ref, "finding not found", normalized_ref=normalized_ref, kind="finding"), + ) + payload = FindingProvenancePayload( + assertion_id=provenance.assertion_id, + claim_key=provenance.claim_key, + target_ref=provenance.target_ref, + finding_kind=provenance.finding_kind, + query_ref=provenance.query_ref, + result_set_ref=provenance.result_set_ref, + baseline_ref=provenance.baseline_ref, + current_ref=provenance.current_ref, + detector_ref=provenance.detector_ref, + status=AssertionStatus.from_string(provenance.status), + evidence=tuple( + FindingEvidenceRefState(ref=item.ref, resolvable=item.resolvable, reason=item.reason) + for item in provenance.evidence + ), + staleness_verdict=provenance.staleness_verdict, + created_at_ms=provenance.created_at_ms, + updated_at_ms=provenance.updated_at_ms, + ) + caveats: tuple[str, ...] = () + if provenance.staleness_verdict != "current": + caveats = (f"finding evidence staleness verdict: {provenance.staleness_verdict}",) + object_refs = tuple( + dict.fromkeys( + ref_value + for ref_value in ( + normalized_ref, + provenance.target_ref, + provenance.query_ref, + provenance.result_set_ref, + ) + if ref_value + ) + ) + return PublicRefResolutionPayload( + ref=ref, + normalized_ref=normalized_ref, + kind="finding", + resolved=True, + payload_kind="finding-provenance", + payload=model_json_document(payload), + title=provenance.claim_key or provenance.finding_kind or "finding", + summary=f"{provenance.finding_kind or 'finding'} ({provenance.status})", + object_refs=object_refs, + evidence_refs=tuple(item.ref for item in provenance.evidence), + caveats=caveats, + actions=(_resolution_action("list target evidence", f"polylogue find {provenance.target_ref} then read"),), + ) + def _resolve_annotation_batch_object_ref( self, archive: Any, diff --git a/polylogue/archive/query/predicate.py b/polylogue/archive/query/predicate.py index ec2840e19b..deb463de92 100644 --- a/polylogue/archive/query/predicate.py +++ b/polylogue/archive/query/predicate.py @@ -2,9 +2,10 @@ from __future__ import annotations +from collections.abc import Mapping, Sequence from dataclasses import dataclass from dataclasses import field as dataclass_field -from typing import Literal, TypeAlias +from typing import Literal, TypeAlias, cast QueryBoolOp: TypeAlias = Literal["and", "or"] QueryCompareOp: TypeAlias = Literal["=", ">", ">=", "<", "<="] @@ -22,6 +23,13 @@ ] QuerySequenceConstraintKind: TypeAlias = Literal["ordered", "next", "within"] +_EXISTS_UNITS: frozenset[str] = frozenset( + {"message", "action", "block", "assertion", "file", "run", "observed-event", "context-snapshot", "delegation"} +) +_COMPARE_OPS: frozenset[str] = frozenset({"=", ">", ">=", "<", "<="}) +_FIELD_SCOPES: frozenset[str] = frozenset({"session", "unit"}) +_SEQUENCE_CONSTRAINT_KINDS: frozenset[str] = frozenset({"ordered", "next", "within"}) + @dataclass(frozen=True) class QuerySequenceConstraint: @@ -225,6 +233,141 @@ def to_payload(self) -> dict[str, object]: ) +def _field_ref_from_payload(payload: object) -> QueryFieldRef: + if not isinstance(payload, Mapping): + raise ValueError("field_ref payload must be an object") + scope = payload.get("scope") + name = payload.get("name") + source_name = payload.get("source_name") + unit = payload.get("unit") + if scope not in _FIELD_SCOPES: + raise ValueError(f"unsupported field_ref scope: {scope!r}") + if not isinstance(name, str) or not name: + raise ValueError("field_ref requires a non-empty 'name'") + if not isinstance(source_name, str) or not source_name: + raise ValueError("field_ref requires a non-empty 'source_name'") + if unit is not None and not isinstance(unit, str): + raise ValueError("field_ref 'unit' must be a string when present") + return QueryFieldRef( + scope=cast(QueryFieldScope, scope), + name=name, + source_name=source_name, + unit=unit, + ) + + +def _sequence_constraint_from_payload(payload: object) -> QuerySequenceConstraint: + if not isinstance(payload, Mapping): + raise ValueError("sequence constraint payload must be an object") + kind = payload.get("kind", "ordered") + within_ms = payload.get("within_ms") + if kind not in _SEQUENCE_CONSTRAINT_KINDS: + raise ValueError(f"unsupported sequence constraint kind: {kind!r}") + if within_ms is not None and (isinstance(within_ms, bool) or not isinstance(within_ms, int)): + raise ValueError("sequence constraint 'within_ms' must be an integer") + return QuerySequenceConstraint(kind=cast(QuerySequenceConstraintKind, kind), within_ms=within_ms) + + +def _payload_list(payload: object, *, field: str) -> Sequence[object]: + if not isinstance(payload, Sequence) or isinstance(payload, (str, bytes)): + raise ValueError(f"{field!r} must be a list") + return payload + + +def predicate_from_payload(payload: Mapping[str, object]) -> QueryPredicate: + """Reconstruct a typed predicate from its own ``to_payload()`` projection. + + Every branch below inverts one dataclass's own lossless ``to_payload()`` + mapping (see the corresponding ``to_payload`` above each predicate class + in this module), so round-tripping a value through ``to_payload`` then + ``predicate_from_payload`` always reproduces an equal predicate. This is + deliberately *not* a reverse-compiler over free-form or legacy text: it + only understands the closed, versioned shape this module itself emits + (``polylogue.query-definition.v1``). Callers that hold a legacy protocol + v0 canonical plan (an opaque saved-view JSON request, not this predicate + grammar) must not route it through this function -- see + ``polylogue.core.query_identity.require_supported_definition_protocol_version`` + and ``polylogue.archive.query.production_evaluator``, which fails closed + on v0 identities before reaching here. + """ + + if not isinstance(payload, Mapping): + raise ValueError("predicate payload must be an object") + kind = payload.get("kind") + if kind == "field": + field = payload.get("field") + op = payload.get("op", "=") + values = payload.get("values", ()) + if not isinstance(field, str) or not field: + raise ValueError("field predicate requires a non-empty 'field'") + if op not in _COMPARE_OPS: + raise ValueError(f"unsupported field predicate op: {op!r}") + raw_values = _payload_list(values, field="values") + if not all(isinstance(value, str) for value in raw_values): + raise ValueError("field predicate 'values' must be a list of strings") + predicate: QueryFieldPredicate = QueryFieldPredicate( + field=field, + values=tuple(cast(str, value) for value in raw_values), + op=cast(QueryCompareOp, op), + ) + field_ref_payload = payload.get("field_ref") + if field_ref_payload is not None: + predicate = predicate.with_field_ref(_field_ref_from_payload(field_ref_payload)) + return predicate + if kind == "not": + child = payload.get("child") + if not isinstance(child, Mapping): + raise ValueError("not predicate requires a 'child' object") + return QueryNotPredicate(predicate_from_payload(child)) + if kind in ("and", "or"): + children = _payload_list(payload.get("children"), field="children") + parsed_children: list[QueryPredicate] = [] + for child in children: + if not isinstance(child, Mapping): + raise ValueError("boolean predicate children must be objects") + parsed_children.append(predicate_from_payload(child)) + return QueryBoolPredicate(kind, tuple(parsed_children)) + if kind == "exists": + unit = payload.get("unit") + child = payload.get("child") + if unit not in _EXISTS_UNITS: + raise ValueError(f"unsupported exists unit: {unit!r}") + if not isinstance(child, Mapping): + raise ValueError("exists predicate requires a 'child' object") + return QueryExistsPredicate(unit=cast(QueryExistsUnit, unit), child=predicate_from_payload(child)) + if kind == "sequence": + steps_payload = _payload_list(payload.get("steps", ()), field="steps") + parsed_steps: list[QueryPredicate] = [] + for step in steps_payload: + if not isinstance(step, Mapping): + raise ValueError("sequence predicate steps must be objects") + parsed_steps.append(predicate_from_payload(step)) + constraints_payload = payload.get("constraints") + constraints: tuple[QuerySequenceConstraint, ...] = () + if constraints_payload is not None: + constraints = tuple( + _sequence_constraint_from_payload(item) + for item in _payload_list(constraints_payload, field="constraints") + ) + return QuerySequencePredicate(steps=tuple(parsed_steps), constraints=constraints) + if kind == "fts": + text = payload.get("text") + if not isinstance(text, str) or not text: + raise ValueError("fts predicate requires non-empty 'text'") + return QueryTextPredicate(text=text) + if kind == "semantic": + text = payload.get("text") + if not isinstance(text, str) or not text: + raise ValueError("semantic predicate requires non-empty 'text'") + return QuerySemanticPredicate(text=text) + if kind == "lineage": + seed = payload.get("seed_session_id") + if not isinstance(seed, str) or not seed: + raise ValueError("lineage predicate requires non-empty 'seed_session_id'") + return QueryLineagePredicate(seed_session_id=seed) + raise ValueError(f"unsupported predicate payload kind: {kind!r}") + + __all__ = [ "QueryBoolOp", "QueryBoolPredicate", @@ -242,4 +385,5 @@ def to_payload(self) -> dict[str, object]: "QuerySequenceConstraint", "QuerySequenceConstraintKind", "QueryTextPredicate", + "predicate_from_payload", ] diff --git a/polylogue/archive/query/production_evaluator.py b/polylogue/archive/query/production_evaluator.py new file mode 100644 index 0000000000..1a64d05cc7 --- /dev/null +++ b/polylogue/archive/query/production_evaluator.py @@ -0,0 +1,244 @@ +"""Real (non-test-double) planner implementation for canonical query plans. + +Before this module existed, :class:`~polylogue.archive.query.evaluator. +CanonicalPlanEvaluator` had exactly one kind of implementation anywhere in the +repository: hand-rolled fakes inside test files. ``ArchiveCanonicalPlanEvaluator`` +is the first production evaluator -- it turns a durable :class:`QueryObject`'s +canonical, protocol-versioned AST back into an executable +:class:`~polylogue.archive.query.plan.SessionQueryPlan` and runs it through the +same :class:`~polylogue.archive.filter.filters.SessionFilter` every other +surface (CLI/MCP/API) uses, then records a bounded ``ops.db`` ``query_runs`` +telemetry row for the execution (surface ``daemon-internal``). + +Only the ``session`` grain and the current (v1) definition protocol are +evaluated. Legacy protocol v0 identities (opaque saved-view JSON, not this +predicate grammar) and non-session grains fail closed with a named error +rather than guessing -- this mirrors the "never reverse-compile lossy +identity JSON" doctrine: this module only inverts the *same* typed +``to_payload()`` shape ``polylogue.archive.query.predicate`` itself owns. +""" + +from __future__ import annotations + +import sqlite3 +import time +import uuid +from contextlib import closing +from importlib.metadata import PackageNotFoundError +from importlib.metadata import version as _package_version +from pathlib import Path +from typing import Literal + +from polylogue.archive.filter.filters import SessionFilter +from polylogue.archive.query.evaluator import ( + CanonicalPlanEvaluator, + QueryEvaluation, + QueryEvaluationRequest, +) +from polylogue.archive.query.expression import RefOperand +from polylogue.archive.query.plan import SessionQueryPlan +from polylogue.archive.query.predicate import predicate_from_payload +from polylogue.core.query_identity import LEGACY_QUERY_DEFINITION_PROTOCOL_VERSION +from polylogue.logging import get_logger +from polylogue.storage.sqlite.query_objects import EvaluationReceipt + +logger = get_logger(__name__) + +_SUPPORTED_GRAINS: frozenset[str] = frozenset({"session"}) + + +class LegacyQueryDefinitionNotExecutableError(ValueError): + """A protocol-v0 (legacy saved-view) query has no executable planner form.""" + + +class UnsupportedEvaluationGrainError(ValueError): + """The production evaluator does not yet execute this relation grain.""" + + +def _polylogue_runtime_build_ref() -> str: + try: + return f"polylogue:{_package_version('polylogue')}" + except PackageNotFoundError: + return "polylogue:unknown" + + +def _pragma_user_version(conn: sqlite3.Connection) -> int: + row = conn.execute("PRAGMA user_version").fetchone() + return int(row[0]) if row is not None else 0 + + +def _tier_generation(db_path: Path, *, label: str) -> str: + if not db_path.exists(): + return f"{label}:absent" + try: + with closing(sqlite3.connect(f"file:{db_path}?mode=ro", uri=True, timeout=5.0)) as conn: + return f"{label}:v{_pragma_user_version(conn)}" + except sqlite3.Error: + logger.warning("production-evaluator: could not read %s generation", label, exc_info=True) + return f"{label}:unknown" + + +def _index_epoch(index_db: Path) -> str: + """Index generation identifier: schema version plus a session watermark.""" + if not index_db.exists(): + return "index:absent" + try: + with closing(sqlite3.connect(f"file:{index_db}?mode=ro", uri=True, timeout=5.0)) as conn: + version = _pragma_user_version(conn) + row = conn.execute("SELECT MAX(updated_at_ms) FROM sessions").fetchone() + watermark = int(row[0]) if row is not None and row[0] is not None else 0 + return f"index:v{version}:{watermark}" + except sqlite3.Error: + logger.warning("production-evaluator: could not read index epoch", exc_info=True) + return "index:unknown" + + +class ArchiveCanonicalPlanEvaluator(CanonicalPlanEvaluator): + """Evaluate durable canonical query definitions against the live archive. + + ``db_path`` is the ``index.db`` path (the same convention used throughout + ``daemon/convergence_stages.py``); ``source.db``/``user.db``/``ops.db`` + are resolved as siblings. + """ + + def __init__(self, db_path: Path, *, surface: str = "daemon-internal") -> None: + self._db_path = db_path + self._archive_root = db_path.parent + self._surface = surface + + def evaluate(self, request: QueryEvaluationRequest) -> QueryEvaluation: + query = request.query + if query.definition_protocol_version == LEGACY_QUERY_DEFINITION_PROTOCOL_VERSION: + raise LegacyQueryDefinitionNotExecutableError( + f"query:{query.query_hash} uses the legacy protocol-v0 definition; " + "legacy saved-view identities predate the executable predicate grammar " + "and cannot be re-evaluated by the planner" + ) + if query.grain not in _SUPPORTED_GRAINS: + raise UnsupportedEvaluationGrainError( + f"query:{query.query_hash} has grain {query.grain!r}; the production " + f"evaluator currently executes only {sorted(_SUPPORTED_GRAINS)!r}" + ) + ast = query.canonical_plan.get("ast") + if not isinstance(ast, dict): + raise ValueError(f"query:{query.query_hash} canonical plan has no executable 'ast'") + + from polylogue.archive.query.expression import _bind_predicate_context # planner-internal seam + + predicate = predicate_from_payload(ast) + bound = _bind_predicate_context(predicate, unit="session") + plan = SessionQueryPlan(boolean_predicate=bound) + session_filter = SessionFilter.from_query_plan(plan, archive_root=self._archive_root) + + from polylogue.api.sync.bridge import run_coroutine_sync + + started_at_ms = int(time.time() * 1000) + try: + # NOT list_summaries(): that caps at the default page limit (50, + # see SessionFilter.list_summaries docstring) and this evaluation + # is unconditionally labeled exactness="exact" below. A watched + # query matching more than 50 sessions would silently drop + # members past the first page from both the returned evaluation + # and the query_runs telemetry row while still claiming an exact + # enumeration -- and standing-query drift detection computes its + # membership merkle root directly from member_refs, so it would + # never notice new members added past the cap. list_all_summaries + # resolves every matching summary (default_limit=1_000_000, + # functionally unbounded for any real archive), mirroring the + # same exact-enumeration requirement count_archive/delete/mark + # already rely on (#1873). + summaries = run_coroutine_sync(session_filter.list_all_summaries()) + except Exception: + logger.warning("production-evaluator: evaluation failed for query:%s", query.query_hash, exc_info=True) + raise + duration_ms = int(time.time() * 1000) - started_at_ms + + excluded_prefixes = tuple(request.excluded_origin_prefixes) + member_refs = tuple( + f"session:{summary.id}" + for summary in summaries + if not any(str(summary.origin).startswith(prefix) for prefix in excluded_prefixes) + ) + excluded_refs = set(request.excluded_scope_refs) + if excluded_refs: + member_refs = tuple(ref for ref in member_refs if ref not in excluded_refs) + + index_generation = _index_epoch(self._db_path) + receipt = EvaluationReceipt( + receipt_id=f"receipt-{uuid.uuid4().hex}", + source_generation=_tier_generation(self._archive_root / "source.db", label="source"), + user_generation=_tier_generation(self._archive_root / "user.db", label="user"), + index_generation=index_generation, + runtime_build_ref=_polylogue_runtime_build_ref(), + ) + evaluation = QueryEvaluation( + grain="session", + member_refs=member_refs, + corpus_epoch=index_generation, + exactness="exact", + receipt=receipt, + ) + self._record_query_run( + query_hash=query.query_hash, + purpose=request.purpose, + started_at_ms=started_at_ms, + duration_ms=duration_ms, + evaluation=evaluation, + ) + return evaluation + + def resolve_cohort(self, operand: RefOperand) -> QueryEvaluation: + raise NotImplementedError( + f"cohort substrate is not implemented yet ({operand.reference.format()}); see polylogue-rxdo.6" + ) + + def _record_query_run( + self, + *, + query_hash: str, + purpose: str, + started_at_ms: int, + duration_ms: int, + evaluation: QueryEvaluation, + ) -> None: + """Best-effort ops-tier telemetry. Never lets a recording failure fail a read.""" + ops_db = self._archive_root / "ops.db" + if not ops_db.exists(): + return + try: + from polylogue.storage.sqlite.archive_tiers.ops_write import record_query_run + + with closing(sqlite3.connect(ops_db, timeout=5.0)) as conn: + record_query_run( + conn, + run_id=f"qr_{uuid.uuid4().hex}", + query_hash=query_hash, + actor=None, + surface=self._surface, + verb=purpose, + request=None, + lowered_spec=None, + archive_epoch=evaluation.corpus_epoch, + started_at_ms=started_at_ms, + duration_ms=duration_ms, + status="ok", + degraded=None, + unit=evaluation.grain, + member_count=len(evaluation.member_refs), + exactness=evaluation.exactness, + result_fingerprint=None, + sample_refs=evaluation.member_refs[:20], + ) + except Exception: + logger.warning("production-evaluator: query_run recording failed", exc_info=True) + + +Surface = Literal["cli", "mcp", "daemon-web", "api", "daemon-internal"] + + +__all__ = [ + "ArchiveCanonicalPlanEvaluator", + "LegacyQueryDefinitionNotExecutableError", + "Surface", + "UnsupportedEvaluationGrainError", +] diff --git a/polylogue/daemon/convergence_stages.py b/polylogue/daemon/convergence_stages.py index e771b18e80..2f01d2d666 100644 --- a/polylogue/daemon/convergence_stages.py +++ b/polylogue/daemon/convergence_stages.py @@ -556,11 +556,13 @@ def execute_sessions(session_ids: Sequence[str]) -> StageExecuteReturn: def make_default_convergence_stages(db_path: Path) -> tuple[ConvergenceStage, ...]: """Build the daemon's default post-ingest convergence stage set.""" + from polylogue.archive.query.production_evaluator import ArchiveCanonicalPlanEvaluator + return ( make_fts_stage(db_path), make_embed_stage(db_path), make_insights_stage(db_path), - make_standing_query_stage(db_path), + make_standing_query_stage(db_path, evaluator=ArchiveCanonicalPlanEvaluator(db_path)), ) diff --git a/polylogue/insights/improvement_loops.py b/polylogue/insights/improvement_loops.py new file mode 100644 index 0000000000..a17a20528d --- /dev/null +++ b/polylogue/insights/improvement_loops.py @@ -0,0 +1,215 @@ +"""Declarative registry of improvement-loop specs (polylogue-rxdo.11). + +Every closed-loop mechanism in the epic's design is the same 5-tuple: watch +(a standing query or other signal source) -> measure (a content-addressed +metric) -> propose (a recipe emitting candidates, never auto-applying) -> +judge (the existing assertion judgment lifecycle) -> bump (a content-addressed +artifact version). This module is the "declare loops like insight +descriptors in one LOOP_REGISTRY" contract: a single place naming every loop +instance and its five parts, so loop health becomes a queryable fact instead +of scattered prose across beads. + +Scope note (read before adding an ``active`` entry): the corrective +acceptance criteria for polylogue-rxdo.11 require the first two pilots (L1 +recall-relevance, L2 classifier-residue) to *execute* through one shared +scheduler/state contract, not merely be declared. That scheduler does not +exist yet, and L1's own signal source (polylogue-37t.17's read-access log) +is itself unimplemented. Every entry below is therefore ``status="horizon"`` +-- this registry is the declaration surface the eventual scheduler will read, +not a claim that any loop is running. Flipping an entry to ``"active"`` +requires the shared scheduler/state module plus that loop's own watch/measure/ +propose/judge/bump wiring to exist and be tested; do not flip it to make this +registry look more complete than it is. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Literal + +LoopStatus = Literal["horizon", "active"] + + +@dataclass(frozen=True, slots=True) +class ImprovementLoopSpec: + """One declared instance of the watch/measure/propose/judge/bump 5-tuple.""" + + loop_id: str + title: str + watch: str + """What signal source the loop observes (a standing query ref, event stream, etc.).""" + measure: str + """The content-addressed metric/definition the loop reduces its signal to.""" + propose: str + """What produces candidates from the measurement. Never auto-applies.""" + judge: str + """The gate that turns a candidate into an accepted change (existing assertion lifecycle unless noted).""" + artifact: str + """What content-addressed artifact the loop bumps a version of when judged.""" + status: LoopStatus + implementation_ref: str | None = None + """Bead id (or module) that owns this loop's concrete wiring, if any.""" + + +LOOP_REGISTRY: dict[str, ImprovementLoopSpec] = { + spec.loop_id: spec + for spec in ( + ImprovementLoopSpec( + loop_id="L1", + title="Recall relevance", + watch="context-delivery receipts (injected refs) + downstream usage (cite/quote/re-read)", + measure="per-item usage rate", + propose="retrieval ranker reweighting from implicit relevance feedback", + judge="operator/agent judgment lifecycle (assertion candidate -> accepted)", + artifact="ranker:", + status="horizon", + implementation_ref="polylogue-37t.17", + ), + ImprovementLoopSpec( + loop_id="L2", + title="Classifier residue", + watch="PACK-A/B classifier 'unclassified residue' confessions", + measure="residue volume per command shape (standing query)", + propose="agent-drafted rules for the top residue clusters", + judge="operator/agent judgment lifecycle", + artifact="classifier:", + status="horizon", + ), + ImprovementLoopSpec( + loop_id="L3", + title="Judge calibration", + watch="agent-judge vs operator-gold overlap", + measure="per-judge per-dimension agreement", + propose="weight/routing updates", + judge="operator confirmation of routing policy changes", + artifact="judge-routing-policy:", + status="horizon", + implementation_ref="polylogue-rxdo.9.12", + ), + ImprovementLoopSpec( + loop_id="L4", + title="Orchestration prompt outcomes", + watch="stored lane-prompt artifacts + outcomes (PR merged, review iterations, cost, time)", + measure="prompt-feature x outcome correlations", + propose="findings + prompt template diffs", + judge="operator adoption of a template version", + artifact="prompt-template:", + status="horizon", + ), + ImprovementLoopSpec( + loop_id="L5", + title="Detector precision", + watch="pathology/finding detector candidates + their judgments", + measure="per-detector precision (standing query)", + propose="threshold/rule adjustments", + judge="operator/agent judgment lifecycle", + artifact="detector:", + status="horizon", + ), + ImprovementLoopSpec( + loop_id="L6", + title="Title/summary CTR", + watch="query-run telemetry + subsequent read events (implicit click-through)", + measure="per-title-source CTR, rank-at-click recorded to avoid position bias", + propose="title-generation strategy ranking", + judge="operator strategy flag flip", + artifact="title-strategy:", + status="horizon", + implementation_ref="polylogue-rxdo.3", + ), + ImprovementLoopSpec( + loop_id="L7", + title="Compaction regret", + watch="compaction boundaries + later agent re-derivations of discarded prefix content", + measure="regret = re-derived mass that was discarded (embedding match)", + propose="compaction policy tuning (what to preserve)", + judge="operator adoption of a policy version", + artifact="compaction-policy:", + status="horizon", + implementation_ref="polylogue-gjg.3", + ), + ImprovementLoopSpec( + loop_id="L8", + title="Cost routing", + watch="routing decisions + judged outcomes", + measure="tier efficiency frontier", + propose="routing advisor updates", + judge="operator adoption", + artifact="routing-advisor:", + status="horizon", + ), + ImprovementLoopSpec( + loop_id="L9", + title="Ontology drift", + watch="taxonomy/ontology usage drift signals", + measure="drift metric (per polylogue-dve1's design)", + propose="ontology revision candidates", + judge="operator adoption", + artifact="ontology:", + status="horizon", + implementation_ref="polylogue-dve1", + ), + ImprovementLoopSpec( + loop_id="L10", + title="Elicitation value (meta-loop)", + watch="every recorded judgment's downstream decision impact (retrospective re-derivation diff)", + measure="decision-impact per judgment type", + propose="asking-policy update (solicit highest-expected-impact next)", + judge="operator adoption of a policy version", + artifact="asking-policy:", + status="horizon", + ), + ImprovementLoopSpec( + loop_id="L11", + title="Declaration recall", + watch="retrospective PACK-D detection of undeclared corrections/claims vs declared markers", + measure="per-agent recall score", + propose="skill/preamble revisions", + judge="operator adoption (explicit, revocable policy assertion; no blocking enforcement)", + artifact="skill:", + status="horizon", + implementation_ref="polylogue-37t.2", + ), + ImprovementLoopSpec( + loop_id="L12", + title="Curriculum", + watch="query-run telemetry", + measure="recipe value", + propose="curriculum diff", + judge="operator gate", + artifact="skill:", + status="horizon", + implementation_ref="polylogue-xv1u", + ), + ImprovementLoopSpec( + loop_id="L13", + title="Capture-coverage error", + watch="sessions-known-to-exist vs archived, per origin", + measure="coverage gap volume", + propose="budgeted alerts / remediation candidates", + judge="operator/agent judgment lifecycle", + artifact="capture-coverage-policy:", + status="horizon", + implementation_ref="polylogue-3uw", + ), + ) +} + + +def active_loops() -> tuple[ImprovementLoopSpec, ...]: + """Loops the shared scheduler is actually executing today (empty until it exists).""" + return tuple(spec for spec in LOOP_REGISTRY.values() if spec.status == "active") + + +def horizon_loops() -> tuple[ImprovementLoopSpec, ...]: + """Loops declared but not yet wired to a real scheduler.""" + return tuple(spec for spec in LOOP_REGISTRY.values() if spec.status == "horizon") + + +__all__ = [ + "LOOP_REGISTRY", + "ImprovementLoopSpec", + "LoopStatus", + "active_loops", + "horizon_loops", +] diff --git a/polylogue/storage/sqlite/finding_provenance.py b/polylogue/storage/sqlite/finding_provenance.py new file mode 100644 index 0000000000..46e7868332 --- /dev/null +++ b/polylogue/storage/sqlite/finding_provenance.py @@ -0,0 +1,134 @@ +"""Queryable provenance projection over one ``AssertionKind.FINDING`` claim. + +Findings are ordinary assertion rows (``polylogue.finding.v1`` value payload, +see ``storage/sqlite/archive_tiers/user_write.py``): a prior review flagged +that finding provenance "must be QUERYABLE, not prose". This module answers +that -- given a finding's assertion id, it re-derives the finding's own +declared evidence refs (``query_ref``, ``result_set_ref``, ``baseline_ref``, +``current_ref``) plus every generic ``evidence_refs`` entry, resolves each +against live user-tier storage, and reports an honest current/stale/unknown +staleness verdict. It does not (yet) carry a code SHA or corpus-datasheet +hash -- those require build-info threading that is out of scope here and +tracked as a named follow-up (see the module docstring in +``polylogue.surfaces.payloads.FindingProvenancePayload``). +""" + +from __future__ import annotations + +import sqlite3 +from dataclasses import dataclass +from typing import Literal + +from polylogue.core.enums import AssertionKind +from polylogue.core.refs import ObjectRef +from polylogue.storage.sqlite.archive_tiers.user_write import ArchiveAssertionEnvelope, read_assertion_envelope +from polylogue.storage.sqlite.query_objects import get_query, get_result_set + +StalenessVerdict = Literal["current", "stale", "unknown"] + + +@dataclass(frozen=True, slots=True) +class FindingEvidenceResolution: + ref: str + resolvable: bool + reason: str | None = None + + +@dataclass(frozen=True, slots=True) +class FindingProvenance: + assertion_id: str + claim_key: str | None + target_ref: str + finding_kind: str | None + query_ref: str | None + result_set_ref: str | None + baseline_ref: str | None + current_ref: str | None + detector_ref: str | None + status: str + evidence: tuple[FindingEvidenceResolution, ...] + staleness_verdict: StalenessVerdict + created_at_ms: int + updated_at_ms: int + + +def compute_finding_provenance(conn: sqlite3.Connection, assertion_id: str) -> FindingProvenance | None: + """Return the provenance projection for one finding, or ``None`` if absent/not-a-finding.""" + + envelope = read_assertion_envelope(conn, assertion_id) + if envelope is None or envelope.kind != AssertionKind.FINDING.value: + return None + return _provenance_from_envelope(conn, envelope) + + +def _provenance_from_envelope(conn: sqlite3.Connection, envelope: ArchiveAssertionEnvelope) -> FindingProvenance: + value = envelope.value if isinstance(envelope.value, dict) else {} + query_ref = _str_or_none(value.get("query_ref")) + result_set_ref = _str_or_none(value.get("result_set_ref")) + baseline_ref = _str_or_none(value.get("baseline_ref")) + current_ref = _str_or_none(value.get("current_ref")) + finding_kind = _str_or_none(value.get("finding_kind")) + + declared_refs = [ref for ref in (query_ref, result_set_ref, baseline_ref, current_ref) if ref is not None] + all_refs = list(dict.fromkeys([*declared_refs, *envelope.evidence_refs])) + resolutions = tuple(_resolve_evidence_ref(conn, ref) for ref in all_refs) + resolved_by_ref = {resolution.ref: resolution.resolvable for resolution in resolutions} + + if not declared_refs: + staleness: StalenessVerdict = "unknown" + elif all(resolved_by_ref.get(ref, False) for ref in declared_refs): + staleness = "current" + elif any(ref in resolved_by_ref and not resolved_by_ref[ref] for ref in declared_refs): + staleness = "stale" + else: + staleness = "unknown" + + return FindingProvenance( + assertion_id=envelope.assertion_id, + claim_key=envelope.key, + target_ref=envelope.target_ref, + finding_kind=finding_kind, + query_ref=query_ref, + result_set_ref=result_set_ref, + baseline_ref=baseline_ref, + current_ref=current_ref, + detector_ref=envelope.author_ref, + status=envelope.status, + evidence=resolutions, + staleness_verdict=staleness, + created_at_ms=envelope.created_at_ms, + updated_at_ms=envelope.updated_at_ms, + ) + + +def _resolve_evidence_ref(conn: sqlite3.Connection, ref: str) -> FindingEvidenceResolution: + try: + parsed = ObjectRef.parse(ref) + except ValueError: + return FindingEvidenceResolution(ref=ref, resolvable=False, reason="unparseable ref") + if parsed.kind == "query": + found = get_query(conn, parsed.object_id) is not None + return FindingEvidenceResolution(ref=ref, resolvable=found, reason=None if found else "query not found") + if parsed.kind == "result-set": + found = get_result_set(conn, parsed.object_id) is not None + return FindingEvidenceResolution(ref=ref, resolvable=found, reason=None if found else "result set not found") + if parsed.kind == "assertion": + found = read_assertion_envelope(conn, parsed.object_id) is not None + return FindingEvidenceResolution(ref=ref, resolvable=found, reason=None if found else "assertion not found") + return FindingEvidenceResolution( + ref=ref, + resolvable=False, + reason=f"resolution not implemented for ref kind {parsed.kind!r}", + ) + + +def _str_or_none(value: object) -> str | None: + return value if isinstance(value, str) else None + + +__all__ = [ + "FindingEvidenceResolution", + "FindingProvenance", + "StalenessVerdict", + "compute_finding_provenance", +] diff --git a/polylogue/surfaces/payloads.py b/polylogue/surfaces/payloads.py index 94ce325910..80c8bdd588 100644 --- a/polylogue/surfaces/payloads.py +++ b/polylogue/surfaces/payloads.py @@ -1633,6 +1633,44 @@ def from_envelope(cls, envelope: ArchiveAssertionEnvelope) -> AssertionClaimPayl ) +class FindingEvidenceRefState(SurfacePayloadModel): + """Live resolution state for one evidence ref cited by a finding.""" + + ref: str + resolvable: bool + reason: str | None = None + + +class FindingProvenancePayload(SurfacePayloadModel): + """Queryable provenance projection over one ``AssertionKind.FINDING`` claim. + + Surfaces the evidence-ancestry fields the finding-provenance doctrine + calls for (finding id, claim key, target, query/result/baseline/current + refs, detector ref) as structured data instead of prose, plus a live + per-ref resolution check and an honest staleness verdict. This is + deliberately *not* the full W3C-PROV-style stanza requested alongside it + (no code SHA / corpus-datasheet hash -- those need build-info threading + tracked separately as follow-up); it is the bounded slice of "queryable, + not prose" provenance available from the finding's own durable evidence + refs today. + """ + + assertion_id: str + claim_key: str | None + target_ref: str + finding_kind: str | None + query_ref: str | None + result_set_ref: str | None + baseline_ref: str | None + current_ref: str | None + detector_ref: str | None + status: AssertionStatus + evidence: tuple[FindingEvidenceRefState, ...] + staleness_verdict: Literal["current", "stale", "unknown"] + created_at_ms: int + updated_at_ms: int + + class AssertionClaimListPayload(SurfacePayloadModel): """Shared list envelope for assertion-backed lifecycle claims.""" @@ -3473,6 +3511,8 @@ def validate_metadata_key(key: object) -> str | None: "FacetTimeRange", "FacetFamilyStatusPayload", "FacetsResponse", + "FindingEvidenceRefState", + "FindingProvenancePayload", "ArchiveDebtActionPayload", "ArchiveDebtKind", "ArchiveDebtListPayload", diff --git a/tests/unit/api/test_facade_contracts.py b/tests/unit/api/test_facade_contracts.py index 7b53d60175..4d797d7b3f 100644 --- a/tests/unit/api/test_facade_contracts.py +++ b/tests/unit/api/test_facade_contracts.py @@ -2522,7 +2522,6 @@ async def test_resolve_ref_returns_bounded_session_message_block_and_runtime_pay ("query", "sha256:deadbeef"), ("query-run", "sha256:deadbeef:run-1"), ("result-set", "sha256:deadbeef:run-1:result-1"), - ("finding", "finding-hash-1"), ("cohort", "cohort-1"), ("analysis", "analysis-1"), ], @@ -2532,12 +2531,15 @@ async def test_resolve_ref_returns_typed_pending_payload_for_analysis_provenance ) -> None: """polylogue-rxdo.1: refs land ahead of storage; resolution stubs cleanly. - ``query``/``query-run``/``result-set``/``finding``/``cohort``/``analysis`` + ``query``/``query-run``/``result-set``/``cohort``/``analysis`` are registered ObjectRefKind values with no backing table yet - (polylogue-rxdo.2/.3/.4/.8). resolve_ref must not raise and - must not silently pretend to resolve them — it returns a typed pending - payload carrying reason=substrate-pending so a client can distinguish - "not implemented yet" from "does not exist". + (polylogue-rxdo.2/.3/.8). ``finding`` graduated out of this pending set + in polylogue-rxdo.4 -- see + ``test_resolve_ref_returns_finding_provenance_payload`` below. + resolve_ref must not raise and must not silently pretend to resolve + these remaining kinds — it returns a typed pending payload carrying + reason=substrate-pending so a client can distinguish "not implemented + yet" from "does not exist". """ archive = _archive(tmp_path) try: @@ -2857,6 +2859,87 @@ async def test_resolve_ref_returns_assertion_payload(tmp_path: Path) -> None: await archive.close() +async def test_resolve_ref_returns_finding_provenance_payload(tmp_path: Path) -> None: + """polylogue-rxdo.4: ``finding:`` refs resolve to a queryable provenance projection. + + Evidence refs (the finding's own declared ``query_ref``/``result_set_ref`` + plus generic ``evidence_refs``) are re-resolved live, so the payload + reports an honest current/stale/unknown staleness verdict rather than + prose. + """ + from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database + from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier + from polylogue.storage.sqlite.archive_tiers.user_write import FindingAssertion, upsert_findings_as_assertions + from polylogue.storage.sqlite.query_objects import put_query, put_result_set + + archive = _archive(tmp_path) + try: + user_db = archive.config.archive_root / "user.db" + initialize_archive_database(user_db, ArchiveTier.USER) + with sqlite3.connect(user_db) as conn: + query = put_query( + conn, + {"field": "origin", "value": "codex-session"}, + grain="session", + lane="dialogue", + rank_policy="mixed", + created_at_ms=1, + ) + result_set = put_result_set( + conn, + result_set_id="finding-ref-resolution-rs", + query_hash=query.query_hash, + grain="session", + corpus_epoch="index:g1", + member_refs=("session:codex-session:ref-resolution-finding",), + exactness="exact", + persistence_class="finding", + created_at_ms=1, + ) + envelopes = upsert_findings_as_assertions( + conn, + [ + FindingAssertion( + claim_key="ref-resolution-claim", + target_ref=f"query:{query.query_hash}", + body_text="One session matched.", + finding_kind="measure", + statistic={"op": "count", "value": 1, "unit": "members"}, + n=1, + query_ref=f"query:{query.query_hash}", + result_set_ref=f"result-set:{result_set.result_set_id}", + detector_ref="agent:ref-resolution-detector", + ) + ], + now_ms=1, + ) + conn.commit() + assertion_id = envelopes[0].assertion_id + + payload = await archive.resolve_ref(f"finding:{assertion_id}") + + assert payload.resolved is True + assert payload.kind == "finding" + assert payload.payload_kind == "finding-provenance" + assert payload.payload is not None + assert payload.payload["assertion_id"] == assertion_id + assert payload.payload["finding_kind"] == "measure" + assert payload.payload["query_ref"] == f"query:{query.query_hash}" + assert payload.payload["result_set_ref"] == f"result-set:{result_set.result_set_id}" + assert payload.payload["staleness_verdict"] == "current" + evidence_states = {item["ref"]: item["resolvable"] for item in payload.payload["evidence"]} + assert evidence_states[f"query:{query.query_hash}"] is True + assert evidence_states[f"result-set:{result_set.result_set_id}"] is True + assert payload.caveats == () + + missing = await archive.resolve_ref("finding:does-not-exist") + assert missing.resolved is False + assert missing.kind == "finding" + assert missing.payload is None + finally: + await archive.close() + + def _delegation_parent_session(*, provider_session_id: str, with_dispatch: bool) -> ParsedSession: """Ingest-shaped parent fixture: writes real session/message/block rows through the live archive writer (``ArchiveStore.write_parsed`` -> diff --git a/tests/unit/archive/query/test_predicate_payload_roundtrip.py b/tests/unit/archive/query/test_predicate_payload_roundtrip.py new file mode 100644 index 0000000000..38385ea7ac --- /dev/null +++ b/tests/unit/archive/query/test_predicate_payload_roundtrip.py @@ -0,0 +1,100 @@ +"""Losslessness of ``predicate_from_payload`` against every predicate's own ``to_payload()``. + +``polylogue.archive.query.production_evaluator`` depends on this round trip +to reconstruct an executable predicate from a durable ``query:`` +definition without reverse-compiling arbitrary/lossy text. If any predicate +variant's payload shape drifts from its reconstructor (a field renamed on one +side but not the other), these tests fail immediately instead of surfacing as +a silent evaluator misbehavior later. +""" + +from __future__ import annotations + +import pytest + +from polylogue.archive.query.predicate import ( + QueryBoolPredicate, + QueryExistsPredicate, + QueryFieldPredicate, + QueryFieldRef, + QueryLineagePredicate, + QueryNotPredicate, + QueryPredicate, + QuerySemanticPredicate, + QuerySequenceConstraint, + QuerySequencePredicate, + QueryTextPredicate, + predicate_from_payload, +) + +_ROUNDTRIP_CASES: tuple[QueryPredicate, ...] = ( + QueryFieldPredicate(field="origin", values=("codex-session",), op="="), + QueryFieldPredicate(field="origin", values=("codex-session",), op="=").with_field_ref( + QueryFieldRef(scope="session", name="origin", source_name="origin") + ), + QueryFieldPredicate(field="count", values=("3",), op=">=").with_field_ref( + QueryFieldRef(scope="unit", name="count", source_name="count", unit="message") + ), + QueryNotPredicate(QueryFieldPredicate(field="origin", values=("codex-session",), op="=")), + QueryBoolPredicate( + "and", + ( + QueryFieldPredicate(field="origin", values=("codex-session",), op="="), + QueryFieldPredicate(field="repo", values=("polylogue",), op="="), + ), + ), + QueryBoolPredicate( + "or", + ( + QueryFieldPredicate(field="origin", values=("codex-session",), op="="), + QueryNotPredicate(QueryFieldPredicate(field="repo", values=("polylogue",), op="=")), + ), + ), + QueryExistsPredicate(unit="block", child=QueryFieldPredicate(field="tool_name", values=("Bash",), op="=")), + QuerySequencePredicate(action_terms=("plan", "edit", "test")), + QuerySequencePredicate( + steps=( + QueryFieldPredicate(field="action", values=("plan",), op="="), + QueryFieldPredicate(field="action", values=("edit",), op="="), + ), + constraints=(QuerySequenceConstraint(kind="within", within_ms=60_000),), + ), + QueryTextPredicate(text="deploy with caveats"), + QuerySemanticPredicate(text="deploy with caveats"), + QueryLineagePredicate(seed_session_id="codex-session:abc123"), +) + + +@pytest.mark.parametrize("predicate", _ROUNDTRIP_CASES, ids=lambda p: type(p).__name__) +def test_payload_roundtrip_is_lossless(predicate: QueryPredicate) -> None: + payload = predicate.to_payload() + reconstructed = predicate_from_payload(payload) + assert reconstructed == predicate + # The reconstruction must also re-serialize to the identical payload, not + # merely compare equal as a dataclass (catches asymmetric defaults). + assert reconstructed.to_payload() == payload + + +def test_unsupported_predicate_kind_fails_closed() -> None: + with pytest.raises(ValueError, match="unsupported predicate payload kind"): + predicate_from_payload({"kind": "made-up"}) + + +def test_field_predicate_rejects_unsupported_op() -> None: + with pytest.raises(ValueError, match="unsupported field predicate op"): + predicate_from_payload({"kind": "field", "field": "origin", "op": "!=", "values": ["x"]}) + + +def test_exists_predicate_rejects_unsupported_unit() -> None: + with pytest.raises(ValueError, match="unsupported exists unit"): + predicate_from_payload({"kind": "exists", "unit": "made-up", "child": {"kind": "fts", "text": "x"}}) + + +def test_boolean_predicate_requires_children_list() -> None: + with pytest.raises(ValueError, match="'children'"): + predicate_from_payload({"kind": "and", "children": "not-a-list"}) + + +def test_not_predicate_requires_object_child() -> None: + with pytest.raises(ValueError, match="'child'"): + predicate_from_payload({"kind": "not", "child": "not-an-object"}) diff --git a/tests/unit/archive/query/test_production_evaluator.py b/tests/unit/archive/query/test_production_evaluator.py new file mode 100644 index 0000000000..22517ef718 --- /dev/null +++ b/tests/unit/archive/query/test_production_evaluator.py @@ -0,0 +1,232 @@ +"""Real (non-fake) production evaluator: canonical plan -> live archive rows. + +These tests exercise the actual dependency the earlier substrate PRs (#2813, +#2826) left unwired: ``ArchiveCanonicalPlanEvaluator`` reconstructs a typed +predicate from a durable ``query:`` definition and runs it through the +same ``SessionFilter`` execution path every real surface uses -- no test +double stands in for the planner. Removing the predicate reconstruction (or +the ``SessionFilter`` call) makes ``test_evaluate_matches_real_archive_rows`` +fail, since the assertion depends on the archive actually filtering by +origin. Removing the ``record_query_run`` call makes +``test_evaluate_records_a_production_query_run`` fail, since it reads the +row back from ``ops.db``. +""" + +from __future__ import annotations + +import sqlite3 +from pathlib import Path + +import pytest + +from polylogue.archive.message.roles import Role +from polylogue.archive.query.evaluator import QueryEvaluationRequest +from polylogue.archive.query.production_evaluator import ( + ArchiveCanonicalPlanEvaluator, + LegacyQueryDefinitionNotExecutableError, + UnsupportedEvaluationGrainError, +) +from polylogue.core.enums import BlockType, Provider +from polylogue.core.query_identity import LEGACY_QUERY_DEFINITION_PROTOCOL_VERSION, JsonValue +from polylogue.sources.parsers.base import ParsedContentBlock, ParsedMessage, ParsedSession +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 + + +def _seed_archive(archive_root: Path) -> None: + archive_root.mkdir(parents=True, exist_ok=True) + with ArchiveStore(archive_root) as archive: + for provider, native_id, title in ( + (Provider.CODEX, "codex-1", "codex session"), + (Provider.CLAUDE_CODE, "claude-1", "claude session"), + ): + archive.write_parsed( + ParsedSession( + source_name=provider, + provider_session_id=native_id, + title=title, + created_at="2026-01-01T00:00:00+00:00", + updated_at="2026-01-01T00:01:00+00:00", + messages=[ + ParsedMessage( + provider_message_id=f"{native_id}-m1", + role=Role.USER, + text="hello", + timestamp="2026-01-01T00:00:00+00:00", + blocks=[ParsedContentBlock(type=BlockType.TEXT, text="hello")], + ) + ], + ) + ) + initialize_archive_database(archive_root / "user.db", ArchiveTier.USER) + initialize_archive_database(archive_root / "ops.db", ArchiveTier.OPS) + + +def _origin_query(conn: sqlite3.Connection, *, origin: str) -> QueryObject: + ast: dict[str, JsonValue] = { + "kind": "field", + "field": "origin", + "op": "=", + "values": [origin], + } + return put_query(conn, ast, grain="session", lane="dialogue", rank_policy="mixed", created_at_ms=1) + + +def test_evaluate_matches_real_archive_rows(tmp_path: Path) -> None: + 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") + conn.commit() + + evaluator = ArchiveCanonicalPlanEvaluator(archive_root / "index.db") + evaluation = evaluator.evaluate(QueryEvaluationRequest(query=query, purpose="reference")) + + assert evaluation.grain == "session" + assert evaluation.exactness == "exact" + assert len(evaluation.member_refs) == 1 + assert evaluation.member_refs[0].startswith("session:codex-session:") + assert evaluation.receipt.runtime_build_ref.startswith("polylogue:") + + +def test_evaluate_does_not_truncate_at_the_default_page_limit(tmp_path: Path) -> None: + """A watched query matching >50 sessions must enumerate every member. + + ``SessionFilter.list_summaries`` caps at a default page limit of 50 rows. + If the evaluator used that page-capped read path instead of + ``list_all_summaries``, this test's 60th+ matching session would be + silently dropped from ``member_refs`` while ``exactness`` still claimed + ``"exact"`` -- corrupting the merkle-root-based standing-query drift + detection that trusts ``member_refs`` as a complete enumeration. + """ + archive_root = tmp_path / "archive" + archive_root.mkdir(parents=True, exist_ok=True) + session_count = 60 + with ArchiveStore(archive_root) as archive: + for i in range(session_count): + native_id = f"codex-{i:03d}" + archive.write_parsed( + ParsedSession( + source_name=Provider.CODEX, + provider_session_id=native_id, + title=f"codex session {i}", + created_at="2026-01-01T00:00:00+00:00", + updated_at="2026-01-01T00:01:00+00:00", + messages=[ + ParsedMessage( + provider_message_id=f"{native_id}-m1", + role=Role.USER, + text="hello", + timestamp="2026-01-01T00:00:00+00:00", + blocks=[ParsedContentBlock(type=BlockType.TEXT, text="hello")], + ) + ], + ) + ) + initialize_archive_database(archive_root / "user.db", ArchiveTier.USER) + initialize_archive_database(archive_root / "ops.db", ArchiveTier.OPS) + + with sqlite3.connect(archive_root / "user.db") as conn: + query = _origin_query(conn, origin="codex-session") + conn.commit() + + evaluator = ArchiveCanonicalPlanEvaluator(archive_root / "index.db") + evaluation = evaluator.evaluate(QueryEvaluationRequest(query=query, purpose="standing-watch")) + + assert evaluation.exactness == "exact" + assert len(evaluation.member_refs) == session_count + + with sqlite3.connect(archive_root / "ops.db") as conn: + (member_count,) = conn.execute("SELECT member_count FROM query_runs").fetchone() + assert member_count == session_count + + +def test_evaluate_excludes_origin_prefix(tmp_path: Path) -> None: + """The self-trigger firewall drops members whose origin matches an excluded prefix.""" + 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") + conn.commit() + + evaluator = ArchiveCanonicalPlanEvaluator(archive_root / "index.db") + evaluation = evaluator.evaluate( + QueryEvaluationRequest(query=query, purpose="standing-watch", excluded_origin_prefixes=("codex-",)) + ) + + assert evaluation.member_refs == () + + +def test_evaluate_excludes_scope_refs(tmp_path: Path) -> None: + """The self-trigger firewall also drops explicitly excluded member refs.""" + 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") + conn.commit() + + evaluator = ArchiveCanonicalPlanEvaluator(archive_root / "index.db") + baseline = evaluator.evaluate(QueryEvaluationRequest(query=query, purpose="reference")) + assert len(baseline.member_refs) == 1 + + evaluation = evaluator.evaluate( + QueryEvaluationRequest(query=query, purpose="standing-watch", excluded_scope_refs=(baseline.member_refs[0],)) + ) + assert evaluation.member_refs == () + + +def test_evaluate_records_a_production_query_run(tmp_path: Path) -> None: + archive_root = tmp_path / "archive" + _seed_archive(archive_root) + with sqlite3.connect(archive_root / "user.db") as conn: + query = _origin_query(conn, origin="claude-code-session") + conn.commit() + + evaluator = ArchiveCanonicalPlanEvaluator(archive_root / "index.db") + evaluator.evaluate(QueryEvaluationRequest(query=query, purpose="reference")) + + with sqlite3.connect(archive_root / "ops.db") as conn: + rows = conn.execute("SELECT query_hash, surface, member_count, exactness FROM query_runs").fetchall() + assert len(rows) == 1 + assert rows[0][0] == query.query_hash + assert rows[0][1] == "daemon-internal" + assert rows[0][2] == 1 + assert rows[0][3] == "exact" + + +def test_evaluate_rejects_legacy_protocol_v0() -> None: + legacy = QueryObject( + query_hash="b" * 64, + canonical_plan={"field": "origin", "value": "codex-session"}, + grain="session", + lane="dialogue", + rank_policy="mixed", + definition_protocol_version=LEGACY_QUERY_DEFINITION_PROTOCOL_VERSION, + ) + evaluator = ArchiveCanonicalPlanEvaluator(Path("/nonexistent/index.db")) + with pytest.raises(LegacyQueryDefinitionNotExecutableError): + evaluator.evaluate(QueryEvaluationRequest(query=legacy, purpose="reference")) + + +def test_evaluate_rejects_unsupported_grain(tmp_path: Path) -> None: + archive_root = tmp_path / "archive" + _seed_archive(archive_root) + with sqlite3.connect(archive_root / "user.db") as conn: + ast: dict[str, JsonValue] = {"kind": "field", "field": "tool_name", "op": "=", "values": ["Bash"]} + query = put_query(conn, ast, grain="action", lane="dialogue", rank_policy="mixed", created_at_ms=1) + conn.commit() + + evaluator = ArchiveCanonicalPlanEvaluator(archive_root / "index.db") + with pytest.raises(UnsupportedEvaluationGrainError): + evaluator.evaluate(QueryEvaluationRequest(query=query, purpose="reference")) + + +def test_resolve_cohort_is_not_yet_implemented(tmp_path: Path) -> None: + from polylogue.archive.query.expression import RefOperand + from polylogue.core.refs import ObjectRef + + evaluator = ArchiveCanonicalPlanEvaluator(tmp_path / "index.db") + with pytest.raises(NotImplementedError): + evaluator.resolve_cohort(RefOperand(ObjectRef(kind="cohort", object_id="team"))) diff --git a/tests/unit/daemon/test_standing_queries_default_evaluator.py b/tests/unit/daemon/test_standing_queries_default_evaluator.py new file mode 100644 index 0000000000..456f58d1f2 --- /dev/null +++ b/tests/unit/daemon/test_standing_queries_default_evaluator.py @@ -0,0 +1,91 @@ +"""End-to-end proof of the default daemon evaluator injection (polylogue-rxdo.5). + +Every earlier standing-query test injects a hand-rolled fake evaluator. +``make_default_convergence_stages`` previously called +``make_standing_query_stage(db_path)`` with no evaluator at all, which left +the stage permanently inert in production (``check_sessions``/ +``execute_sessions`` both short-circuit to a no-op when ``evaluator is +None``). This test proves the *production* wiring: no fake is injected here, +only the real ``ArchiveCanonicalPlanEvaluator`` reached through +``make_default_convergence_stages``, running against a real ingested +archive. +""" + +from __future__ import annotations + +import sqlite3 +from pathlib import Path + +from polylogue.archive.message.roles import Role +from polylogue.core.enums import AssertionKind, BlockType, Provider +from polylogue.daemon.convergence import DaemonConverger +from polylogue.daemon.convergence_stages import make_default_convergence_stages +from polylogue.sources.parsers.base import ParsedContentBlock, ParsedMessage, ParsedSession +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 put_query, put_query_name + + +def _seed_archive_with_one_codex_session(archive_root: Path) -> str: + archive_root.mkdir(parents=True, exist_ok=True) + with ArchiveStore(archive_root) as archive: + session_id = archive.write_parsed( + ParsedSession( + source_name=Provider.CODEX, + provider_session_id="codex-1", + title="codex session", + created_at="2026-01-01T00:00:00+00:00", + updated_at="2026-01-01T00:01:00+00:00", + messages=[ + ParsedMessage( + provider_message_id="codex-1-m1", + role=Role.USER, + text="hello", + timestamp="2026-01-01T00:00:00+00:00", + blocks=[ParsedContentBlock(type=BlockType.TEXT, text="hello")], + ) + ], + ) + ) + initialize_archive_database(archive_root / "user.db", ArchiveTier.USER) + initialize_archive_database(archive_root / "ops.db", ArchiveTier.OPS) + return session_id + + +def test_default_stage_set_evaluates_a_watched_query_without_an_injected_fake(tmp_path: Path) -> None: + archive_root = tmp_path / "archive" + session_id = _seed_archive_with_one_codex_session(archive_root) + + with sqlite3.connect(archive_root / "user.db") as conn: + query = put_query( + conn, + {"kind": "field", "field": "origin", "op": "=", "values": ["codex-session"]}, + grain="session", + lane="dialogue", + rank_policy="mixed", + created_at_ms=1, + ) + put_query_name(conn, name="codex-watch", query_hash=query.query_hash, watch=True, updated_at_ms=2) + conn.commit() + + stages = make_default_convergence_stages(archive_root / "index.db") + standing_stage = next(stage for stage in stages if stage.name == "standing-queries") + converger = DaemonConverger(stages=(standing_stage,), max_workers=1) + states, _timings = converger.converge_sessions((session_id,)) + assert states[session_id].stages["standing-queries"].value == "done" + + with sqlite3.connect(archive_root / "user.db") as conn: + baseline_row = conn.execute("SELECT member_count FROM result_sets WHERE persistence_class = 'watch'").fetchone() + assert baseline_row is not None + assert baseline_row[0] == 1 + # First observation only establishes a baseline; there is no prior + # membership to diff against yet, so no candidate finding fires. + finding_count = conn.execute( + "SELECT COUNT(*) FROM assertions WHERE kind = ?", (AssertionKind.FINDING.value,) + ).fetchone()[0] + assert finding_count == 0 + + with sqlite3.connect(archive_root / "ops.db") as conn: + run_count = conn.execute("SELECT COUNT(*) FROM query_runs WHERE surface = 'daemon-internal'").fetchone()[0] + assert run_count == 1 diff --git a/tests/unit/insights/test_improvement_loops.py b/tests/unit/insights/test_improvement_loops.py new file mode 100644 index 0000000000..21e29dea66 --- /dev/null +++ b/tests/unit/insights/test_improvement_loops.py @@ -0,0 +1,38 @@ +"""Structural invariants for the improvement-loop declaration registry (polylogue-rxdo.11).""" + +from __future__ import annotations + +from polylogue.insights.improvement_loops import LOOP_REGISTRY, active_loops, horizon_loops + + +def test_registry_keys_match_loop_ids() -> None: + for key, spec in LOOP_REGISTRY.items(): + assert key == spec.loop_id + + +def test_every_spec_declares_the_full_five_tuple() -> None: + for spec in LOOP_REGISTRY.values(): + assert spec.watch.strip() + assert spec.measure.strip() + assert spec.propose.strip() + assert spec.judge.strip() + assert spec.artifact.strip() + assert ":" in spec.artifact, f"{spec.loop_id} artifact must be content-addressed" + + +def test_no_loop_is_active_until_a_shared_scheduler_exists() -> None: + """Corrective AC: no loop may claim an active daemon schedule at this closure. + + If this test starts failing because someone flipped a spec to + ``status="active"``, the shared scheduler/state module (and that loop's + watch/measure/propose/judge/bump wiring) must exist and be tested first + -- update this test deliberately, not by loosening the assertion. + """ + assert active_loops() == () + assert len(horizon_loops()) == len(LOOP_REGISTRY) + + +def test_l1_and_l2_are_declared_as_the_first_pilot_pair() -> None: + assert "L1" in LOOP_REGISTRY + assert "L2" in LOOP_REGISTRY + assert LOOP_REGISTRY["L1"].implementation_ref == "polylogue-37t.17"