diff --git a/polylogue/api/archive.py b/polylogue/api/archive.py index 56a9e660c0..bd871d5dae 100644 --- a/polylogue/api/archive.py +++ b/polylogue/api/archive.py @@ -182,6 +182,7 @@ "claim": AssertionKind.DECISION, "correction": AssertionKind.CORRECTION, "lesson": AssertionKind.LESSON, + "caveat": AssertionKind.CAVEAT, "highlight": AssertionKind.HIGHLIGHT, "prompt_eval": AssertionKind.PROMPT_EVAL, } diff --git a/polylogue/cli/commands/embed.py b/polylogue/cli/commands/embed.py index df19a9bc08..71faf80bc4 100644 --- a/polylogue/cli/commands/embed.py +++ b/polylogue/cli/commands/embed.py @@ -29,6 +29,7 @@ from polylogue.cli.shared.embed_stats import show_embedding_stats from polylogue.cli.shared.types import AppEnv +from polylogue.core.enums import OperationStatus if TYPE_CHECKING: from polylogue.storage.archive_identity import ArchiveLocation @@ -179,6 +180,12 @@ class BackfillResultPayload(TypedDict): sessions: list[BackfillSessionPayload] +_ARCHIVE_BACKFILL_STATUS_MAP: dict[str, OperationStatus] = { + "complete": OperationStatus.COMPLETED, + "stopped": OperationStatus.INTERRUPTED, +} + + def _render_backfill_json(payload: BackfillResultPayload) -> None: click.echo(json.dumps(payload, indent=2, sort_keys=True)) @@ -875,11 +882,11 @@ def _record_archive_backfill_run( ops_db = (configured_root / "ops.db") if configured_root is not None else index_db.with_name("ops.db") initialize_archive_database(ops_db, ArchiveTier.OPS) - # Vocabulary boundary: the CLI payload says 'stopped'/'complete'; the - # ops-tier CHECK (archive_tiers/ops.py) admits - # 'running'/'completed'/'failed'/'cancelled'. Translate here, at the sole - # write site, so the two vocabularies never meet anywhere else. - terminal_status = "cancelled" if status == "stopped" else "completed" + try: + terminal_status = _ARCHIVE_BACKFILL_STATUS_MAP[status] + except KeyError as exc: + choices = ", ".join(_ARCHIVE_BACKFILL_STATUS_MAP) + raise ValueError(f"unknown archive backfill status {status!r}; expected one of: {choices}") from exc with closing(sqlite3.connect(ops_db, timeout=30.0)) as conn: upsert_embedding_catchup_run( conn, diff --git a/polylogue/cli/commands/note.py b/polylogue/cli/commands/note.py index dc5ad8029e..2f63531f64 100644 --- a/polylogue/cli/commands/note.py +++ b/polylogue/cli/commands/note.py @@ -16,7 +16,7 @@ MAX_NOTE_STDIN_BYTES = 256 * 1024 -_KIND_NAMES = ("note", "claim", "correction", "lesson", "highlight", "prompt_eval") +_KIND_NAMES = ("note", "claim", "correction", "lesson", "caveat", "highlight", "prompt_eval") def _scope_refs(repo: str | None, topic: str | None) -> tuple[str, ...]: diff --git a/polylogue/core/enums.py b/polylogue/core/enums.py index bce56ba6d0..4b1d9cad7f 100644 --- a/polylogue/core/enums.py +++ b/polylogue/core/enums.py @@ -14,6 +14,26 @@ def __str__(self) -> str: return self.value +class OperationStatus(PolylogueStrEnum): + """Scheduling and lifecycle states shared by operation surfaces.""" + + ACCEPTED = "accepted" + REJECTED = "rejected" + PENDING = "pending" + RUNNING = "running" + COMPLETED = "completed" + FAILED = "failed" + INTERRUPTED = "interrupted" + + +OPERATION_LIFECYCLE_STATUSES: tuple[OperationStatus, ...] = ( + OperationStatus.RUNNING, + OperationStatus.COMPLETED, + OperationStatus.FAILED, + OperationStatus.INTERRUPTED, +) + + def enum_values(enum_type: type[PolylogueStrEnum]) -> tuple[str, ...]: """Return persisted values for a closed enum.""" return tuple(item.value for item in enum_type) @@ -740,6 +760,8 @@ def from_string(cls, value: str | RawAuthorityVerdict) -> RawAuthorityVerdict: "MaterialOrigin", "MessageType", "Origin", + "OperationStatus", + "OPERATION_LIFECYCLE_STATUSES", "PasteBoundary", "PlanStage", "PolylogueStrEnum", diff --git a/polylogue/maintenance/planner.py b/polylogue/maintenance/planner.py index 97318babe5..a3f2b7def4 100644 --- a/polylogue/maintenance/planner.py +++ b/polylogue/maintenance/planner.py @@ -37,6 +37,7 @@ from typing import TYPE_CHECKING from polylogue.config import Config +from polylogue.core.enums import OperationStatus if TYPE_CHECKING: from polylogue.storage.repair import ArchiveDebtStatus @@ -125,13 +126,9 @@ def _coerce_backfill_kind(value: object) -> BackfillKind: return _RETIRED_STORED_KIND_MAP.get(text, BackfillKind.DERIVED_REBUILD) -class BackfillStatus(str, Enum): - """Lifecycle status of a backfill operation.""" - - PENDING = "pending" - RUNNING = "running" - COMPLETED = "completed" - FAILED = "failed" +# Compatibility name for callers that imported the former planner-local enum. +# The operation model itself uses the canonical enum directly below. +BackfillStatus = OperationStatus @dataclass(frozen=True) @@ -240,7 +237,7 @@ class BackfillOperation: operation_id: str kind: BackfillKind targets: tuple[str, ...] - status: BackfillStatus = BackfillStatus.PENDING + status: OperationStatus = OperationStatus.PENDING progress: float = 0.0 started_at: str | None = None completed_at: str | None = None @@ -302,11 +299,11 @@ def from_dict(cls, payload: dict[str, object]) -> BackfillOperation: op_id = str(payload.get("operation_id", "")) kind = _coerce_backfill_kind(payload.get("kind", BackfillKind.DERIVED_REBUILD.value)) - status_raw = str(payload.get("status", BackfillStatus.PENDING.value)) + status_raw = str(payload.get("status", OperationStatus.PENDING.value)) try: - status = BackfillStatus(status_raw) + status = OperationStatus(status_raw) except ValueError: - status = BackfillStatus.PENDING + status = OperationStatus.PENDING targets_raw = payload.get("targets") or () if not isinstance(targets_raw, (list, tuple)): targets_raw = () diff --git a/polylogue/operations/operation_status.py b/polylogue/operations/operation_status.py index 1025adb630..c72e4292de 100644 --- a/polylogue/operations/operation_status.py +++ b/polylogue/operations/operation_status.py @@ -1,47 +1,14 @@ -"""``OperationStatus``: the scheduling status enum shared by every Operation ack. +"""Compatibility import for the canonical ``OperationStatus`` enum. Split out of :mod:`polylogue.operations.operation_contract` (polylogue-8s70): -that module's other export, ``OperationFollowUp``, subclasses -``polylogue.surfaces.payloads.SurfacePayloadModel`` (a pydantic base), which -pulls in a wide swath of the archive/semantic pricing and payload machinery -(observed ~280ms of import cost). ``OperationStatus`` is a plain ``str, Enum`` -with zero runtime dependency on any of that -- but importing it from the same -module used to force the whole chain onto callers who only want the status -enum, notably :mod:`polylogue.readiness.capability`, which is on the hot path -of every ``polylogue status``/``polylogue agents status`` invocation. +The canonical enum lives in :mod:`polylogue.core.enums` so storage DDL and +operation surfaces share one vocabulary without a dependency cycle. This +module remains the stable import path for callers that only need the status +type. """ from __future__ import annotations -from enum import Enum - - -class OperationStatus(str, Enum): - """Scheduling status carried by every Operation ack. - - Terminal vs in-flight semantics are intentionally simple: - - * ``ACCEPTED`` — the operation passed validation and was admitted into - the registry. A follow-up hint is populated so clients can poll for - completion. - * ``REJECTED`` — the request failed validation, lacked permissions, - collided with an existing idempotency key, or otherwise could not be - admitted. ``error`` is populated; ``follow_up`` is omitted. - * ``PENDING`` — admitted but not yet started. Distinguishes the brief - window where an ack has been issued before the worker picks up the - work; useful for queued operations. - * ``RUNNING`` / ``COMPLETED`` / ``FAILED`` are re-emitted by status - surfaces after the work has begun or finished (see - :mod:`polylogue.readiness.capability`). The initial scheduling - response is always one of ``ACCEPTED``, ``REJECTED``, or ``PENDING``. - """ - - ACCEPTED = "accepted" - REJECTED = "rejected" - PENDING = "pending" - RUNNING = "running" - COMPLETED = "completed" - FAILED = "failed" - +from polylogue.core.enums import OperationStatus __all__ = ["OperationStatus"] diff --git a/polylogue/readiness/capability.py b/polylogue/readiness/capability.py index 2eb8c62fed..f463532136 100644 --- a/polylogue/readiness/capability.py +++ b/polylogue/readiness/capability.py @@ -798,6 +798,7 @@ def component_from_operation_status( OperationStatus.COMPLETED: CapabilityReadinessState.READY, OperationStatus.REJECTED: CapabilityReadinessState.BLOCKED, OperationStatus.FAILED: CapabilityReadinessState.BLOCKED, + OperationStatus.INTERRUPTED: CapabilityReadinessState.BLOCKED, }[operation_status] return ComponentReadiness(component=component, scope=scope, state=state, summary=summary) diff --git a/polylogue/storage/sqlite/archive_tiers/bootstrap.py b/polylogue/storage/sqlite/archive_tiers/bootstrap.py index d7bc2c7f97..5cb6fb94e1 100644 --- a/polylogue/storage/sqlite/archive_tiers/bootstrap.py +++ b/polylogue/storage/sqlite/archive_tiers/bootstrap.py @@ -85,12 +85,14 @@ def initialize_archive_tier(conn: sqlite3.Connection, tier: ArchiveTier) -> None if tier is ArchiveTier.OPS: from polylogue.storage.sqlite.archive_tiers.ops_write import ( ensure_embedding_catchup_run_outcome_columns, + ensure_ops_status_checks, ) _ensure_ops_runtime_columns(conn) _ensure_ops_cursor_lag_sample_columns(conn) _ensure_ops_ingest_attempt_outcome_columns(conn) ensure_embedding_catchup_run_outcome_columns(conn) + ensure_ops_status_checks(conn) _ensure_schema_drift_samples_check(conn) if tier is ArchiveTier.INDEX: from polylogue.storage.sqlite.archive_tiers.pricing_seed import seed_price_catalog diff --git a/polylogue/storage/sqlite/archive_tiers/ops.py b/polylogue/storage/sqlite/archive_tiers/ops.py index be56587036..9259ba2be8 100644 --- a/polylogue/storage/sqlite/archive_tiers/ops.py +++ b/polylogue/storage/sqlite/archive_tiers/ops.py @@ -4,11 +4,12 @@ from typing import get_args -from polylogue.core.enums import IngestOutcome, Origin +from polylogue.core.enums import OPERATION_LIFECYCLE_STATUSES, IngestOutcome, Origin from polylogue.schemas.drift_sentinel import DriftClassification from polylogue.storage.sqlite.archive_tiers.common import check, literal_check, nullable_check OPS_SCHEMA_VERSION = 1 +_OPS_RUN_STATUS_CHECK = literal_check("status", *(status.value for status in OPERATION_LIFECYCLE_STATUSES)) # Split out of OPS_DDL (polylogue-sd9s) so the ops-bootstrap convergence step # that repairs a stale live CHECK (``_ensure_schema_drift_samples_check`` in @@ -34,6 +35,51 @@ ON schema_drift_samples(observed_at_ms DESC); """ +_OPS_INGEST_ATTEMPTS_DDL = f""" +CREATE TABLE IF NOT EXISTS ingest_attempts ( + attempt_id TEXT PRIMARY KEY, + source_path TEXT, + origin TEXT CHECK ({check("origin", Origin)} OR origin IS NULL), + status TEXT NOT NULL CHECK({_OPS_RUN_STATUS_CHECK}), + phase TEXT, + storage_route TEXT, + started_at_ms INTEGER NOT NULL, + heartbeat_at_ms INTEGER, + finished_at_ms INTEGER, + parsed_raw_count INTEGER NOT NULL DEFAULT 0 CHECK(parsed_raw_count >= 0), + materialized_count INTEGER NOT NULL DEFAULT 0 CHECK(materialized_count >= 0), + error_message TEXT, + source_paths_json TEXT NOT NULL DEFAULT '[]', + -- polylogue-cnu3: typed, structurally-classified disposition -- never + -- guessed from ``error_message`` text. ``outcome_code`` defaults to + -- ``legacy_unknown`` so every pre-existing row (written before this + -- vocabulary existed) stays honestly queryable as unclassified rather + -- than silently guessed into a real class (AC4). + outcome_code TEXT NOT NULL DEFAULT 'legacy_unknown' CHECK ({check("outcome_code", IngestOutcome)}), + retryable INTEGER CHECK(retryable IN (0, 1)), + evidence_ref TEXT, + diagnostic TEXT, + remediation TEXT +) STRICT; +""" + +_OPS_EMBEDDING_CATCHUP_RUNS_DDL = f""" +CREATE TABLE IF NOT EXISTS embedding_catchup_runs ( + run_id TEXT PRIMARY KEY, + started_at_ms INTEGER NOT NULL, + finished_at_ms INTEGER, + status TEXT NOT NULL CHECK({_OPS_RUN_STATUS_CHECK}), + origin TEXT CHECK ({nullable_check("origin", Origin)}), + scanned_sessions INTEGER NOT NULL DEFAULT 0 CHECK(scanned_sessions >= 0), + embedded_sessions INTEGER NOT NULL DEFAULT 0 CHECK(embedded_sessions >= 0), + skipped_sessions INTEGER NOT NULL DEFAULT 0 CHECK(skipped_sessions >= 0), + error_count INTEGER NOT NULL DEFAULT 0 CHECK(error_count >= 0), + embedded_messages INTEGER NOT NULL DEFAULT 0 CHECK(embedded_messages >= 0), + estimated_cost_usd REAL, + error_message TEXT +) STRICT; +""" + OPS_DDL = f""" CREATE TABLE IF NOT EXISTS ingest_cursor ( source_path TEXT PRIMARY KEY, @@ -68,31 +114,7 @@ CREATE INDEX IF NOT EXISTS idx_ingest_cursor_attention ON ingest_cursor(failure_count, excluded, source_path); -CREATE TABLE IF NOT EXISTS ingest_attempts ( - attempt_id TEXT PRIMARY KEY, - source_path TEXT, - origin TEXT CHECK ({check("origin", Origin)} OR origin IS NULL), - status TEXT NOT NULL CHECK(status IN ('running', 'completed', 'failed', 'interrupted')), - phase TEXT, - storage_route TEXT, - started_at_ms INTEGER NOT NULL, - heartbeat_at_ms INTEGER, - finished_at_ms INTEGER, - parsed_raw_count INTEGER NOT NULL DEFAULT 0 CHECK(parsed_raw_count >= 0), - materialized_count INTEGER NOT NULL DEFAULT 0 CHECK(materialized_count >= 0), - error_message TEXT, - source_paths_json TEXT NOT NULL DEFAULT '[]', - -- polylogue-cnu3: typed, structurally-classified disposition -- never - -- guessed from ``error_message`` text. ``outcome_code`` defaults to - -- ``legacy_unknown`` so every pre-existing row (written before this - -- vocabulary existed) stays honestly queryable as unclassified rather - -- than silently guessed into a real class (AC4). - outcome_code TEXT NOT NULL DEFAULT 'legacy_unknown' CHECK ({check("outcome_code", IngestOutcome)}), - retryable INTEGER CHECK(retryable IN (0, 1)), - evidence_ref TEXT, - diagnostic TEXT, - remediation TEXT -) STRICT; +{_OPS_INGEST_ATTEMPTS_DDL} CREATE INDEX IF NOT EXISTS idx_ingest_attempts_status ON ingest_attempts(status, heartbeat_at_ms); @@ -182,25 +204,13 @@ ON daemon_lifecycle(started_at_ms DESC); -- Sole live embedding_catchup_runs table (writer: ops_write.upsert_embedding_catchup_run). --- Status vocabulary: the CLI backfill payload says 'stopped'/'complete'; --- cli/commands/embed.py:_record_archive_backfill_run translates those to --- 'cancelled'/'completed' at this write boundary. A pre-split monolith table --- of the same name (different shape, statuses incl. 'stopped'/'interrupted') --- survives read-only via storage/embeddings/progress.py. -CREATE TABLE IF NOT EXISTS embedding_catchup_runs ( - run_id TEXT PRIMARY KEY, - started_at_ms INTEGER NOT NULL, - finished_at_ms INTEGER, - status TEXT NOT NULL CHECK(status IN ('running', 'completed', 'failed', 'cancelled')), - origin TEXT CHECK ({nullable_check("origin", Origin)}), - scanned_sessions INTEGER NOT NULL DEFAULT 0 CHECK(scanned_sessions >= 0), - embedded_sessions INTEGER NOT NULL DEFAULT 0 CHECK(embedded_sessions >= 0), - skipped_sessions INTEGER NOT NULL DEFAULT 0 CHECK(skipped_sessions >= 0), - error_count INTEGER NOT NULL DEFAULT 0 CHECK(error_count >= 0), - embedded_messages INTEGER NOT NULL DEFAULT 0 CHECK(embedded_messages >= 0), - estimated_cost_usd REAL, - error_message TEXT -) STRICT; +-- Status vocabulary is generated from the canonical operation lifecycle enum. +-- The public CLI payload retains its 'stopped'/'complete' display vocabulary; +-- cli/commands/embed.py maps those values to typed operation statuses at this +-- write boundary. A pre-split monolith table of the same name (different +-- shape, statuses incl. 'stopped'/'interrupted') survives read-only via +-- storage/embeddings/progress.py. +{_OPS_EMBEDDING_CATCHUP_RUNS_DDL} -- Bulk/archive-wide secret-candidate scan coverage (polylogue-layg.1). -- Sole live table for the bounded, resumable ``scan_archive_for_secret_candidates`` diff --git a/polylogue/storage/sqlite/archive_tiers/ops_write.py b/polylogue/storage/sqlite/archive_tiers/ops_write.py index c11e50ca87..4d833ac0bb 100644 --- a/polylogue/storage/sqlite/archive_tiers/ops_write.py +++ b/polylogue/storage/sqlite/archive_tiers/ops_write.py @@ -10,7 +10,7 @@ import uuid from dataclasses import dataclass -from polylogue.core.enums import IngestOutcome, Origin +from polylogue.core.enums import IngestOutcome, OperationStatus, Origin from polylogue.pipeline.ingest_outcomes import IngestAttemptDisposition MCP_CALL_LOG_RETENTION_MS = 90 * 24 * 60 * 60 * 1000 @@ -610,7 +610,7 @@ def upsert_ingest_cursor( def record_ingest_attempt( conn: sqlite3.Connection, *, - status: str, + status: OperationStatus | str, source_path: str | None = None, origin: Origin | str | None = None, phase: str | None = None, @@ -634,6 +634,7 @@ def record_ingest_attempt( """ if attempt_id is None: attempt_id = str(uuid.uuid4()) + status_value = status.value if isinstance(status, OperationStatus) else status has_storage_route = _table_has_column(conn, "ingest_attempts", "storage_route") route_column = "storage_route,\n " if has_storage_route else "" route_value = "?, " if has_storage_route else "" @@ -706,7 +707,7 @@ def record_ingest_attempt( attempt_id, source_path, _origin_value(origin), - status, + status_value, phase, *route_params, *outcome_params, @@ -1061,7 +1062,7 @@ def upsert_embedding_catchup_run( run_id: str | None = None, started_at_ms: int, finished_at_ms: int | None = None, - status: str, + status: OperationStatus | str, origin: Origin | str | None = None, scanned_sessions: int = 0, embedded_sessions: int = 0, @@ -1074,6 +1075,7 @@ def upsert_embedding_catchup_run( """Create or replace one ``embedding_catchup_runs`` row and return ``run_id``.""" if run_id is None: run_id = str(uuid.uuid4()) + status_value = status.value if isinstance(status, OperationStatus) else status ensure_embedding_catchup_run_outcome_columns(conn) conn.execute( """ @@ -1109,7 +1111,7 @@ def upsert_embedding_catchup_run( run_id, started_at_ms, finished_at_ms, - status, + status_value, _origin_value(origin), scanned_sessions, embedded_sessions, @@ -1185,6 +1187,85 @@ def ensure_embedding_catchup_run_outcome_columns(conn: sqlite3.Connection) -> No conn.execute(f"ALTER TABLE embedding_catchup_runs ADD COLUMN {name} {definition}") +def ensure_ops_status_checks(conn: sqlite3.Connection) -> None: + """Converge same-version OPS status CHECKs without discarding rows. + + The disposable tier intentionally has no migration chain, but SQLite does + not rewrite a table constraint when ``CREATE TABLE IF NOT EXISTS`` runs. + Rebuild only a table whose live status CHECK predates the canonical + lifecycle vocabulary. The replacement happens in the caller's transaction + and copies every row; the old embedding ``cancelled`` spelling is the + established equivalent of canonical ``interrupted`` and is normalized + while the row is copied. + """ + from polylogue.storage.sqlite.archive_tiers.ops import ( + _OPS_EMBEDDING_CATCHUP_RUNS_DDL, + _OPS_INGEST_ATTEMPTS_DDL, + _OPS_RUN_STATUS_CHECK, + ) + + _rebuild_ops_status_table_if_stale( + conn, + table="ingest_attempts", + table_ddl=_OPS_INGEST_ATTEMPTS_DDL, + status_check=_OPS_RUN_STATUS_CHECK, + indexes=( + "CREATE INDEX IF NOT EXISTS idx_ingest_attempts_status ON ingest_attempts(status, heartbeat_at_ms)", + "CREATE INDEX IF NOT EXISTS idx_ingest_attempts_storage_route ON ingest_attempts(storage_route)", + "CREATE INDEX IF NOT EXISTS idx_ingest_attempts_outcome_code " + "ON ingest_attempts(outcome_code, started_at_ms)", + ), + ) + _rebuild_ops_status_table_if_stale( + conn, + table="embedding_catchup_runs", + table_ddl=_OPS_EMBEDDING_CATCHUP_RUNS_DDL, + status_check=_OPS_RUN_STATUS_CHECK, + ) + + +def _rebuild_ops_status_table_if_stale( + conn: sqlite3.Connection, + *, + table: str, + table_ddl: str, + status_check: str, + indexes: tuple[str, ...] = (), +) -> None: + """Replace one stale status-constrained OPS table while retaining rows.""" + row = conn.execute( + "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = ?", + (table,), + ).fetchone() + if row is None or row[0] is None or status_check in str(row[0]): + return + + temporary_table = f"__polylogue_{table}_converged" + conn.execute(f"DROP TABLE IF EXISTS {temporary_table}") + temporary_ddl = table_ddl.replace( + f"CREATE TABLE IF NOT EXISTS {table}", + f"CREATE TABLE {temporary_table}", + 1, + ) + conn.executescript(temporary_ddl) + + columns = tuple(str(info[1]) for info in conn.execute(f"PRAGMA table_info({table})")) + target_columns = tuple(str(info[1]) for info in conn.execute(f"PRAGMA table_info({temporary_table})")) + if not set(target_columns).issubset(columns): + missing = set(target_columns) - set(columns) + raise RuntimeError(f"cannot converge {table}: missing source columns {sorted(missing)}") + select_columns = ", ".join( + "CASE status WHEN 'cancelled' THEN 'interrupted' ELSE status END" if column == "status" else column + for column in target_columns + ) + target_sql = ", ".join(target_columns) + conn.execute(f"INSERT INTO {temporary_table} ({target_sql}) SELECT {select_columns} FROM {table}") + conn.execute(f"DROP TABLE {table}") + conn.execute(f"ALTER TABLE {temporary_table} RENAME TO {table}") + for index_ddl in indexes: + conn.execute(index_ddl) + + def _embedding_catchup_run_outcome_columns(conn: sqlite3.Connection) -> dict[str, str]: existing = {str(row[1]) for row in conn.execute("PRAGMA table_info(embedding_catchup_runs)")} return { @@ -1577,6 +1658,7 @@ def _json_loads(raw_json: str | None) -> dict[str, object]: "OpsCompactState", "add_convergence_debt", "ensure_embedding_catchup_run_outcome_columns", + "ensure_ops_status_checks", "list_cursor_lag_samples", "list_fts_drift_samples", "list_schema_drift_samples", diff --git a/tests/unit/cli/test_embed_status_fast.py b/tests/unit/cli/test_embed_status_fast.py index 257f1c0ad5..f06ce13f07 100644 --- a/tests/unit/cli/test_embed_status_fast.py +++ b/tests/unit/cli/test_embed_status_fast.py @@ -961,6 +961,28 @@ def test_status_json_reads_latest_catchup_from_ops_db(tmp_path: Path) -> None: assert payload["latest_material_catchup_run"] == latest +def test_archive_backfill_stop_persists_canonical_interrupted_status(tmp_path: Path) -> None: + from polylogue.cli.commands.embed import _record_archive_backfill_run + + _record_archive_backfill_run( + tmp_path / "index.db", + started_at_ms=1_767_225_700_000, + status="stopped", + processed_sessions=2, + embedded_sessions=1, + skipped_sessions=0, + error_count=0, + embedded_messages=3, + estimated_cost_usd=0.001, + stop_reason="time limit reached", + configured_root=tmp_path, + ) + + with sqlite3.connect(tmp_path / "ops.db") as conn: + row = conn.execute("SELECT status FROM embedding_catchup_runs WHERE run_id IS NOT NULL").fetchone() + assert row == ("interrupted",) + + def test_status_json_distinguishes_latest_material_archive_catchup(tmp_path: Path) -> None: db_anchor = tmp_path / "index.db" archive_db = tmp_path / "index.db" diff --git a/tests/unit/cli/test_note.py b/tests/unit/cli/test_note.py index a58ba3bcf4..6c54d335e9 100644 --- a/tests/unit/cli/test_note.py +++ b/tests/unit/cli/test_note.py @@ -87,6 +87,7 @@ def test_terminal_note_kind_options_all_land_as_candidates(cli_workspace: dict[s "claim": AssertionKind.DECISION, "correction": AssertionKind.CORRECTION, "lesson": AssertionKind.LESSON, + "caveat": AssertionKind.CAVEAT, "highlight": AssertionKind.HIGHLIGHT, "prompt_eval": AssertionKind.PROMPT_EVAL, } diff --git a/tests/unit/core/test_readiness_capability.py b/tests/unit/core/test_readiness_capability.py index 3c5191c730..4fcf217182 100644 --- a/tests/unit/core/test_readiness_capability.py +++ b/tests/unit/core/test_readiness_capability.py @@ -662,10 +662,12 @@ def test_insight_entry_operation_and_catchup_adapters() -> None: ) ) operation = component_from_operation_status(OperationStatus.RUNNING, component="demo_import") + interrupted_operation = component_from_operation_status(OperationStatus.INTERRUPTED, component="demo_import") catchup = component_from_catchup_status(SimpleNamespace(mode="idle", failed_file_count=0, succeeded_file_count=7)) assert insight.state is CapabilityReadinessState.POISONED assert insight.evidence_refs == ("session_insight_status",) assert operation.state is CapabilityReadinessState.REBUILDING + assert interrupted_operation.state is CapabilityReadinessState.BLOCKED assert catchup.state is CapabilityReadinessState.READY assert catchup.counts["succeeded_file_count"] == 7 diff --git a/tests/unit/operations/test_operation_contract.py b/tests/unit/operations/test_operation_contract.py index b0452a0785..969d421632 100644 --- a/tests/unit/operations/test_operation_contract.py +++ b/tests/unit/operations/test_operation_contract.py @@ -13,6 +13,7 @@ import pytest from pydantic import ValidationError +from polylogue.core.enums import OperationStatus as CoreOperationStatus from polylogue.operations import ( ImportAck, ImportRequest, @@ -245,6 +246,7 @@ class TestOperationStatusEnum: """OperationStatus is a closed enum — adding values is an explicit change.""" def test_known_values(self) -> None: + assert OperationStatus is CoreOperationStatus assert {s.value for s in OperationStatus} == { "accepted", "rejected", @@ -252,4 +254,5 @@ def test_known_values(self) -> None: "running", "completed", "failed", + "interrupted", } diff --git a/tests/unit/storage/test_archive_tiers_ops_write.py b/tests/unit/storage/test_archive_tiers_ops_write.py index da7531541a..2ad2e60238 100644 --- a/tests/unit/storage/test_archive_tiers_ops_write.py +++ b/tests/unit/storage/test_archive_tiers_ops_write.py @@ -6,7 +6,7 @@ import pytest from polylogue.core.enums import Origin -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.ops_write import ( ROUTE_OBSERVATION_ROW_CAP, ArchiveCursorLagSample, @@ -49,6 +49,94 @@ def _connect(path: Path) -> sqlite3.Connection: return conn +def test_existing_ops_db_converges_old_status_checks_and_preserves_rows(tmp_path: Path) -> None: + """Same-version OPS bootstrap repairs old checks before interrupted writes.""" + ops_db = tmp_path / "ops.db" + conn = sqlite3.connect(ops_db) + try: + conn.executescript( + """ + CREATE TABLE ingest_attempts ( + attempt_id TEXT PRIMARY KEY, + source_path TEXT, + origin TEXT, + status TEXT NOT NULL CHECK(status IN ('running', 'completed', 'failed')), + phase TEXT, + storage_route TEXT, + started_at_ms INTEGER NOT NULL, + heartbeat_at_ms INTEGER, + finished_at_ms INTEGER, + parsed_raw_count INTEGER NOT NULL DEFAULT 0, + materialized_count INTEGER NOT NULL DEFAULT 0, + error_message TEXT, + source_paths_json TEXT NOT NULL DEFAULT '[]', + outcome_code TEXT NOT NULL DEFAULT 'legacy_unknown', + retryable INTEGER, + evidence_ref TEXT, + diagnostic TEXT, + remediation TEXT + ) STRICT; + CREATE TABLE embedding_catchup_runs ( + run_id TEXT PRIMARY KEY, + started_at_ms INTEGER NOT NULL, + finished_at_ms INTEGER, + status TEXT NOT NULL CHECK(status IN ('running', 'completed', 'failed', 'cancelled')), + origin TEXT, + scanned_sessions INTEGER NOT NULL DEFAULT 0, + embedded_sessions INTEGER NOT NULL DEFAULT 0, + skipped_sessions INTEGER NOT NULL DEFAULT 0, + error_count INTEGER NOT NULL DEFAULT 0, + embedded_messages INTEGER NOT NULL DEFAULT 0, + estimated_cost_usd REAL, + error_message TEXT + ) STRICT; + PRAGMA user_version = 1; + """ + ) + with pytest.raises(sqlite3.IntegrityError): + conn.execute( + "INSERT INTO embedding_catchup_runs (run_id, started_at_ms, status) " + "VALUES ('rejected', 1, 'interrupted')" + ) + conn.execute( + "INSERT INTO ingest_attempts (attempt_id, status, started_at_ms) VALUES ('legacy-attempt', 'completed', 1)" + ) + conn.execute( + "INSERT INTO embedding_catchup_runs " + "(run_id, started_at_ms, status, embedded_messages) VALUES ('legacy-run', 2, 'cancelled', 4)" + ) + conn.commit() + finally: + conn.close() + + initialize_archive_database(ops_db, ArchiveTier.OPS) + initialize_archive_database(ops_db, ArchiveTier.OPS) + + with sqlite3.connect(ops_db) as conn: + embedding_sql = conn.execute( + "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'embedding_catchup_runs'" + ).fetchone()[0] + assert "status IN ('running', 'completed', 'failed', 'interrupted')" in embedding_sql + assert "cancelled" not in embedding_sql + assert conn.execute("SELECT status, embedded_messages FROM embedding_catchup_runs").fetchone() == ( + "interrupted", + 4, + ) + assert conn.execute("SELECT status FROM ingest_attempts WHERE attempt_id = 'legacy-attempt'").fetchone() == ( + "completed", + ) + + record_ingest_attempt(conn, attempt_id="new-attempt", status="interrupted", started_at_ms=3) + upsert_embedding_catchup_run(conn, run_id="new-run", status="interrupted", started_at_ms=4) + + assert conn.execute("SELECT status FROM ingest_attempts WHERE attempt_id = 'new-attempt'").fetchone() == ( + "interrupted", + ) + assert conn.execute("SELECT status FROM embedding_catchup_runs WHERE run_id = 'new-run'").fetchone() == ( + "interrupted", + ) + + def test_ops_upsert_ingest_cursor_updates_single_row(tmp_path: Path) -> None: conn = _connect(tmp_path / "ops.db")