From 1be4192d293df348be9198d3b305a9631f03571b Mon Sep 17 00:00:00 2001 From: Jiri Puc Date: Tue, 21 Jul 2026 17:28:08 +0200 Subject: [PATCH] feat: add operational correctness surfaces --- .../ops_operational_correctness_surfaces.yaml | 36 ++ plan/plans/phase-7-scale-ops.md | 25 +- src/tests/adapters/test_telemetry.py | 48 ++ src/tests/spine/test_operations.py | 508 ++++++++++++++++++ src/tests/surfaces/test_ops_cli.py | 96 ++++ .../adapters/selfhost/__init__.py | 2 + .../adapters/selfhost/telemetry.py | 45 ++ .../adapters/testing/__init__.py | 2 + .../adapters/testing/telemetry.py | 22 + src/ultimate_memory/eval/lifecycle.py | 18 +- src/ultimate_memory/model/__init__.py | 24 + src/ultimate_memory/model/operations.py | 153 ++++++ src/ultimate_memory/model/processing.py | 4 + .../profiles/selfhost_operations.py | 92 ++++ src/ultimate_memory/spine/__init__.py | 6 + src/ultimate_memory/spine/lifecycle.py | 17 + src/ultimate_memory/spine/operations.py | 296 ++++++++++ src/ultimate_memory/spine/work_ledger.py | 108 ++++ src/ultimate_memory/surfaces/cli.py | 83 ++- src/ultimate_memory/workers/__init__.py | 2 + src/ultimate_memory/workers/base.py | 109 +++- src/ultimate_memory/workers/operations.py | 42 ++ website/src/app/docs/project-status/page.mdx | 2 +- website/src/app/docs/reference/cli/page.mdx | 54 +- 24 files changed, 1767 insertions(+), 27 deletions(-) create mode 100644 plan/implementation_evals/eval_checks/ops_operational_correctness_surfaces.yaml create mode 100644 src/tests/adapters/test_telemetry.py create mode 100644 src/tests/spine/test_operations.py create mode 100644 src/tests/surfaces/test_ops_cli.py create mode 100644 src/ultimate_memory/adapters/selfhost/telemetry.py create mode 100644 src/ultimate_memory/adapters/testing/telemetry.py create mode 100644 src/ultimate_memory/model/operations.py create mode 100644 src/ultimate_memory/profiles/selfhost_operations.py create mode 100644 src/ultimate_memory/spine/operations.py create mode 100644 src/ultimate_memory/workers/operations.py diff --git a/plan/implementation_evals/eval_checks/ops_operational_correctness_surfaces.yaml b/plan/implementation_evals/eval_checks/ops_operational_correctness_surfaces.yaml new file mode 100644 index 00000000..0f08b48e --- /dev/null +++ b/plan/implementation_evals/eval_checks/ops_operational_correctness_surfaces.yaml @@ -0,0 +1,36 @@ +schema_version: 1 +id: ops_operational_correctness_surfaces +type: deterministic +description: Typed bounded inspection, one-row DLQ replay, strict worker telemetry, and existing P2/P3 rebuild CLI wiring pass their focused proofs. +tags: [ops, D7, D60, D61] +script: | + import json + import os + from pathlib import Path + import subprocess + import sys + + context = json.loads(Path(sys.argv[1]).read_text()) + root = Path(context["project_root"]) + if not os.environ.get("UGM_DATABASE_URL"): + print("UGM_DATABASE_URL must name an isolated PostgreSQL test database") + sys.exit(1) + result = subprocess.run( + [ + "uv", + "run", + "pytest", + "src/tests/adapters/test_telemetry.py", + "src/tests/spine/test_operations.py", + "src/tests/surfaces/test_ops_cli.py", + "-q", + "--tb=short", + ], + cwd=root, + text=True, + capture_output=True, + ) + print(result.stdout) + if result.stderr: + print(result.stderr, file=sys.stderr) + sys.exit(result.returncode) diff --git a/plan/plans/phase-7-scale-ops.md b/plan/plans/phase-7-scale-ops.md index ff44fc9f..21b05e89 100644 --- a/plan/plans/phase-7-scale-ops.md +++ b/plan/plans/phase-7-scale-ops.md @@ -31,7 +31,7 @@ their round trips. | WP-7.1 | Backfill lanes + seeding + reprocessing orchestration (version bumps) | orchestration §3–4 | Phase 6 | lane machinery | steady-state unaffected during backfill test | done | | WP-7.2 | Reproducible scale battery: D23 partitions/indexes, hub entities/lineages, recount cost, and provider-neutral read/write batching | schema §12; D23; lifecycle §11.5; orchestration §5; retrieval §13.7 | WP-7.1 | fixed synthetic profiles + report | shapes and batching invariants recorded; timings remain measurements, not hosted SLAs | done | | WP-7.3 | Cost metering + configurable budget enforcement | orchestration §4; schema §2 `cost_ledger` | WP-7.1 | enforcement + admin inspection | explicit fixture ceiling parks and resumes an over-budget lane; attribution is visible | done | -| WP-7.4 | Operational correctness surfaces + drills: typed telemetry, pipeline/DLQ inspection and replay, P2/P3 rebuild, currency-ledger audit | orchestration §6–7; D7, D60–D61 | WP-7.1 | telemetry/admin surfaces + deterministic drills | failures remain visible and drills pass without a dashboard or hosted control plane | planned | +| WP-7.4 | Operational correctness surfaces + drills: typed telemetry, pipeline/DLQ inspection and replay, P2/P3 rebuild, currency-ledger audit | orchestration §6–7; D7, D60–D61 | WP-7.1 | telemetry/admin surfaces + deterministic drills | failures remain visible and drills pass without a dashboard or hosted control plane | done | | WP-7.5 | **Hard-delete end-to-end**: design first, then purge active P1/P2/P3/K surfaces and prevent restore resurrection through the portable purge record/adapter contract | new design (gate #24); lifecycle §8; k_layers §10; S55 | gate #24 | forget pipeline | **S55 CI gate ON and green** across library-controlled surfaces + restore canary | blocked(#24) | | WP-7.6 | **Release engineering**: semver across PyPI + GHCR images + pinned compose; migrations-before-workers upgrade drill; quickstart cold-start release gate | packaging §1, §5–6; D62 | WP-7.1, rename/CLA gate | release pipeline | tagged release produces all artifacts; upgrade drill green; quickstart under target | blocked(rename-gate) | | WP-7.7 | **Export/import round-trip**: `ugm export`/`import` (Postgres dump + buckets + K repo + deletion state; projections rebuild on import) | packaging §6; D7, D62 | WP-7.1, WP-7.5 | export/import CLI | round-trip drill → no forgotten-data resurrection + S-battery subset green | planned | @@ -97,3 +97,26 @@ model-using stage records a deterministic logical call key and cascade tier befo response. OpenRouter responses without cost/token accounting fail visibly instead of degrading to zero spend; the deterministic test provider emits zero-cost usage so end-to-end worker tests prove the same production attribution path without network calls. + +## WP-7.4 implementation + +`OperationalCatalog.inspect` reads one repeatable PostgreSQL snapshot and emits a typed, +deployment-scoped report. Route/status counts and the two latest projection pointers are bounded by +their closed vocabularies. Every variable diagnostic reports its complete total plus a sample capped +by the one typed `UGM_OPERATIONAL_SAMPLE_LIMIT` setting: DLQ rows, DLQ stage/error/version groups, +poison targets, the component versions observed for each sampled poison target, and currency-ledger +mismatches. Error class is derived once from the last non-empty traceback line; complete tracebacks +and payloads remain available on sampled rows. + +`WorkLedger.replay_dead_letter` owns the only replay mutation. It locks one deployment-owned DLQ +row, refuses any other status, preserves consumed attempts and `last_error`, grants an explicit +additional attempt allowance, validates an optional lane against the immutable stage, and returns +the authoritative committed route and due time. A thin service announces that result through the +ordinary queue port after commit; there is no bulk replay controller or second work ledger. + +The existing worker exception boundary emits one provider-neutral `worker.run` event after each +committed success, failure, or budget park and emits nothing for `NO_WORK`. Exception export receives +the original exception object and exporter failures propagate. Self-hosting can write one JSON line +per event with the full exception cause chain; tests use an in-memory recorder. `ugm ops inspect`, +`replay`, and `rebuild` are thin local admin commands. Rebuild selects the existing production +`GraphRebuildWorker` or `CorpusFsBuilder`, so the drill cannot diverge into a recovery-only path. diff --git a/src/tests/adapters/test_telemetry.py b/src/tests/adapters/test_telemetry.py new file mode 100644 index 00000000..0a9e9376 --- /dev/null +++ b/src/tests/adapters/test_telemetry.py @@ -0,0 +1,48 @@ +"""Provider-neutral and self-host telemetry adapter contracts.""" + +from datetime import datetime +from datetime import UTC +from io import StringIO +import json + +from ultimate_memory.adapters.selfhost import JsonLineTelemetry +from ultimate_memory.model import TelemetryAttribute +from ultimate_memory.model import TelemetryEvent + + +def _event() -> TelemetryEvent: + return TelemetryEvent( + name="worker.run", + occurred_at=datetime(2026, 7, 21, tzinfo=UTC), + attributes=(TelemetryAttribute(name="outcome", value="dead_lettered"),), + ) + + +def test_json_lines_preserves_exception_cause_chain() -> None: + """The local exporter keeps structured data and both chained exceptions.""" + stream = StringIO() + telemetry = JsonLineTelemetry(stream=stream) + try: + try: + raise KeyError("root cause") + except KeyError as cause: + raise ValueError("outer failure") from cause + except ValueError as error: + telemetry.export_exception(event=_event(), exception=error) + + payload = json.loads(stream.getvalue()) + assert payload["name"] == "worker.run" + assert payload["exception"]["type"] == "ValueError" + assert "KeyError: 'root cause'" in payload["exception"]["traceback"] + assert "ValueError: outer failure" in payload["exception"]["traceback"] + + +def test_json_lines_flushes_one_event_per_line() -> None: + """Ordinary event output is valid compact JSON with one trailing newline.""" + stream = StringIO() + JsonLineTelemetry(stream=stream).export_event(event=_event()) + assert stream.getvalue().count("\n") == 1 + assert json.loads(stream.getvalue())["attributes"][0] == { + "name": "outcome", + "value": "dead_lettered", + } diff --git a/src/tests/spine/test_operations.py b/src/tests/spine/test_operations.py new file mode 100644 index 00000000..746fea38 --- /dev/null +++ b/src/tests/spine/test_operations.py @@ -0,0 +1,508 @@ +"""WP-7.4 bounded inspection, explicit DLQ replay, and worker telemetry proofs.""" + +from collections.abc import Iterator +from datetime import datetime +from datetime import timedelta +from datetime import UTC +from decimal import Decimal +from pathlib import Path +from uuid import UUID +from uuid import uuid4 + +from alembic import command +from alembic.config import Config +from pydantic import ValidationError +import pytest +from sqlalchemy import create_engine +from sqlalchemy import text +from sqlalchemy.engine import Engine + +from ultimate_memory.adapters.testing import RecordingTaskQueue +from ultimate_memory.adapters.testing import RecordingTelemetry +from ultimate_memory.model import CostBudget +from ultimate_memory.model import DeploymentBootstrapInput +from ultimate_memory.model import EnqueueWork +from ultimate_memory.model import LaneRouteError +from ultimate_memory.model import NonRetryableHandlerError +from ultimate_memory.model import PipelineStage +from ultimate_memory.model import ProcessingLane +from ultimate_memory.model import ProcessingTarget +from ultimate_memory.model import RecordCall +from ultimate_memory.model import RunResultOutcome +from ultimate_memory.model import WorkNotDeadLetterError +from ultimate_memory.spine import DeploymentBootstrapper +from ultimate_memory.spine import error_class_from_traceback +from ultimate_memory.spine import OperationalCatalog +from ultimate_memory.spine import OperationalSettings +from ultimate_memory.spine import WorkLedger +from ultimate_memory.spine import WorkLedgerSettings +from ultimate_memory.spine.settings import load_database_settings +from ultimate_memory.workers import DeadLetterReplayer +from ultimate_memory.workers import HandlerOutcome +from ultimate_memory.workers import HandlerRegistry +from ultimate_memory.workers import Worker + +_ROOT = Path(__file__).resolve().parents[3] +_DEPLOYMENT_ID = UUID("74000000-0000-0000-0000-000000000001") + + +@pytest.fixture(scope="module") +def database_engine() -> Iterator[Engine]: + """Apply structural head to the caller-provided isolated PostgreSQL DB.""" + try: + database_url = load_database_settings().sqlalchemy_url() + except ValidationError: + pytest.skip("UGM_DATABASE_URL is required for real operational proofs") + config = Config(str(_ROOT / "alembic.ini")) + config.set_main_option("sqlalchemy.url", database_url) + command.downgrade(config=config, revision="base") + command.upgrade(config=config, revision="head") + engine = create_engine(database_url) + try: + yield engine + finally: + engine.dispose() + + +@pytest.fixture(autouse=True) +def bootstrapped_deployment(database_engine: Engine) -> None: + """Give every proof fresh deployment-owned state.""" + with database_engine.begin() as connection: + connection.execute(text("TRUNCATE TABLE deployments CASCADE")) + DeploymentBootstrapper(engine=database_engine).bootstrap_deployment( + deployment_input=DeploymentBootstrapInput( + deployment_id=_DEPLOYMENT_ID, + slug="operations-test", + name="Operational correctness proofs", + default_language="en", + raw_bucket="mem://raw", + artifacts_bucket="mem://artifacts", + corpusfs_bucket="mem://corpusfs", + ) + ) + + +class _PermanentFailure: + """Raise a fresh permanent error and retain its identity for telemetry.""" + + def __init__(self) -> None: + self.errors: list[NonRetryableHandlerError] = [] + + def handle(self, **_: object) -> HandlerOutcome: + error = NonRetryableHandlerError("broken payload") + self.errors.append(error) + raise error + + +class _Success: + """Complete without follow-up work.""" + + def handle(self, **_: object) -> HandlerOutcome: + return HandlerOutcome() + + +class _RetryableFailure: + """Raise an ordinary error so the worker schedules another attempt.""" + + def handle(self, **_: object) -> HandlerOutcome: + raise RuntimeError("retry this work") + + +class _BrokenExporter: + """Prove telemetry failures are not swallowed after the state commit.""" + + def export_event(self, **_: object) -> None: + raise RuntimeError("export unavailable") + + def export_exception(self, **_: object) -> None: + raise RuntimeError("export unavailable") + + +def _work(*, target_id: UUID, component_version: str) -> EnqueueWork: + return EnqueueWork( + deployment_id=_DEPLOYMENT_ID, + target_kind=ProcessingTarget.DOCUMENT, + target_id=target_id, + stage=PipelineStage.CONVERT, + component_version=component_version, + content_hash=f"hash-{component_version}", + lane=ProcessingLane.STEADY, + payload={"component": component_version}, + ) + + +def test_bounded_inspection_replay_and_original_exception( + database_engine: Engine, +) -> None: + """Totals stay complete while samples cap; replay preserves failure history.""" + ledger = WorkLedger(engine=database_engine, settings=WorkLedgerSettings()) + target_id = uuid4() + processing_ids = tuple( + ledger.enqueue( + work=_work(target_id=target_id, component_version=version) + ).processing_id + for version in ("convert-v1", "convert-v2") + ) + handler = _PermanentFailure() + registry = HandlerRegistry() + registry.register(stage=PipelineStage.CONVERT, handler=handler) + telemetry = RecordingTelemetry() + worker = Worker(ledger=ledger, registry=registry, telemetry=telemetry) + + assert ( + worker.run_one( + deployment_id=_DEPLOYMENT_ID, + stage=PipelineStage.CONVERT, + lane=ProcessingLane.STEADY, + ).outcome + is RunResultOutcome.DEAD_LETTERED + ) + assert ( + worker.run_one( + deployment_id=_DEPLOYMENT_ID, + stage=PipelineStage.CONVERT, + lane=ProcessingLane.STEADY, + ).outcome + is RunResultOutcome.DEAD_LETTERED + ) + before_no_work = len(telemetry.exceptions) + assert ( + worker.run_one( + deployment_id=_DEPLOYMENT_ID, + stage=PipelineStage.CONVERT, + lane=ProcessingLane.STEADY, + ).outcome + is RunResultOutcome.NO_WORK + ) + assert len(telemetry.exceptions) == before_no_work + assert [exported[1] for exported in telemetry.exceptions] == handler.errors + assert all(event.name == "worker.run" for event, _ in telemetry.exceptions) + + with database_engine.begin() as connection: + connection.execute( + text( + "INSERT INTO projection_snapshots (snapshot_id, deployment_id, " + "plane, version, gcs_uri, status, is_latest, published_at) " + "VALUES (:snapshot_id, :deployment_id, " + "CAST(:plane AS projection_plane), :version, :uri, 'published', " + "true, now())" + ), + [ + { + "snapshot_id": uuid4(), + "deployment_id": _DEPLOYMENT_ID, + "plane": plane, + "version": f"{plane}-drill", + "uri": f"mem://{plane}", + } + for plane in ("P2_graph", "P3_corpusfs") + ], + ) + + report = OperationalCatalog( + engine=database_engine, settings=OperationalSettings(sample_limit=1) + ).inspect(deployment_id=_DEPLOYMENT_ID) + assert report.dead_letters.total == 2 + assert report.dead_letters.group_total == 2 + assert len(report.dead_letters.groups) == 1 + assert len(report.dead_letters.items) == 1 + assert report.dead_letters.groups[0].error_class.endswith( + "NonRetryableHandlerError" + ) + assert report.poison_targets.total == 1 + assert len(report.poison_targets.items) == 1 + assert report.poison_targets.items[0].component_versions == ("convert-v1",) + assert report.poison_targets.items[0].component_version_total == 2 + assert sum(route.count for route in report.routes) == 2 + assert tuple(snapshot.plane for snapshot in report.latest_projections) == ( + "P2_graph", + "P3_corpusfs", + ) + + processing_id = processing_ids[0] + with database_engine.connect() as connection: + failed = ( + connection.execute( + text( + "SELECT attempts, last_error FROM processing_state " + "WHERE processing_id = :processing_id" + ), + {"processing_id": processing_id}, + ) + .mappings() + .one() + ) + queue = RecordingTaskQueue() + replayed = DeadLetterReplayer(ledger=ledger, queue=queue).replay( + deployment_id=_DEPLOYMENT_ID, + processing_id=processing_id, + attempt_allowance=1, + lane=ProcessingLane.BACKFILL, + ) + assert replayed.attempts == failed["attempts"] == 1 + assert replayed.max_attempts == 2 + assert replayed.route.lane is ProcessingLane.BACKFILL + assert queue.announcements[0].processing_id == processing_id + with database_engine.connect() as connection: + reopened = ( + connection.execute( + text( + "SELECT status::text, attempts, max_attempts, last_error, " + "defer_reason, finished_at FROM processing_state " + "WHERE processing_id = :processing_id" + ), + {"processing_id": processing_id}, + ) + .mappings() + .one() + ) + assert reopened == { + "status": "pending", + "attempts": 1, + "max_attempts": 2, + "last_error": failed["last_error"], + "defer_reason": None, + "finished_at": None, + } + with pytest.raises(WorkNotDeadLetterError): + ledger.replay_dead_letter( + deployment_id=_DEPLOYMENT_ID, processing_id=processing_id + ) + + scheduled_at = datetime.now(UTC) + timedelta(hours=1) + scheduled = ledger.replay_dead_letter( + deployment_id=_DEPLOYMENT_ID, + processing_id=processing_ids[1], + attempt_allowance=2, + not_before=scheduled_at, + ) + assert scheduled.attempts == 1 + assert scheduled.max_attempts == 3 + assert scheduled.not_before == scheduled_at + with database_engine.connect() as connection: + defer_reason = connection.execute( + text( + "SELECT defer_reason::text FROM processing_state " + "WHERE processing_id = :processing_id" + ), + {"processing_id": processing_ids[1]}, + ).scalar_one() + assert defer_reason == "scheduled" + + unlaned_id = ledger.enqueue( + work=EnqueueWork( + deployment_id=_DEPLOYMENT_ID, + target_kind=ProcessingTarget.SNAPSHOT, + target_id=uuid4(), + stage=PipelineStage.BUILD_SNAPSHOT, + component_version="snapshot-broken", + content_hash="snapshot-hash", + lane=None, + ) + ).processing_id + unlaned_registry = HandlerRegistry() + unlaned_registry.register( + stage=PipelineStage.BUILD_SNAPSHOT, handler=_PermanentFailure() + ) + Worker(ledger=ledger, registry=unlaned_registry).run_one( + deployment_id=_DEPLOYMENT_ID, stage=PipelineStage.BUILD_SNAPSHOT, lane=None + ) + with pytest.raises(LaneRouteError): + ledger.replay_dead_letter( + deployment_id=_DEPLOYMENT_ID, + processing_id=unlaned_id, + lane=ProcessingLane.STEADY, + ) + with pytest.raises(ValueError, match="at least one"): + ledger.replay_dead_letter( + deployment_id=_DEPLOYMENT_ID, processing_id=unlaned_id, attempt_allowance=0 + ) + + +def test_currency_audit_uses_the_lifecycle_invariant(database_engine: Engine) -> None: + """A no-event claim must retain the schema's initial current=true state.""" + doc_id, claim_id = uuid4(), uuid4() + with database_engine.begin() as connection: + connection.execute( + text( + "INSERT INTO documents (doc_id, deployment_id, source_kind, " + "source_ref, title) VALUES (:doc, :deployment, 'test', 'currency', " + "'Currency')" + ), + {"doc": doc_id, "deployment": _DEPLOYMENT_ID}, + ) + connection.execute( + text( + "INSERT INTO claims (claim_id, deployment_id, doc_id, chunk_id, " + "claim_text, source_span, char_start, char_end, anchor_ok, " + "window_membership_ok, extractor_version, is_current_testimony) " + "VALUES (:claim, :deployment, :doc, :chunk, 'claim', 'span', 0, 4, " + "true, true, 'extractor-v1', false)" + ), + { + "claim": claim_id, + "deployment": _DEPLOYMENT_ID, + "doc": doc_id, + "chunk": uuid4(), + }, + ) + catalog = OperationalCatalog( + engine=database_engine, settings=OperationalSettings(sample_limit=1) + ) + audit = catalog.inspect(deployment_id=_DEPLOYMENT_ID).currency + assert audit.claims == 1 + assert audit.mismatch_total == 1 + assert audit.mismatches[0].claim_id == claim_id + assert audit.mismatches[0].ledger_current is True + with database_engine.begin() as connection: + connection.execute( + text("UPDATE claims SET is_current_testimony = true WHERE claim_id = :id"), + {"id": claim_id}, + ) + assert catalog.inspect(deployment_id=_DEPLOYMENT_ID).currency.mismatch_total == 0 + + +def test_success_state_commits_before_exporter_failure(database_engine: Engine) -> None: + """Telemetry is strict, but its failure cannot roll back completed work.""" + ledger = WorkLedger(engine=database_engine, settings=WorkLedgerSettings()) + processing_id = ledger.enqueue( + work=_work(target_id=uuid4(), component_version="convert-success") + ).processing_id + registry = HandlerRegistry() + registry.register(stage=PipelineStage.CONVERT, handler=_Success()) + worker = Worker(ledger=ledger, registry=registry, telemetry=_BrokenExporter()) + with pytest.raises(RuntimeError, match="export unavailable"): + worker.run_one( + deployment_id=_DEPLOYMENT_ID, + stage=PipelineStage.CONVERT, + lane=ProcessingLane.STEADY, + ) + with database_engine.connect() as connection: + status = connection.execute( + text( + "SELECT status::text FROM processing_state " + "WHERE processing_id = :processing_id" + ), + {"processing_id": processing_id}, + ).scalar_one() + assert status == "succeeded" + + +def test_retry_is_announced_before_exporter_failure(database_engine: Engine) -> None: + """Strict telemetry failure cannot suppress an already-committed retry wake.""" + ledger = WorkLedger( + engine=database_engine, + settings=WorkLedgerSettings(retry_backoff_base_s=0.0, retry_backoff_max_s=0.0), + ) + processing_id = ledger.enqueue( + work=_work(target_id=uuid4(), component_version="convert-retry") + ).processing_id + registry = HandlerRegistry() + registry.register(stage=PipelineStage.CONVERT, handler=_RetryableFailure()) + queue = RecordingTaskQueue() + worker = Worker( + ledger=ledger, registry=registry, queue=queue, telemetry=_BrokenExporter() + ) + + with pytest.raises(RuntimeError, match="export unavailable"): + worker.run_one( + deployment_id=_DEPLOYMENT_ID, + stage=PipelineStage.CONVERT, + lane=ProcessingLane.STEADY, + ) + + assert queue.announcements[0].processing_id == processing_id + with database_engine.connect() as connection: + status = connection.execute( + text( + "SELECT status::text FROM processing_state " + "WHERE processing_id = :processing_id" + ), + {"processing_id": processing_id}, + ).scalar_one() + assert status == "failed" + + +def test_budget_park_emits_one_worker_event_without_running_handler( + database_engine: Engine, +) -> None: + """A preflight park is visible and consumes no handler attempt.""" + ledger = WorkLedger( + engine=database_engine, + settings=WorkLedgerSettings( + budgets=( + CostBudget( + deployment_id=_DEPLOYMENT_ID, + stage=PipelineStage.CONVERT, + lane=ProcessingLane.STEADY, + window_seconds=3_600, + ceiling_usd=Decimal("1.00"), + ), + ) + ), + ) + charged = ledger.enqueue( + work=_work(target_id=uuid4(), component_version="convert-charged") + ).processing_id + claimed = ledger.claim_one( + deployment_id=_DEPLOYMENT_ID, + stage=PipelineStage.CONVERT, + lane=ProcessingLane.STEADY, + ) + assert claimed is not None + ledger.record_call( + call=RecordCall( + processing_id=charged, + call_key="provider-call", + model_name="fixture-model", + cost_usd=Decimal("1.00"), + ) + ) + ledger.complete(processing_id=charged) + parked_id = ledger.enqueue( + work=_work(target_id=uuid4(), component_version="convert-parked") + ).processing_id + registry = HandlerRegistry() + registry.register(stage=PipelineStage.CONVERT, handler=_Success()) + telemetry = RecordingTelemetry() + + result = Worker(ledger=ledger, registry=registry, telemetry=telemetry).run_one( + deployment_id=_DEPLOYMENT_ID, + stage=PipelineStage.CONVERT, + lane=ProcessingLane.STEADY, + ) + + assert result.processing_id == parked_id + assert result.outcome is RunResultOutcome.BUDGET_PARKED + assert len(telemetry.events) == 1 + attributes = {item.name: item.value for item in telemetry.events[0].attributes} + assert attributes["outcome"] == "budget_parked" + assert attributes["attempt"] is None + with database_engine.connect() as connection: + state = ( + connection.execute( + text( + "SELECT status::text, defer_reason::text, attempts " + "FROM processing_state WHERE processing_id = :processing_id" + ), + {"processing_id": parked_id}, + ) + .mappings() + .one() + ) + assert state == {"status": "pending", "defer_reason": "budget", "attempts": 0} + + +@pytest.mark.parametrize( + ("error", "expected"), + [ + ("Traceback\nValueError: bad", "ValueError"), + ("Traceback\npackage.CustomError", "package.CustomError"), + ("not an exception line", "unknown"), + (None, "unknown"), + ], +) +def test_error_class_derivation_is_pure_and_total( + error: str | None, expected: str +) -> None: + assert error_class_from_traceback(error) == expected diff --git a/src/tests/surfaces/test_ops_cli.py b/src/tests/surfaces/test_ops_cli.py new file mode 100644 index 00000000..9f4fdb4f --- /dev/null +++ b/src/tests/surfaces/test_ops_cli.py @@ -0,0 +1,96 @@ +"""Thin CLI wiring for the existing P2/P3 rebuild implementations.""" + +import json +from pathlib import Path +from uuid import UUID + +import pytest +import sqlalchemy + +from ultimate_memory.spine import settings as settings_module +from ultimate_memory.surfaces import cli_main +from ultimate_memory.workers import CorpusFsBuilder +from ultimate_memory.workers import GraphRebuildWorker + +_DEPLOYMENT_ID = UUID("74000000-0000-0000-0000-000000000001") + + +class _Settings: + def sqlalchemy_url(self) -> str: + return "postgresql+psycopg://unused" + + +class _Engine: + def __init__(self) -> None: + self.disposed = False + + def dispose(self) -> None: + self.disposed = True + + +@pytest.mark.parametrize("plane", ["p2", "p3"]) +def test_ops_rebuild_invokes_the_existing_builder( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], + plane: str, +) -> None: + """The admin surface adds no second rebuild implementation.""" + engine = _Engine() + calls: list[dict[str, object]] = [] + monkeypatch.setattr(settings_module, "load_database_settings", lambda: _Settings()) + monkeypatch.setattr(sqlalchemy, "create_engine", lambda _url: engine) + + def rebuild( + _self: GraphRebuildWorker, + *, + deployment_id: UUID, + workdir: Path, + version: str | None = None, + ) -> dict[str, object]: + calls.append( + { + "plane": "p2", + "deployment_id": deployment_id, + "workdir": workdir, + "version": version, + } + ) + return {"plane": "p2", "published": True} + + def build( + _self: CorpusFsBuilder, *, deployment_id: UUID, version: str | None = None + ) -> dict[str, object]: + calls.append( + {"plane": "p3", "deployment_id": deployment_id, "version": version} + ) + return {"plane": "p3", "published": True} + + monkeypatch.setattr(GraphRebuildWorker, "rebuild", rebuild) + monkeypatch.setattr(CorpusFsBuilder, "build", build) + workdir = tmp_path / "work" + result = cli_main( + [ + "ops", + "rebuild", + "--plane", + plane, + "--deployment", + str(_DEPLOYMENT_ID), + "--snapshot-root", + str(tmp_path / "snapshots"), + "--workdir", + str(workdir), + "--version", + "drill-v1", + ] + ) + + assert result == 0 + assert calls[0]["plane"] == plane + assert calls[0]["deployment_id"] == _DEPLOYMENT_ID + assert calls[0]["version"] == "drill-v1" + if plane == "p2": + assert calls[0]["workdir"] == workdir + assert json.loads(capsys.readouterr().out) == {"plane": plane, "published": True} + assert engine.disposed is True diff --git a/src/ultimate_memory/adapters/selfhost/__init__.py b/src/ultimate_memory/adapters/selfhost/__init__.py index bafe314c..7df88bf6 100644 --- a/src/ultimate_memory/adapters/selfhost/__init__.py +++ b/src/ultimate_memory/adapters/selfhost/__init__.py @@ -11,11 +11,13 @@ from ultimate_memory.adapters.selfhost.queue import SelfHostTaskQueue from ultimate_memory.adapters.selfhost.queue import SelfHostWorkerLoop from ultimate_memory.adapters.selfhost.queue import TokenBucket +from ultimate_memory.adapters.selfhost.telemetry import JsonLineTelemetry from ultimate_memory.adapters.selfhost.watcher import LocalDirectoryWatcher __all__ = ( "LanceChunkIndex", "LocalDirectoryWatcher", + "JsonLineTelemetry", "LocalFSObjectStore", "AuditedRawReader", "LocalMountPublisher", diff --git a/src/ultimate_memory/adapters/selfhost/telemetry.py b/src/ultimate_memory/adapters/selfhost/telemetry.py new file mode 100644 index 00000000..319e78ff --- /dev/null +++ b/src/ultimate_memory/adapters/selfhost/telemetry.py @@ -0,0 +1,45 @@ +"""Simple JSON-lines telemetry for self-hosted process logs.""" + +import json +import sys +from threading import Lock +import traceback +from typing import TextIO + +from ultimate_memory.model import TelemetryEvent + + +class JsonLineTelemetry: + """Write one complete structured event per line to a process-owned stream.""" + + def __init__(self, *, stream: TextIO | None = None) -> None: + """Use stdout by default and serialize concurrent worker writes.""" + self._stream = stream or sys.stdout + self._lock = Lock() + + def export_event(self, *, event: TelemetryEvent) -> None: + """Write and flush one ordinary event.""" + self._write(payload=event.model_dump(mode="json")) + + def export_exception( + self, *, event: TelemetryEvent, exception: BaseException + ) -> None: + """Write the event plus the supplied exception's complete cause chain.""" + payload = event.model_dump(mode="json") + payload["exception"] = { + "type": type(exception).__qualname__, + "message": str(exception), + "traceback": "".join( + traceback.TracebackException.from_exception(exception).format( + chain=True + ) + ), + } + self._write(payload=payload) + + def _write(self, *, payload: dict[str, object]) -> None: + """Keep every event atomic at the stream boundary.""" + line = json.dumps(payload, sort_keys=True, separators=(",", ":")) + with self._lock: + self._stream.write(line + "\n") + self._stream.flush() diff --git a/src/ultimate_memory/adapters/testing/__init__.py b/src/ultimate_memory/adapters/testing/__init__.py index 81fb63fa..11966e34 100644 --- a/src/ultimate_memory/adapters/testing/__init__.py +++ b/src/ultimate_memory/adapters/testing/__init__.py @@ -4,10 +4,12 @@ from ultimate_memory.adapters.testing.model_provider import FakeModelProvider from ultimate_memory.adapters.testing.queue import RecordedAnnouncement from ultimate_memory.adapters.testing.queue import RecordingTaskQueue +from ultimate_memory.adapters.testing.telemetry import RecordingTelemetry __all__ = ( "FakeModelProvider", "NoopCostMeter", "RecordedAnnouncement", "RecordingTaskQueue", + "RecordingTelemetry", ) diff --git a/src/ultimate_memory/adapters/testing/telemetry.py b/src/ultimate_memory/adapters/testing/telemetry.py new file mode 100644 index 00000000..b670f5c2 --- /dev/null +++ b/src/ultimate_memory/adapters/testing/telemetry.py @@ -0,0 +1,22 @@ +"""In-memory telemetry exporter preserving exception identity for tests.""" + +from ultimate_memory.model import TelemetryEvent + + +class RecordingTelemetry: + """Record structured events and original exception objects in call order.""" + + def __init__(self) -> None: + """Start with no exports.""" + self.events: list[TelemetryEvent] = [] + self.exceptions: list[tuple[TelemetryEvent, BaseException]] = [] + + def export_event(self, *, event: TelemetryEvent) -> None: + """Record a non-exception event.""" + self.events.append(event) + + def export_exception( + self, *, event: TelemetryEvent, exception: BaseException + ) -> None: + """Record the exact exception instance supplied by the worker.""" + self.exceptions.append((event, exception)) diff --git a/src/ultimate_memory/eval/lifecycle.py b/src/ultimate_memory/eval/lifecycle.py index 73e8842d..47c80cbd 100644 --- a/src/ultimate_memory/eval/lifecycle.py +++ b/src/ultimate_memory/eval/lifecycle.py @@ -41,6 +41,7 @@ from ultimate_memory.model import CanaryCase from ultimate_memory.model import EvalSuite from ultimate_memory.model import LifecycleReport +from ultimate_memory.spine.lifecycle import CURRENCY_CACHE_MISMATCH_SQL def run_lifecycle_suite( @@ -213,22 +214,7 @@ def _load_canaries( # ledger is exactly the corruption this exists to catch. "currency_cache_matches_ledger", False, - text( - """ - SELECT cl.claim_id - FROM claims cl - LEFT JOIN LATERAL ( - SELECT e.became_current - FROM testimony_currency_events e - WHERE e.claim_id = cl.claim_id - ORDER BY e.occurred_at DESC, e.event_id DESC - LIMIT 1 - ) last ON true - WHERE cl.deployment_id = :deployment_id - AND cl.is_current_testimony - <> coalesce(last.became_current, true) - """ - ), + text(CURRENCY_CACHE_MISMATCH_SQL), ), ( # D54: BOTH cached counts must equal a recompute (quiescent only — diff --git a/src/ultimate_memory/model/__init__.py b/src/ultimate_memory/model/__init__.py index 7102945e..d117864a 100644 --- a/src/ultimate_memory/model/__init__.py +++ b/src/ultimate_memory/model/__init__.py @@ -218,6 +218,17 @@ from ultimate_memory.model.object_store import ObjectKeyEscapesRootError from ultimate_memory.model.operational_scale import OperationalScaleMeasurement from ultimate_memory.model.operational_scale import OperationalScaleReport +from ultimate_memory.model.operations import CurrencyLedgerAudit +from ultimate_memory.model.operations import CurrencyMismatch +from ultimate_memory.model.operations import DeadLetterGroup +from ultimate_memory.model.operations import DeadLetterRecord +from ultimate_memory.model.operations import DeadLetterReplayResult +from ultimate_memory.model.operations import DeadLetterReport +from ultimate_memory.model.operations import OperationalReport +from ultimate_memory.model.operations import PipelineRouteStatus +from ultimate_memory.model.operations import PoisonTargetRecord +from ultimate_memory.model.operations import PoisonTargetReport +from ultimate_memory.model.operations import ProjectionSnapshotState from ultimate_memory.model.processing import BackfillNotDrainedError from ultimate_memory.model.processing import BackfillSeedRequest from ultimate_memory.model.processing import BackfillSeedResult @@ -238,6 +249,7 @@ from ultimate_memory.model.processing import RunResultOutcome from ultimate_memory.model.processing import UnknownStageHandlerError from ultimate_memory.model.processing import WorkLedgerError +from ultimate_memory.model.processing import WorkNotDeadLetterError from ultimate_memory.model.processing import WorkNotFoundError from ultimate_memory.model.processing import WorkNotRunningError from ultimate_memory.model.queue import PipelineStage @@ -309,6 +321,8 @@ "CostBudget", "CostBudgetStatus", "CostTierSpend", + "CurrencyLedgerAudit", + "CurrencyMismatch", "ConnectorNotFoundError", "ContextPrefix", "ConversionError", @@ -322,6 +336,10 @@ "DecisionRecord", "DecisionType", "DeferReason", + "DeadLetterGroup", + "DeadLetterRecord", + "DeadLetterReplayResult", + "DeadLetterReport", "DeploymentBootstrapConflictError", "DeploymentBootstrapInput", "DeploymentBootstrapResult", @@ -373,6 +391,7 @@ "ObservationVerdict", "OperationalScaleMeasurement", "OperationalScaleReport", + "OperationalReport", "OtherPredicateGrammarError", "P1ChunkRow", "P1ClaimRow", @@ -382,9 +401,13 @@ "PerimeterCredential", "PipelineComponent", "PipelineStage", + "PipelineRouteStatus", + "PoisonTargetRecord", + "PoisonTargetReport", "ProcessingLane", "ProcessingStatus", "ProcessingTarget", + "ProjectionSnapshotState", "ProviderAccountingError", "ProviderCallUsage", "PublishedMounts", @@ -538,6 +561,7 @@ "UploadRecord", "Validity", "WorkLedgerError", + "WorkNotDeadLetterError", "WorkNotFoundError", "WorkNotRunningError", "merge_authored_review_payloads", diff --git a/src/ultimate_memory/model/operations.py b/src/ultimate_memory/model/operations.py new file mode 100644 index 00000000..921aff23 --- /dev/null +++ b/src/ultimate_memory/model/operations.py @@ -0,0 +1,153 @@ +"""Typed, bounded operational inspection and dead-letter replay values.""" + +from typing import Literal +from uuid import UUID + +from pydantic import BaseModel +from pydantic import ConfigDict +from pydantic import Field + +from ultimate_memory.model.processing import ProcessingStatus +from ultimate_memory.model.processing import ProcessingTarget +from ultimate_memory.model.queue import PipelineStage +from ultimate_memory.model.queue import ProcessingLane +from ultimate_memory.model.queue import QueueRoute +from ultimate_memory.model.queue import UTCDateTime + + +class PipelineRouteStatus(BaseModel): + """One deployment route/status aggregate; the enum vocabulary bounds its size.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + stage: PipelineStage + lane: ProcessingLane | None + status: ProcessingStatus + count: int = Field(ge=0) + + +class DeadLetterGroup(BaseModel): + """One dead-letter aggregate used to spot a failing version or exception.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + stage: PipelineStage + error_class: str + component_version: str + count: int = Field(ge=1) + oldest_enqueued_at: UTCDateTime + latest_finished_at: UTCDateTime | None + + +class DeadLetterRecord(BaseModel): + """One bounded DLQ sample retaining the complete diagnostic and replay input.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + processing_id: UUID + target_kind: ProcessingTarget + target_id: UUID + stage: PipelineStage + component_version: str + content_hash: str + lane: ProcessingLane | None + attempts: int = Field(ge=0) + max_attempts: int = Field(ge=1) + error_class: str + last_error: str | None + payload: dict[str, object] | None + enqueued_at: UTCDateTime + finished_at: UTCDateTime | None + + +class DeadLetterReport(BaseModel): + """Deployment DLQ totals plus independently bounded aggregate/detail samples.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + total: int = Field(ge=0) + group_total: int = Field(ge=0) + groups: tuple[DeadLetterGroup, ...] + items: tuple[DeadLetterRecord, ...] + + +class PoisonTargetRecord(BaseModel): + """A target dead-lettered by at least two component versions for one stage.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + target_kind: ProcessingTarget + target_id: UUID + stage: PipelineStage + component_version_total: int = Field(ge=2) + component_versions: tuple[str, ...] + dead_letters: int = Field(ge=2) + + +class PoisonTargetReport(BaseModel): + """Total poison targets and a bounded sample.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + total: int = Field(ge=0) + items: tuple[PoisonTargetRecord, ...] + + +class ProjectionSnapshotState(BaseModel): + """The current P2 or P3 snapshot pointer, if that plane has one.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + snapshot_id: UUID + plane: Literal["P2_graph", "P3_corpusfs"] + version: str + store_uri: str + row_counts: dict[str, object] | None + built_at: UTCDateTime + published_at: UTCDateTime | None + + +class CurrencyMismatch(BaseModel): + """A claim whose cached currency differs from the append-only ledger truth.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + claim_id: UUID + cached_current: bool + ledger_current: bool + + +class CurrencyLedgerAudit(BaseModel): + """Deployment claim count and bounded currency-ledger mismatch evidence.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + claims: int = Field(ge=0) + mismatch_total: int = Field(ge=0) + mismatches: tuple[CurrencyMismatch, ...] + + +class OperationalReport(BaseModel): + """One coherent, deployment-scoped operational inspection snapshot.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + deployment_id: UUID + generated_at: UTCDateTime + routes: tuple[PipelineRouteStatus, ...] + dead_letters: DeadLetterReport + poison_targets: PoisonTargetReport + latest_projections: tuple[ProjectionSnapshotState, ...] + currency: CurrencyLedgerAudit + + +class DeadLetterReplayResult(BaseModel): + """Authoritative route and due time after reopening one dead-letter row.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + processing_id: UUID + route: QueueRoute + not_before: UTCDateTime + attempts: int = Field(ge=0) + max_attempts: int = Field(ge=1) diff --git a/src/ultimate_memory/model/processing.py b/src/ultimate_memory/model/processing.py index f4cfa224..94e83b49 100644 --- a/src/ultimate_memory/model/processing.py +++ b/src/ultimate_memory/model/processing.py @@ -212,6 +212,10 @@ class WorkNotRunningError(WorkLedgerError): """An operation that requires a running attempt hit a non-running row.""" +class WorkNotDeadLetterError(WorkLedgerError): + """An operation that requires a dead-letter row hit another status.""" + + class NonRetryableHandlerError(Exception): """A handler failure classified as permanent: the work dead-letters immediately.""" diff --git a/src/ultimate_memory/profiles/selfhost_operations.py b/src/ultimate_memory/profiles/selfhost_operations.py new file mode 100644 index 00000000..2e6466db --- /dev/null +++ b/src/ultimate_memory/profiles/selfhost_operations.py @@ -0,0 +1,92 @@ +"""Self-host composition root for the optional local operational commands.""" + +from datetime import datetime +from pathlib import Path +from typing import Self +from uuid import UUID + +import sqlalchemy +from sqlalchemy.engine import Engine + +from ultimate_memory.adapters.selfhost import LocalFSObjectStore +from ultimate_memory.adapters.selfhost import SelfHostTaskQueue +from ultimate_memory.model import DeadLetterReplayResult +from ultimate_memory.model import OperationalReport +from ultimate_memory.model import ProcessingLane +from ultimate_memory.spine import OperationalCatalog +from ultimate_memory.spine import OperationalSettings +from ultimate_memory.spine import ProjectionCatalog +from ultimate_memory.spine import WorkLedger +from ultimate_memory.spine import WorkLedgerSettings +from ultimate_memory.spine.settings import load_database_settings +from ultimate_memory.workers import CorpusFsBuilder +from ultimate_memory.workers import DeadLetterReplayer +from ultimate_memory.workers import GraphRebuildWorker + + +class SelfHostOperations: + """Compose local adapters around one explicitly owned database engine.""" + + def __init__(self, *, engine: Engine) -> None: + """Take ownership of an engine created for one CLI invocation.""" + self._engine = engine + + @classmethod + def from_settings(cls) -> Self: + """Create the local composition from the typed database setting.""" + return cls( + engine=sqlalchemy.create_engine(load_database_settings().sqlalchemy_url()) + ) + + def close(self) -> None: + """Dispose the command-owned connection pool.""" + self._engine.dispose() + + def inspect(self, *, deployment_id: UUID) -> OperationalReport: + """Build one bounded typed report.""" + return OperationalCatalog( + engine=self._engine, settings=OperationalSettings() + ).inspect(deployment_id=deployment_id) + + def replay( + self, + *, + deployment_id: UUID, + processing_id: UUID, + attempt_allowance: int, + lane: ProcessingLane | None, + not_before: datetime | None, + ) -> DeadLetterReplayResult: + """Compose the authoritative replay transition with local delivery.""" + ledger = WorkLedger(engine=self._engine, settings=WorkLedgerSettings()) + return DeadLetterReplayer( + ledger=ledger, queue=SelfHostTaskQueue(ledger=ledger) + ).replay( + deployment_id=deployment_id, + processing_id=processing_id, + attempt_allowance=attempt_allowance, + lane=lane, + not_before=not_before, + ) + + def rebuild( + self, + *, + plane: str, + deployment_id: UUID, + snapshot_root: Path, + workdir: Path, + version: str, + ) -> dict[str, object]: + """Invoke the existing whole-rebuild implementation for P2 or P3.""" + catalog = ProjectionCatalog(engine=self._engine) + store = LocalFSObjectStore(root=snapshot_root) + if plane == "p2": + return GraphRebuildWorker(catalog=catalog, snapshot_store=store).rebuild( + deployment_id=deployment_id, workdir=workdir, version=version + ) + if plane == "p3": + return CorpusFsBuilder(catalog=catalog, snapshot_store=store).build( + deployment_id=deployment_id, version=version + ) + raise ValueError(f"unknown projection plane {plane!r}") diff --git a/src/ultimate_memory/spine/__init__.py b/src/ultimate_memory/spine/__init__.py index b2364a0a..6eefbae3 100644 --- a/src/ultimate_memory/spine/__init__.py +++ b/src/ultimate_memory/spine/__init__.py @@ -28,6 +28,9 @@ ) from ultimate_memory.spine.observation_adjudication import ObservationAdjudicator from ultimate_memory.spine.observation_adjudication import ObservationSettings +from ultimate_memory.spine.operations import error_class_from_traceback +from ultimate_memory.spine.operations import OperationalCatalog +from ultimate_memory.spine.operations import OperationalSettings from ultimate_memory.spine.projection import ProjectionCatalog from ultimate_memory.spine.recipes import CANONICAL_RECIPES from ultimate_memory.spine.recipes import RecipeRegistry @@ -62,6 +65,9 @@ "OBSERVATION_ADJUDICATOR_VERSION", "ObservationAdjudicator", "ObservationSettings", + "OperationalCatalog", + "OperationalSettings", + "error_class_from_traceback", "RESOLVER_VERSION", "LifecycleCatalog", "KnowledgeCompilationError", diff --git a/src/ultimate_memory/spine/lifecycle.py b/src/ultimate_memory/spine/lifecycle.py index c0f54e89..62bb41c1 100644 --- a/src/ultimate_memory/spine/lifecycle.py +++ b/src/ultimate_memory/spine/lifecycle.py @@ -19,6 +19,23 @@ from ultimate_memory.model import CurrencyTransition from ultimate_memory.model import ReconciliationDelta +CURRENCY_CACHE_MISMATCH_SQL = """ + SELECT cl.claim_id, + cl.is_current_testimony AS cached_current, + coalesce(last.became_current, true) AS ledger_current + FROM claims cl + LEFT JOIN LATERAL ( + SELECT e.became_current + FROM testimony_currency_events e + WHERE e.claim_id = cl.claim_id + ORDER BY e.occurred_at DESC, e.event_id DESC + LIMIT 1 + ) last ON true + WHERE cl.deployment_id = :deployment_id + AND cl.is_current_testimony <> coalesce(last.became_current, true) +""" +"""Single source for the D33 currency-cache versus append-only-ledger invariant.""" + class LifecycleCatalog: """Currency transitions, the D54 recount, per-shape closure, deletion.""" diff --git a/src/ultimate_memory/spine/operations.py b/src/ultimate_memory/spine/operations.py new file mode 100644 index 00000000..9d97cfba --- /dev/null +++ b/src/ultimate_memory/spine/operations.py @@ -0,0 +1,296 @@ +"""Bounded read-only operational inspection over the durable Postgres truth.""" + +import re +from uuid import UUID + +from pydantic import Field +from pydantic_settings import BaseSettings +from pydantic_settings import SettingsConfigDict +from sqlalchemy import text +from sqlalchemy.engine import Connection +from sqlalchemy.engine import Engine +from sqlalchemy.sql.elements import TextClause + +from ultimate_memory.model import CurrencyLedgerAudit +from ultimate_memory.model import CurrencyMismatch +from ultimate_memory.model import DeadLetterGroup +from ultimate_memory.model import DeadLetterRecord +from ultimate_memory.model import DeadLetterReport +from ultimate_memory.model import OperationalReport +from ultimate_memory.model import PipelineRouteStatus +from ultimate_memory.model import PoisonTargetRecord +from ultimate_memory.model import PoisonTargetReport +from ultimate_memory.model import ProjectionSnapshotState +from ultimate_memory.spine.lifecycle import CURRENCY_CACHE_MISMATCH_SQL + +_ERROR_CLASS = re.compile(r"^[A-Za-z_][A-Za-z0-9_.]*$") + + +class OperationalSettings(BaseSettings): + """One cap shared by every variable-size operational detail collection.""" + + model_config = SettingsConfigDict(env_prefix="UGM_OPERATIONAL_", extra="ignore") + + sample_limit: int = Field(default=20, ge=1, le=1_000) + + +class OperationalCatalog: + """Build a coherent deployment report without a dashboard or control plane.""" + + def __init__(self, *, engine: Engine, settings: OperationalSettings) -> None: + """Bind an engine and the single explicit sample bound.""" + self._engine = engine + self._settings = settings + + def inspect(self, *, deployment_id: UUID) -> OperationalReport: + """Return aggregate truth plus bounded evidence in one repeatable-read view.""" + parameters = { + "deployment_id": deployment_id, + "sample_limit": self._settings.sample_limit, + } + with self._engine.connect().execution_options( + isolation_level="REPEATABLE READ" + ) as connection: + generated_at = connection.execute(text("SELECT now()")).scalar_one() + routes = tuple( + PipelineRouteStatus(**row) + for row in connection.execute(_ROUTE_STATUS, parameters).mappings() + ) + dead_letters = self._dead_letters( + connection=connection, parameters=parameters + ) + poison_targets = self._poison_targets( + connection=connection, parameters=parameters + ) + latest_projections = tuple( + ProjectionSnapshotState(**row) + for row in connection.execute( + _LATEST_PROJECTIONS, parameters + ).mappings() + ) + currency = self._currency(connection=connection, parameters=parameters) + return OperationalReport( + deployment_id=deployment_id, + generated_at=generated_at, + routes=routes, + dead_letters=dead_letters, + poison_targets=poison_targets, + latest_projections=latest_projections, + currency=currency, + ) + + def _dead_letters( + self, *, connection: Connection, parameters: dict[str, object] + ) -> DeadLetterReport: + """Return DLQ totals and bounded grouped/detail evidence.""" + total = _count(connection=connection, statement=_DLQ_TOTAL, values=parameters) + group_total = _count( + connection=connection, statement=_DLQ_GROUP_TOTAL, values=parameters + ) + groups = tuple( + DeadLetterGroup(**row) + for row in connection.execute(_DLQ_GROUPS, parameters).mappings() + ) + items = tuple( + DeadLetterRecord( + **row, error_class=error_class_from_traceback(row["last_error"]) + ) + for row in connection.execute(_DLQ_ITEMS, parameters).mappings() + ) + return DeadLetterReport( + total=total, group_total=group_total, groups=groups, items=items + ) + + def _poison_targets( + self, *, connection: Connection, parameters: dict[str, object] + ) -> PoisonTargetReport: + """Find targets whose same stage failed under multiple component versions.""" + total = _count( + connection=connection, statement=_POISON_TOTAL, values=parameters + ) + items = tuple( + PoisonTargetRecord( + target_kind=row["target_kind"], + target_id=row["target_id"], + stage=row["stage"], + component_version_total=row["component_version_total"], + component_versions=tuple(row["component_versions"]), + dead_letters=row["dead_letters"], + ) + for row in connection.execute(_POISON_ITEMS, parameters).mappings() + ) + return PoisonTargetReport(total=total, items=items) + + def _currency( + self, *, connection: Connection, parameters: dict[str, object] + ) -> CurrencyLedgerAudit: + """Apply the exact lifecycle invariant and bound its diagnostic rows.""" + claims = _count( + connection=connection, statement=_CLAIM_TOTAL, values=parameters + ) + mismatch_total = _count( + connection=connection, statement=_CURRENCY_TOTAL, values=parameters + ) + mismatches = tuple( + CurrencyMismatch(**row) + for row in connection.execute(_CURRENCY_ITEMS, parameters).mappings() + ) + return CurrencyLedgerAudit( + claims=claims, mismatch_total=mismatch_total, mismatches=mismatches + ) + + +def error_class_from_traceback(error: str | None) -> str: + """Derive one stable class from the last non-empty traceback line.""" + if error is None: + return "unknown" + lines = [line.strip() for line in error.splitlines() if line.strip()] + if not lines: + return "unknown" + candidate = lines[-1].split(":", maxsplit=1)[0].strip() + return candidate if _ERROR_CLASS.fullmatch(candidate) is not None else "unknown" + + +def _count( + *, connection: Connection, statement: TextClause, values: dict[str, object] +) -> int: + """Normalize one PostgreSQL count scalar.""" + return int(connection.execute(statement, values).scalar_one()) + + +_ROUTE_STATUS = text( + """ + SELECT stage, lane, status, count(*) AS count + FROM processing_state + WHERE deployment_id = :deployment_id + GROUP BY stage, lane, status + ORDER BY stage, lane NULLS FIRST, status + """ +) + +_CLASSIFIED_DEAD_LETTERS = """ + WITH dead_letters AS ( + SELECT stage, component_version, enqueued_at, finished_at, + btrim(reverse(split_part( + reverse(rtrim(coalesce(last_error, ''), E' \\t\\n\\r')), + E'\\n', 1 + ))) AS last_line + FROM processing_state + WHERE deployment_id = :deployment_id AND status = 'dead_letter' + ), classified AS ( + SELECT stage, component_version, enqueued_at, finished_at, + CASE + WHEN split_part(last_line, ':', 1) + ~ '^[A-Za-z_][A-Za-z0-9_.]*$' + THEN split_part(last_line, ':', 1) + ELSE 'unknown' + END AS error_class + FROM dead_letters + ) +""" + +_DLQ_TOTAL = text( + """ + SELECT count(*) FROM processing_state + WHERE deployment_id = :deployment_id AND status = 'dead_letter' + """ +) + +_DLQ_GROUP_TOTAL = text( + _CLASSIFIED_DEAD_LETTERS + + """ + SELECT count(*) FROM ( + SELECT 1 FROM classified + GROUP BY stage, error_class, component_version + ) groups + """ +) + +_DLQ_GROUPS = text( + _CLASSIFIED_DEAD_LETTERS + + """ + SELECT stage, error_class, component_version, count(*) AS count, + min(enqueued_at) AS oldest_enqueued_at, + max(finished_at) AS latest_finished_at + FROM classified + GROUP BY stage, error_class, component_version + ORDER BY count(*) DESC, stage, error_class, component_version + LIMIT :sample_limit + """ +) + +_DLQ_ITEMS = text( + """ + SELECT processing_id, target_kind, target_id, stage, component_version, + content_hash, lane, attempts, max_attempts, last_error, payload, + enqueued_at, finished_at + FROM processing_state + WHERE deployment_id = :deployment_id AND status = 'dead_letter' + ORDER BY finished_at DESC NULLS LAST, processing_id + LIMIT :sample_limit + """ +) + +_POISON_GROUPS = """ + SELECT target_kind, target_id, stage, + count(DISTINCT component_version) AS component_version_total, + count(*) AS dead_letters + FROM processing_state + WHERE deployment_id = :deployment_id AND status = 'dead_letter' + GROUP BY target_kind, target_id, stage + HAVING count(DISTINCT component_version) >= 2 +""" + +_POISON_TOTAL = text("SELECT count(*) FROM (" + _POISON_GROUPS + ") poison") + +_POISON_ITEMS = text( + "WITH poison AS (" + + _POISON_GROUPS + + ")" + + """ + SELECT poison.*, + ARRAY( + SELECT DISTINCT versioned.component_version + FROM processing_state versioned + WHERE versioned.deployment_id = :deployment_id + AND versioned.status = 'dead_letter' + AND versioned.target_kind = poison.target_kind + AND versioned.target_id = poison.target_id + AND versioned.stage = poison.stage + ORDER BY versioned.component_version + LIMIT :sample_limit + ) AS component_versions + FROM poison + ORDER BY poison.dead_letters DESC, poison.target_kind, + poison.target_id, poison.stage + LIMIT :sample_limit + """ +) + +_LATEST_PROJECTIONS = text( + """ + SELECT snapshot_id, plane, version, gcs_uri AS store_uri, row_counts, + built_at, published_at + FROM projection_snapshots + WHERE deployment_id = :deployment_id + AND plane IN ('P2_graph', 'P3_corpusfs') + AND is_latest + ORDER BY plane + """ +) + +_CLAIM_TOTAL = text("SELECT count(*) FROM claims WHERE deployment_id = :deployment_id") + +_CURRENCY_TOTAL = text( + "WITH mismatches AS (" + + CURRENCY_CACHE_MISMATCH_SQL + + ") SELECT count(*) FROM mismatches" +) + +_CURRENCY_ITEMS = text( + CURRENCY_CACHE_MISMATCH_SQL + + """ + ORDER BY cl.claim_id + LIMIT :sample_limit + """ +) diff --git a/src/ultimate_memory/spine/work_ledger.py b/src/ultimate_memory/spine/work_ledger.py index 1203e8bc..4431b30d 100644 --- a/src/ultimate_memory/spine/work_ledger.py +++ b/src/ultimate_memory/spine/work_ledger.py @@ -10,6 +10,7 @@ from dataclasses import dataclass from datetime import datetime +from datetime import timedelta from decimal import Decimal from typing import cast from typing import Self @@ -21,6 +22,7 @@ from pydantic_settings import SettingsConfigDict from sqlalchemy import bindparam from sqlalchemy import Connection +from sqlalchemy import DateTime from sqlalchemy import JSON from sqlalchemy import text from sqlalchemy.engine import Engine @@ -31,12 +33,15 @@ from ultimate_memory.model import CostBudget from ultimate_memory.model import CostBudgetStatus from ultimate_memory.model import CostTierSpend +from ultimate_memory.model import DeadLetterReplayResult from ultimate_memory.model import EnqueueOutcome from ultimate_memory.model import EnqueueWork from ultimate_memory.model import LaneRouteError from ultimate_memory.model import PipelineStage from ultimate_memory.model import ProcessingLane +from ultimate_memory.model import QueueRoute from ultimate_memory.model import RecordCall +from ultimate_memory.model import WorkNotDeadLetterError from ultimate_memory.model import WorkNotFoundError from ultimate_memory.model import WorkNotRunningError from ultimate_memory.spine.catalog_contract import lane_is_valid @@ -305,6 +310,80 @@ def wake(self, *, processing_id: UUID) -> None: with self._engine.begin() as connection: connection.execute(_WAKE, {"processing_id": str(processing_id)}) + def replay_dead_letter( + self, + *, + deployment_id: UUID, + processing_id: UUID, + attempt_allowance: int = 1, + lane: ProcessingLane | None = None, + not_before: datetime | None = None, + ) -> DeadLetterReplayResult: + """Reopen exactly one deployment-owned dead letter for explicit replay. + + The transition preserves the attempts already consumed and the complete + previous error. It grants only the requested additional attempt budget, + defaults to due now, and may reroute only to a lane accepted by the + row's immutable stage. Delivery is announced separately after commit. + """ + if attempt_allowance < 1: + raise ValueError("dead-letter replay requires at least one new attempt") + if not_before is not None and ( + not_before.tzinfo is None or not_before.utcoffset() != timedelta(0) + ): + raise ValueError("dead-letter replay not_before must be UTC") + with self._engine.begin() as connection: + existing = ( + connection.execute( + _SELECT_DEAD_LETTER_FOR_REPLAY, + {"deployment_id": deployment_id, "processing_id": processing_id}, + ) + .mappings() + .one_or_none() + ) + if existing is None: + raise WorkNotFoundError( + f"processing row {processing_id} does not exist in deployment " + f"{deployment_id}" + ) + if existing["status"] != "dead_letter": + raise WorkNotDeadLetterError( + f"processing row {processing_id} has status {existing['status']}; " + "only dead-letter rows can be replayed" + ) + stage = PipelineStage(existing["stage"]) + replay_lane = ( + None + if existing["lane"] is None and lane is None + else lane or ProcessingLane(existing["lane"]) + ) + _require_valid_lane(stage=stage, lane=replay_lane) + attempts = int(existing["attempts"]) + if attempts + attempt_allowance > 32_767: + raise ValueError("dead-letter replay attempt budget exceeds smallint") + replayed = ( + connection.execute( + _REPLAY_DEAD_LETTER, + { + "processing_id": processing_id, + "lane": replay_lane, + "attempt_allowance": attempt_allowance, + "not_before": not_before, + }, + ) + .mappings() + .one() + ) + return DeadLetterReplayResult( + processing_id=processing_id, + route=QueueRoute( + deployment_id=deployment_id, stage=stage, lane=replay_lane + ), + not_before=replayed["not_before"], + attempts=int(replayed["attempts"]), + max_attempts=int(replayed["max_attempts"]), + ) + def record_call(self, *, call: RecordCall) -> bool: """Attribute one billed call to the running attempt; idempotent per call key. @@ -612,6 +691,35 @@ def _claimed_work(*, row: RowMapping) -> ClaimedWork: """ ) +_SELECT_DEAD_LETTER_FOR_REPLAY = text( + """ + SELECT status, stage, lane, attempts + FROM processing_state + WHERE deployment_id = :deployment_id + AND processing_id = :processing_id + FOR UPDATE + """ +) + +_REPLAY_DEAD_LETTER = text( + """ + UPDATE processing_state + SET status = 'pending', + lane = :lane, + defer_reason = CASE + WHEN :not_before IS NOT NULL AND :not_before > now() + THEN 'scheduled'::processing_defer_reason + ELSE NULL + END, + not_before = COALESCE(:not_before, now()), + max_attempts = attempts + :attempt_allowance, + finished_at = NULL + WHERE processing_id = :processing_id + AND status = 'dead_letter' + RETURNING attempts, max_attempts, not_before + """ +).bindparams(bindparam("not_before", type_=DateTime(timezone=True))) + _BUDGET_WINDOW_SPEND = text( """ WITH bounds AS ( diff --git a/src/ultimate_memory/surfaces/cli.py b/src/ultimate_memory/surfaces/cli.py index 8b1ac979..92270bbf 100644 --- a/src/ultimate_memory/surfaces/cli.py +++ b/src/ultimate_memory/surfaces/cli.py @@ -1,14 +1,15 @@ """The ``ugm`` CLI: a dependency-light client plus optional local admin commands. Query, ingest, connector management, and MCP all talk to the deployment HTTP -API. ``ugm review`` and ``ugm budget`` import the server extra and connect to -the spine. +API. ``ugm review``, ``ugm budget``, and ``ugm ops`` import the server extra +and connect to the spine. """ from __future__ import annotations import argparse from datetime import datetime +from importlib import import_module import json from pathlib import Path import sys @@ -42,6 +43,8 @@ def main(argv: list[str] | None = None) -> int: return _run_review(args) if args.command == "budget": return _run_budget(args) + if args.command == "ops": + return _run_ops(args) if args.command == "query": return _run_query(args) if args.command == "ingest": @@ -111,6 +114,54 @@ def _run_budget(args: argparse.Namespace) -> int: engine.dispose() +def _run_ops(args: argparse.Namespace) -> int: + """Compose bounded local inspection, one-row replay, or an existing rebuild.""" + try: + from ultimate_memory.model import ProcessingLane + from ultimate_memory.model import WorkLedgerError + + operations_type = import_module( + "ultimate_memory.profiles.selfhost_operations" + ).SelfHostOperations + except ModuleNotFoundError: + print( + "error: ops commands require the 'ultimate-memory[server]' extra", + file=sys.stderr, + ) + return 1 + + operations = operations_type.from_settings() + try: + if args.ops_command == "inspect": + report = operations.inspect(deployment_id=args.deployment) + print(report.model_dump_json()) + return 0 + if args.ops_command == "replay": + replayed = operations.replay( + deployment_id=args.deployment, + processing_id=args.processing_id, + attempt_allowance=args.attempts, + lane=None if args.lane is None else ProcessingLane(args.lane), + not_before=args.not_before, + ) + print(replayed.model_dump_json()) + return 0 + result = operations.rebuild( + plane=args.plane, + deployment_id=args.deployment, + snapshot_root=args.snapshot_root, + workdir=args.workdir, + version=args.version, + ) + print(json.dumps(result, default=str, sort_keys=True)) + return 0 + except (WorkLedgerError, ValueError, OSError) as error: + print(f"error: {error}", file=sys.stderr) + return 1 + finally: + operations.close() + + def _run_query(args: argparse.Namespace) -> int: """Run a query command through the typed remote SDK.""" with MemoryClient.from_settings() as client: @@ -316,6 +367,34 @@ def _build_parser() -> argparse.ArgumentParser: ) inspect.add_argument("--deployment", type=UUID, required=True) + ops = commands.add_parser( + "ops", help="inspect durable state, replay one DLQ row, or rebuild" + ) + ops_commands = ops.add_subparsers(dest="ops_command", required=True) + ops_inspect = ops_commands.add_parser( + "inspect", help="bounded pipeline, DLQ, projection, and currency report" + ) + ops_inspect.add_argument("--deployment", type=UUID, required=True) + replay = ops_commands.add_parser("replay", help="reopen one dead-letter row") + replay.add_argument("processing_id", type=UUID) + replay.add_argument("--deployment", type=UUID, required=True) + replay.add_argument( + "--attempts", + type=int, + default=1, + help="additional handler attempts to grant (default: 1)", + ) + replay.add_argument("--lane", choices=("steady", "backfill")) + replay.add_argument("--not-before", type=datetime.fromisoformat) + rebuild = ops_commands.add_parser( + "rebuild", help="invoke the existing P2 or P3 full-rebuild path" + ) + rebuild.add_argument("--plane", choices=("p2", "p3"), required=True) + rebuild.add_argument("--deployment", type=UUID, required=True) + rebuild.add_argument("--snapshot-root", type=Path, required=True) + rebuild.add_argument("--workdir", type=Path, required=True) + rebuild.add_argument("--version", required=True) + query = commands.add_parser("query", help="query deployment recipes") query_commands = query.add_subparsers(dest="query_command", required=True) query_commands.add_parser("list", help="list the remote recipe tools") diff --git a/src/ultimate_memory/workers/__init__.py b/src/ultimate_memory/workers/__init__.py index 6b234995..c1e7b6ed 100644 --- a/src/ultimate_memory/workers/__init__.py +++ b/src/ultimate_memory/workers/__init__.py @@ -46,6 +46,7 @@ from ultimate_memory.workers.knowledge_writer import KnowledgeWriterError from ultimate_memory.workers.knowledge_writer import KnowledgeWriterSession from ultimate_memory.workers.knowledge_writer import KnowledgeWriterSettings +from ultimate_memory.workers.operations import DeadLetterReplayer from ultimate_memory.workers.p1 import EmbedClaimsHandler from ultimate_memory.workers.p1 import FACT_LABEL_VERSION from ultimate_memory.workers.p1 import LabelFactsHandler @@ -127,6 +128,7 @@ "SnapshotValidationError", "CycleFinalizer", "DeletionService", + "DeadLetterReplayer", "ReconcileHandler", "RECONCILE_VERSION", "StructureHandler", diff --git a/src/ultimate_memory/workers/base.py b/src/ultimate_memory/workers/base.py index 1acd3e5b..0502a478 100644 --- a/src/ultimate_memory/workers/base.py +++ b/src/ultimate_memory/workers/base.py @@ -6,7 +6,10 @@ either retried with backoff or dead-lettered — it never disappears. """ +from datetime import datetime +from datetime import UTC import logging +import time import traceback from typing import Protocol from typing import runtime_checkable @@ -26,9 +29,12 @@ from ultimate_memory.model import QueueRoute from ultimate_memory.model import RecordCall from ultimate_memory.model import RunResultOutcome +from ultimate_memory.model import TelemetryAttribute +from ultimate_memory.model import TelemetryEvent from ultimate_memory.model import UnknownStageHandlerError from ultimate_memory.ports.cost_meter import CostMeterPort from ultimate_memory.ports.queue import TaskQueuePort +from ultimate_memory.ports.telemetry import TelemetryPort from ultimate_memory.spine.work_ledger import WorkLedger _logger = logging.getLogger(__name__) @@ -122,6 +128,7 @@ def __init__( ledger: WorkLedger, registry: HandlerRegistry, queue: TaskQueuePort | None = None, + telemetry: TelemetryPort | None = None, ) -> None: """Bind the runner to its ledger, handler registry, and optional queue port. @@ -131,6 +138,7 @@ def __init__( self._ledger = ledger self._registry = registry self._queue = queue + self._telemetry = telemetry def run_one( self, *, deployment_id: UUID, stage: PipelineStage, lane: ProcessingLane | None @@ -142,6 +150,7 @@ def run_one( `NonRetryableHandlerError` dead-letters immediately, anything else retries with backoff until the attempt limit dead-letters it. """ + started_ns = time.monotonic_ns() handler = self._registry.handler_for(stage=stage) # before any claim: # an unregistered stage must never strand a claimed row as running. claimed = self._ledger.claim_one( @@ -158,6 +167,17 @@ def run_one( ), not_before_snapshot=claimed.resume_at, ) + self._export_event( + event=_worker_event( + deployment_id=deployment_id, + processing_id=claimed.processing_id, + stage=stage, + lane=lane, + attempt=None, + outcome=RunResultOutcome.BUDGET_PARKED, + started_ns=started_ns, + ) + ) return RunResult( processing_id=claimed.processing_id, outcome=RunResultOutcome.BUDGET_PARKED, @@ -167,7 +187,7 @@ def run_one( ) try: outcome = handler.handle(work=claimed, meter=meter) - except NonRetryableHandlerError: + except NonRetryableHandlerError as exception: _logger.exception( "non-retryable failure in stage %s for %s", claimed.stage, @@ -178,11 +198,23 @@ def run_one( error=traceback.format_exc(), retryable=False, ) + self._export_exception( + event=_worker_event( + deployment_id=claimed.deployment_id, + processing_id=claimed.processing_id, + stage=claimed.stage, + lane=claimed.lane, + attempt=claimed.attempt, + outcome=RunResultOutcome.DEAD_LETTERED, + started_ns=started_ns, + ), + exception=exception, + ) return RunResult( processing_id=claimed.processing_id, outcome=RunResultOutcome.DEAD_LETTERED, ) - except Exception: + except Exception as exception: _logger.exception( "retryable failure in stage %s for %s", claimed.stage, @@ -193,6 +225,11 @@ def run_one( error=traceback.format_exc(), retryable=True, ) + result_outcome = ( + RunResultOutcome.RETRY_SCHEDULED + if retry_at is not None + else RunResultOutcome.DEAD_LETTERED + ) if retry_at is not None and self._queue is not None: self._queue.announce( processing_id=claimed.processing_id, @@ -201,15 +238,75 @@ def run_one( ), not_before_snapshot=retry_at, ) + self._export_exception( + event=_worker_event( + deployment_id=claimed.deployment_id, + processing_id=claimed.processing_id, + stage=claimed.stage, + lane=claimed.lane, + attempt=claimed.attempt, + outcome=result_outcome, + started_ns=started_ns, + ), + exception=exception, + ) return RunResult( - processing_id=claimed.processing_id, - outcome=RunResultOutcome.RETRY_SCHEDULED - if retry_at is not None - else RunResultOutcome.DEAD_LETTERED, + processing_id=claimed.processing_id, outcome=result_outcome ) self._ledger.complete( processing_id=claimed.processing_id, follow_up=outcome.follow_up ) + self._export_event( + event=_worker_event( + deployment_id=claimed.deployment_id, + processing_id=claimed.processing_id, + stage=claimed.stage, + lane=claimed.lane, + attempt=claimed.attempt, + outcome=RunResultOutcome.SUCCEEDED, + started_ns=started_ns, + ) + ) return RunResult( processing_id=claimed.processing_id, outcome=RunResultOutcome.SUCCEEDED ) + + def _export_event(self, *, event: TelemetryEvent) -> None: + """Export one completed transition when telemetry is configured.""" + if self._telemetry is not None: + self._telemetry.export_event(event=event) + + def _export_exception( + self, *, event: TelemetryEvent, exception: BaseException + ) -> None: + """Export the original exception object without hiding exporter failure.""" + if self._telemetry is not None: + self._telemetry.export_exception(event=event, exception=exception) + + +def _worker_event( + *, + deployment_id: UUID, + processing_id: UUID, + stage: PipelineStage, + lane: ProcessingLane | None, + attempt: int | None, + outcome: RunResultOutcome, + started_ns: int, +) -> TelemetryEvent: + """Build the one stable provider-neutral worker event vocabulary.""" + return TelemetryEvent( + name="worker.run", + occurred_at=datetime.now(UTC), + attributes=( + TelemetryAttribute(name="deployment_id", value=str(deployment_id)), + TelemetryAttribute(name="processing_id", value=str(processing_id)), + TelemetryAttribute(name="stage", value=stage.value), + TelemetryAttribute(name="lane", value=None if lane is None else lane.value), + TelemetryAttribute(name="attempt", value=attempt), + TelemetryAttribute(name="outcome", value=outcome.value), + TelemetryAttribute( + name="duration_ms", value=(time.monotonic_ns() - started_ns) / 1e6 + ), + ), + ) diff --git a/src/ultimate_memory/workers/operations.py b/src/ultimate_memory/workers/operations.py new file mode 100644 index 00000000..ffcd0dea --- /dev/null +++ b/src/ultimate_memory/workers/operations.py @@ -0,0 +1,42 @@ +"""Thin operational services that compose spine truth with delivery ports.""" + +from datetime import datetime +from uuid import UUID + +from ultimate_memory.model import DeadLetterReplayResult +from ultimate_memory.model import ProcessingLane +from ultimate_memory.ports.queue import TaskQueuePort +from ultimate_memory.spine.work_ledger import WorkLedger + + +class DeadLetterReplayer: + """Replay one row through the ledger, then announce its committed route.""" + + def __init__(self, *, ledger: WorkLedger, queue: TaskQueuePort) -> None: + """Bind authoritative mutation and non-authoritative delivery.""" + self._ledger = ledger + self._queue = queue + + def replay( + self, + *, + deployment_id: UUID, + processing_id: UUID, + attempt_allowance: int = 1, + lane: ProcessingLane | None = None, + not_before: datetime | None = None, + ) -> DeadLetterReplayResult: + """Commit one explicit replay transition, then wake its returned route.""" + replayed = self._ledger.replay_dead_letter( + deployment_id=deployment_id, + processing_id=processing_id, + attempt_allowance=attempt_allowance, + lane=lane, + not_before=not_before, + ) + self._queue.announce( + processing_id=replayed.processing_id, + route_snapshot=replayed.route, + not_before_snapshot=replayed.not_before, + ) + return replayed diff --git a/website/src/app/docs/project-status/page.mdx b/website/src/app/docs/project-status/page.mdx index 5afa3aec..d6bde577 100644 --- a/website/src/app/docs/project-status/page.mdx +++ b/website/src/app/docs/project-status/page.mdx @@ -22,7 +22,7 @@ Ultimate Memory is being built **design-first**: the complete system — require - **Phase 5, complete** — retrieval complete. The full zero-LLM primitive set is implemented (`fuse`, `rerank`, `transcript`, `delta`, `pages_about`, enumerated `aggregate`, and a streaming batch `scan`, alongside resolve/lookup/search/graph); recipes are registry rows whose declared grain is enforced mechanically; and the self-accounting envelope surfaces contradictions, withdrawn support, mixed-grain parts, identity regime, belief horizons, freshness, and typed negatives. API, CLI, and MCP all render the same deployment registry. The consumption skill is guarded by the repeatable S58 cold-agent eval, while the retrieval spike battery records filtered Lance search at 10 million rows and graph pagination at a 100,000-edge hub. The client-first base wheel now contains the typed SDK, remote CLI and MCP, lineage-aware E0 ingest, and typed remote connector-management commands plus their deployment composition port, without server dependencies; `[server]`, `[connectors-watched-directory]`, and `[k]` name the heavier install surfaces. The public `remember-dev`/`remember` rename remains deliberately gated with release engineering. - **Phase 6, complete** — Plane K now includes the deterministic control plane and crash-safe single-committer driver, exact rule routing and staleness, deterministic fact-sheet and agent-written prose bands, planner/reflection decisions with quarantine and adoption, and authored-page frontmatter, citations, watches, review flags, and debounced workflow dispatch. D73 removed the proposed K3 tier: personal or organizational principles are authored K2 content, supported by compiled scope pages and never machine-promoted or rewritten. - **Phase 3, complete** — the evidence lifecycle: documents that change. Watched sources poll as recorded sync cycles (a local-directory watcher ships; revision and content no-ops, debounce, source-deletion detection); an edited document becomes a new version of its lineage and the full structure route (an LLM-proposed section tree normalized by a deterministic snap) re-reads it; unchanged chunks reuse their prior claims, prefixes, and vectors so cost is proportional to the edit (measured hit rate 0.79 on the spike corpus); reconciliation transitions testimony currency on an append-only ledger, recounts evidence by distinct current lineages, closes solely-supported facts when the source withdrew them (at the sync-cycle barrier, so a moved section is a support swap — never a retract flicker) and flags them for review when only the toolchain changed; deletion removes a document's contribution uniformly while keeping its claims as history; and a standing `lifecycle` eval suite guards the cache/ledger/count invariants with planted regression canaries. Those lifecycle events now feed the graph, corpus filesystem, and Plane-K control plane described above. -- **Phase 7, in progress** — operational correctness now includes resumable initial-load and version-bump backfill on the same work ledger, a reproducible PostgreSQL scale battery for the designed partitions, indexes, hubs, and batching invariants, and optional per-deployment/stage/lane cost ceilings. Budget exhaustion parks healthy work durably without consuming an attempt; the local CLI reports current spend, remaining budget, tier attribution, and parked work from the same authoritative rows. Timings remain measurements rather than hosted SLAs, and no dashboard, billing policy, HA topology, or control plane enters the library. +- **Phase 7, in progress** — operational correctness now includes resumable initial-load and version-bump backfill on the same work ledger, a reproducible PostgreSQL scale battery for the designed partitions, indexes, hubs, and batching invariants, and optional per-deployment/stage/lane cost ceilings. Budget exhaustion parks healthy work durably without consuming an attempt; the local CLI reports current spend, remaining budget, tier attribution, and parked work from the same authoritative rows. Typed worker telemetry preserves complete exception chains, while bounded CLI inspection exposes pipeline/DLQ state, poison targets, P2/P3 pointers, and currency-ledger drift; an explicit one-row replay grants only additional attempts, and rebuild drills call the production P2/P3 builders. Timings remain measurements rather than hosted SLAs, and no dashboard, billing policy, HA topology, or control plane enters the library. ## The build order diff --git a/website/src/app/docs/reference/cli/page.mdx b/website/src/app/docs/reference/cli/page.mdx index 5a0990cf..274da01a 100644 --- a/website/src/app/docs/reference/cli/page.mdx +++ b/website/src/app/docs/reference/cli/page.mdx @@ -1,7 +1,7 @@ export const metadata = { title: "CLI Reference", description: - "The client-first ugm command line: query, ingest, manage connectors, inspect budgets, serve MCP, and drive the local review queue.", + "The client-first ugm command line: query, ingest, manage connectors, inspect operations and budgets, serve MCP, and drive the local review queue.", }; # CLI Reference @@ -111,6 +111,58 @@ When spend reaches its ceiling, due work is parked until the aligned window ends without consuming a handler attempt or becoming a failure; the same row then resumes normally. +## `ugm ops` + +Operational commands require the `[server]` extra and `UGM_DATABASE_URL`. They +are direct, local admin tools over the durable PostgreSQL state — not a +dashboard or hosted control plane. + +### `ugm ops inspect` + +```bash +ugm ops inspect --deployment +``` + +Prints one typed JSON report containing route/status counts, DLQ totals and +groups, a poison-target summary, the current P2/P3 snapshot pointers, and a +currency-ledger audit. Variable-size evidence is returned as a bounded sample; +its complete total is reported separately. `UGM_OPERATIONAL_SAMPLE_LIMIT` +sets the shared cap (default 20, maximum 1000). + +### `ugm ops replay` + +```bash +ugm ops replay \ + --deployment \ + --attempts 1 \ + [--lane steady|backfill] \ + [--not-before 2026-07-22T08:00:00+00:00] +``` + +Reopens exactly one row whose current status is `dead_letter`. Attempts already +consumed and the full previous traceback remain intact; `--attempts` grants +only that many additional handler executions. The stage is immutable, an +optional lane must be valid for it, and the committed row is then announced +through the normal delivery port. Replaying a non-DLQ row fails rather than +silently duplicating work. + +### `ugm ops rebuild` + +```bash +ugm ops rebuild \ + --plane p2 \ + --deployment \ + --snapshot-root ./snapshots \ + --workdir ./rebuild-work \ + --version drill-2026-07-21 +``` + +`--plane p2` invokes the production full graph rebuild; `--plane p3` invokes +the production corpus-filesystem builder. Both publish immutable snapshots +through the same validation and pointer-swap path used normally. The command +adds no separate recovery implementation. `--workdir` is consumed by P2 and +accepted uniformly for P3. + ## `ugm review` The review commands require the `[server]` extra. They connect directly to the