refactor: unify assertion and run status vocabularies

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,
 }

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,

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, ...]:

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",

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 = ()

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"]

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)
 

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

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``