diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md index c8989e41b2..964ee4418a 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md @@ -136,6 +136,20 @@ not the supported legacy writer. The absent arm also removes all four shadow producer modules using the qualified CLI startup isolation above. [Retained duties and rollback](../../reference/canonical-todo-completion-update.md#monitor-source-writer-isolation). +Shadow diagnostics now have an independent read-only boundary: retained outbox +disk codecs live in `local_authority_shadow_history.py`, and status/candidate +read transport in `local_authority_shadow_readback.py`. The public status CLI +does not load source producers; drain still uses its existing orchestration. +Original entry identity, malformed evidence, cursor residue and missing-store +behavior remain unchanged. Python retains historical filesystem decoding and +presentation IO; TS retains history proof, delivery and cleanup authority. +Existing producer imports are aliases for supported callers, not duplicate +implementations. Source and normal-wheel File/SQLite recipient checks compare +complete status results and unchanged original bytes with producers present or +absent. This is a historical-reader exit prerequisite, not writer deletion, +outbox disposition, installed cohort adoption or release-default qualification. +See [operator readback and rollback](../../reference/authority-observation-retirement.md#retained-readback-without-source-producers). + Existing Goal migration, two-policy ownership retirement and storage-format upgrade retain separate receipts and exits. Original-receipt recovery does not justify retaining `legacy` as a live policy. Required migration readers remain. diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md index 9ffd1c77eb..b89cfd7dca 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md @@ -109,6 +109,17 @@ wheel 删源检查须保留后来写入及原 successor；晋升后重放保持 producer module。 见[保留义务及回滚](../../reference/canonical-todo-completion-update.md#monitor-source-writer-isolation)。 +shadow 诊断现在有独立只读边界：历史 outbox 磁盘 codec 位于 +`local_authority_shadow_history.py`，status／candidate read transport 位于 +`local_authority_shadow_readback.py`。公开 status CLI 不加载源 producer；drain +仍走现有 orchestration。原 entry 身份、损坏证据、cursor 残留和缺失 store 行为 +保持。Python 保留历史文件解码及展示 IO；TS 保留历史证明、投递与清理权限。 +旧 producer 的已有导入为支持调用方保留 alias，不复制实现。源码及正常 wheel +的 File/SQLite 接收端检查，在 producer 存在／缺席两臂比较完整 status 响应和 +原文件字节。这是历史 reader 的退役前置，不代表 writer 删除、outbox 处置、 +安装 cohort 采用或发布默认资格。 +见[操作读回与回滚](../../reference/authority-observation-retirement.md#retained-readback-without-source-producers)。 + 存量 Goal 迁移、两策略退役和格式升级仍各有独立回执及出口；原回执恢复不能成为 保留 legacy 活跃策略的理由，必要迁移 reader 保留。 diff --git a/docs/reference/authority-observation-retirement.md b/docs/reference/authority-observation-retirement.md index 2ce9668d75..050a928055 100644 --- a/docs/reference/authority-observation-retirement.md +++ b/docs/reference/authority-observation-retirement.md @@ -61,6 +61,24 @@ can restart the old observer if its setting is still present, so clear the setting on installations that must not resume observation after rollback. No format rewrite or destructive data cleanup accompanies retirement. +## Retained readback without source producers + +`authority-shadow status` reads retained outbox entries and candidate history +through independent read-only modules. It remains available when the old source +capture producers are absent. Pending prepared/committed files, cursor residue, +invalid evidence and the original management binding remain visible. Residue +counts are diagnostic hints: they do not authorize cleanup or prove delivery. +A missing candidate stays missing; status never creates a replacement identity, +repairs bytes, drains an outbox or grants a writer permission. + +The retained Python filesystem codecs preserve the existing wire and errors; +candidate proof and drain effects remain in their TypeScript owners. Supported +unpromoted writes and `authority-shadow drain` still need their current producer +and orchestration modules. This separation permits eventual producer retirement +without discarding historical readback, but does not itself authorize deletion +or disposition of pending originals. Code rollback restores the previous import +layout without changing data or restoring retired write authority. + ## Why two matching reads were not a capture guarantee The observer sampled after the writer released its lock. Another writer could diff --git a/loopx/cli_commands/authority_shadow.py b/loopx/cli_commands/authority_shadow.py index 69cfcaf2c9..729ea6719e 100644 --- a/loopx/cli_commands/authority_shadow.py +++ b/loopx/cli_commands/authority_shadow.py @@ -147,17 +147,20 @@ def handle_authority_shadow_command( action = str(getattr(args, "authority_shadow_command", None)) payload: dict[str, object] try: - from ..control_plane.coordination.local_authority_shadow_adapter import ( - drain_local_authority_shadow_outbox, + from ..control_plane.coordination.local_authority_shadow_readback import ( local_authority_shadow_status, ) - from ..control_plane.coordination.local_authority_shadow_outbox import OutboxError + from ..control_plane.coordination.local_authority_shadow_history import OutboxError try: # The same resolver every writer hook uses, so drain and status address # the lineage those hooks wrote. runtime_root = effective_runtime_root(registry_path, runtime_root_arg) if action == "drain": + from ..control_plane.coordination.local_authority_shadow_adapter import ( + drain_local_authority_shadow_outbox, + ) + if args.max_entries < 1: raise ValueError("--max-entries must be at least 1") result = drain_local_authority_shadow_outbox( diff --git a/loopx/control_plane/coordination/local_authority_shadow_adapter.py b/loopx/control_plane/coordination/local_authority_shadow_adapter.py index c8a37b9aa9..9e304babf4 100644 --- a/loopx/control_plane/coordination/local_authority_shadow_adapter.py +++ b/loopx/control_plane/coordination/local_authority_shadow_adapter.py @@ -11,7 +11,6 @@ from pathlib import Path from typing import Any -from ..projects.registry_codec import load_registry from ...paths import effective_runtime_root as effective_runtime_root, resolve_runtime_root from ...registry import find_registry_goal from ..effect_runtime import effect_runtime_result @@ -22,13 +21,15 @@ LOCAL_AUTHORITY_SHADOW_READ_RESULT_SCHEMA, LOCAL_AUTHORITY_SHADOW_TRANSACTION_EVIDENCE_SCHEMA, ) -from .local_authority_shadow_projection import ( - head_digest, -) -from .configuration import resolve_coordination_runtime_shadow_config, local_authority_shadow_summary +from .configuration import resolve_coordination_runtime_shadow_config from .authority_source_capture import todo_partition_projector as todo_partition_projector -from .shadow_management import read_shadow_capture_binding, shadow_management_state_path +from .shadow_management import shadow_management_state_path from .shadow_goal_scope import shadow_goal_scope +from .local_authority_shadow_readback import ( + RETENTION_PRESSURE_BYTES as RETENTION_PRESSURE_BYTES, + read_local_authority_shadow as read_local_authority_shadow, + local_authority_shadow_status as local_authority_shadow_status, +) @@ -51,8 +52,6 @@ INLINE_DRAIN_LOCK_TIMEOUT_SECONDS = 0.25 SHADOW_DRAIN_SCHEMA = "loopx_shadow_drain_v0" SHADOW_EXACT_DRAIN_SCHEMA = "loopx_shadow_drain_v1" -RETENTION_PRESSURE_BYTES = 8 * 1024 * 1024 -_SEED_WRITE_CLASSES = {"seed", "reseed_after_crash_gap"} @dataclass @@ -108,41 +107,6 @@ def to_payload(self) -> dict[str, Any]: return payload -def read_local_authority_shadow( - *, - runtime_root: Path, - goal_id: str, - store_kind: str = "runtime_shadow", - scan_after_cursor: str | None = None, - scan_limit: int = 0, - receipt_operation_id: str | None = None, - read_model: str = "full", -) -> dict[str, Any]: - """Read-only candidate view through the TypeScript store boundary.""" - - result = effect_runtime_result( - "coordination.runtime_shadow.outbox_read", - { - "schema_version": LOCAL_AUTHORITY_SHADOW_READ_REQUEST_SCHEMA, - "runtime_root": str(runtime_root), - "goal_id": goal_id, - "store_kind": store_kind, - "scan_after_cursor": scan_after_cursor, - "scan_limit": scan_limit, - "receipt_operation_id": receipt_operation_id, - "read_model": read_model, - }, - timeout=15.0, - ) - if ( - not isinstance(result, dict) - or result.get("schema_version") != LOCAL_AUTHORITY_SHADOW_READ_RESULT_SCHEMA - or result.get("goal_id") != goal_id - ): - raise RuntimeError("local authority shadow read result is invalid") - return dict(result) - - def _drain_prelude( result: DrainResult, *, @@ -226,130 +190,6 @@ def drain_local_authority_shadow_outbox( return result -class _CandidateMissing(Exception): - """The candidate store directory does not exist yet.""" - - -def _store_bytes(runtime_root: Path, goal_id: str, *, legacy_observation: bool) -> int: - directory = ( - runtime_root / "authority-shadow" / "file" / goal_id - if legacy_observation - else runtime_root / "authority-shadow" / "file-v0" - ) - if not directory.is_dir(): - return 0 - return sum(path.stat().st_size for path in directory.iterdir() if path.is_file()) - - -def local_authority_shadow_status( - *, - registry_path: Path, - runtime_root: Path | None, - goal_id: str, -) -> dict[str, Any]: - """Operator readback: configuration, outbox backlog, and candidate head facts.""" - - registry = load_registry(registry_path) - goal = find_registry_goal(registry, goal_id) - if not isinstance(goal, dict): - raise ValueError(f"goal {goal_id!r} is not registered") - if runtime_root is None: - runtime_root = resolve_runtime_root(registry, None, registry_path=registry_path) - config = local_authority_shadow_summary(goal) - runtime_config = resolve_coordination_runtime_shadow_config(goal) - management = read_shadow_capture_binding(runtime_root, goal_id) - legacy_observation = ( - config.get("configured") is True - and not runtime_config.enabled - and management["status"] == "missing" - ) - backlog = outbox.outbox_summary(runtime_root, goal_id) - candidate: dict[str, Any] - try: - directory = ( - runtime_root / "authority-shadow" / "file" / goal_id - if legacy_observation - else runtime_root / "authority-shadow" / "file-v0" - ) - if not directory.is_dir() or not any(directory.glob("authority-store-*.json")): - # Reading through the store boundary would mint a store identity; - # a status probe must not create candidate lineage. - raise _CandidateMissing - view = read_local_authority_shadow( - runtime_root=runtime_root, - goal_id=goal_id, - store_kind=( - "legacy_observation" if legacy_observation else "runtime_shadow" - ), - ) - head = view.get("head") if isinstance(view.get("head"), dict) else None - candidate = { - "status": view.get("status"), - "reason_code": view.get("reason_code"), - "store_identity": view.get("store_identity"), - "provider_revision": view.get("provider_revision"), - "cursor": view.get("cursor"), - "head_digest": view.get("head_digest"), - "head_schema_version": head.get("schema_version") if head else None, - "partitions": view.get("partitions"), - "codec_agreement": (head_digest(head) == view.get("head_digest")) - if head - else None, - } - except _CandidateMissing: - candidate = { - "status": "missing", - "reason_code": None, - "store_identity": None, - "provider_revision": None, - "cursor": None, - "head_digest": None, - "head_schema_version": None, - "partitions": None, - "codec_agreement": None, - } - except Exception: - candidate = { - "status": "unavailable", - "reason_code": "shadow_read_failed", - "store_identity": None, - "provider_revision": None, - "cursor": None, - "head_digest": None, - "head_schema_version": None, - "partitions": None, - "codec_agreement": None, - } - candidate["store_kind"] = "legacy_observation" if legacy_observation else "runtime_shadow" - candidate["historical_only"] = legacy_observation - try: - store_bytes = _store_bytes( - runtime_root, goal_id, legacy_observation=legacy_observation - ) - storage_error = None - except OSError: - store_bytes = None - storage_error = "shadow_store_unavailable" - return { - "ok": all(item["invalid"] is None for item in backlog.values()) - and storage_error is None - and management["status"] != "hold", - "action": "status", - "goal_id": goal_id, - "config": config, - "runtime_config": asdict(runtime_config), - "management": management, - "storage_error": storage_error, - "runtime_root_digest": outbox.runtime_root_digest(runtime_root), - "outbox": backlog, - "candidate": candidate, - "store_bytes": store_bytes, - "retention_pressure": store_bytes > RETENTION_PRESSURE_BYTES - if store_bytes is not None - else None, - } - - def capture_evidence( *, goal_id: str, diff --git a/loopx/control_plane/coordination/local_authority_shadow_history.py b/loopx/control_plane/coordination/local_authority_shadow_history.py new file mode 100644 index 0000000000..73f591a1f6 --- /dev/null +++ b/loopx/control_plane/coordination/local_authority_shadow_history.py @@ -0,0 +1,492 @@ +"""Read-only historical outbox codecs and filesystem inventory. + +These retained Python IO codecs preserve legacy diagnostic behavior. They do +not allocate sequences, write markers, reclaim residue or authorize delivery; +TypeScript owns candidate history proof and receipt-proven drain effects. +""" +from __future__ import annotations + +import json +import math +import os +import re +from collections.abc import Mapping +from dataclasses import dataclass +from datetime import datetime, timedelta +from pathlib import Path +from typing import Any + +from ..content_digest import ENVELOPED_SHA256_PATTERN +from .local_authority_shadow_projection import PARTITIONS, sha256_digest, text_digest +from .coordination_state_contract_generated import ( + LOCAL_AUTHORITY_SHADOW_DRAIN_CURSOR_SCHEMA, + LOCAL_AUTHORITY_SHADOW_OUTBOX_COMMIT_SCHEMA, + LOCAL_AUTHORITY_SHADOW_OUTBOX_ENTRY_SCHEMA, +) + +OUTBOX_ENTRY_SCHEMA = LOCAL_AUTHORITY_SHADOW_OUTBOX_ENTRY_SCHEMA +OUTBOX_COMMIT_SCHEMA = LOCAL_AUTHORITY_SHADOW_OUTBOX_COMMIT_SCHEMA +DRAIN_CURSOR_SCHEMA = LOCAL_AUTHORITY_SHADOW_DRAIN_CURSOR_SCHEMA +SOURCE_MARKDOWN = "markdown_active_state" +SOURCE_STATE_EVENT_LOG = "state_event_log" +SOURCE_TASK_LEASE = "task_lease_record" +WRITER_RUNTIME_PYTHON = "python" +WRITER_RUNTIME_TYPESCRIPT = "typescript" +ENTRY_ID_PREFIX = "local-shadow-tx-" +_ENTRY_FILE = re.compile( + r"^(?P\d{10})-(?Plocal-shadow-tx-[0-9a-f]{64})\.(?Pprepared|committed)\.json$" +) + + +class OutboxError(RuntimeError): + """Typed outbox failure; never escapes into a primary write.""" + + def __init__(self, reason_code: str, message: str) -> None: + super().__init__(message) + self.reason_code = reason_code + +def outbox_root(runtime_root: Path, goal_id: str) -> Path: + return runtime_root / "authority-shadow" / "outbox" / goal_id + + +def partition_directory(runtime_root: Path, goal_id: str, partition: str) -> Path: + if partition not in PARTITIONS: + raise OutboxError( + "invalid_partition", f"unknown outbox partition {partition!r}" + ) + return outbox_root(runtime_root, goal_id) / partition + +def lease_directory(runtime_root: Path, goal_id: str) -> Path: + return runtime_root / "goals" / goal_id / "task-leases" + + +def entry_identity( + *, + goal_id: str, + partition: str, + seq: int, + source_ref: str, + capture_lineage_id: str, + source_root_digest: str, +) -> str: + """Bind the entry id to the exact primary bytes (or event) it records.""" + + return ENTRY_ID_PREFIX + sha256_digest( + { + "goal_id": goal_id, + "partition": partition, + "seq": seq, + "source_ref": source_ref, + "capture_lineage_id": capture_lineage_id, + "source_root_digest": source_root_digest, + } + ).removeprefix("sha256:") + + +def entry_file_name(seq: int, entry_id: str, phase: str) -> str: + return f"{seq:010d}-{entry_id}.{phase}.json" + +def _as_object(value: object) -> dict[str, Any]: + """Narrow an untyped JSON value to an object; anything else is empty.""" + + return dict(value) if isinstance(value, Mapping) else {} + + +def _load_json(path: Path) -> dict[str, Any]: + try: + raw = json.loads(path.read_text(encoding="utf-8")) + except (ValueError, UnicodeError) as error: + raise OutboxError( + "outbox_file_invalid", f"{path.name} is not valid JSON" + ) from error + except FileNotFoundError: + raise + except OSError as error: + raise OutboxError( + "outbox_file_unavailable", f"{path.name} cannot be read" + ) from error + if not isinstance(raw, dict): + raise OutboxError("outbox_file_invalid", f"{path.name} is not a JSON object") + return raw + + +@dataclass(frozen=True, slots=True) +class OutboxEntry: + """One two-phase entry as found on disk.""" + + partition: str + seq: int + entry_id: str + prepared_path: Path + committed_path: Path | None + prepared: dict[str, Any] + committed: dict[str, Any] | None + + @property + def is_committed(self) -> bool: + return self.committed is not None + + @property + def source_ref(self) -> str: + return str(record_source_ref(self.prepared)) + + def projection(self) -> dict[str, Any] | None: + """The partition projection recorded for this entry, if any.""" + + for record in (self.committed, self.prepared): + if isinstance(record, dict) and isinstance(record.get("projection"), dict): + return dict(record["projection"]) + return None + + def recorded_partition_digest(self) -> str | None: + for record in (self.committed, self.prepared): + if isinstance(record, dict) and isinstance( + record.get("partition_digest"), str + ): + return str(record["partition_digest"]) + return None + + +_EntryKey = tuple[int, str] + + +def _index_entry_files( + directory: Path, +) -> tuple[dict[_EntryKey, Path], dict[_EntryKey, Path]]: + """Map ``(seq, entry_id)`` to the prepared and committed files present.""" + + prepared: dict[_EntryKey, Path] = {} + committed: dict[_EntryKey, Path] = {} + try: + paths = list(directory.iterdir()) + except FileNotFoundError: + return prepared, committed + except OSError as error: + raise OutboxError( + "outbox_file_unavailable", "outbox inventory cannot be read" + ) from error + for path in paths: + if path.name == "drain-cursor.json" and not path.is_symlink(): + continue + match = _ENTRY_FILE.fullmatch(path.name) + if match is None or path.is_symlink() or not path.is_file(): + raise OutboxError( + "outbox_file_invalid", "outbox contains unclassified evidence" + ) + if int(match.group("seq")) < 1: + raise OutboxError( + "outbox_file_invalid", "outbox sequence is outside its range" + ) + key = (int(match.group("seq")), match.group("entry_id")) + target = prepared if match.group("phase") == "prepared" else committed + target[key] = path + return prepared, committed + + +_WRITER_RUNTIMES = frozenset({WRITER_RUNTIME_PYTHON, WRITER_RUNTIME_TYPESCRIPT}) +_SOURCE_KINDS = frozenset({SOURCE_MARKDOWN, SOURCE_STATE_EVENT_LOG, SOURCE_TASK_LEASE}) + + +def _load_prepared_record( + path: Path, + *, + directory: Path, + seq: int, + entry_id: str, +) -> dict[str, Any]: + """Load one prepared record and prove it binds this directory and entry id. + + The file name, the directory (``outbox//``), the record's + own goal/partition fields, and the entry id recomputed from the record's + source reference must all agree; a record that was copied, edited, or + written for another goal fails closed before it can reach the candidate. + """ + + record = _load_json(path) + writer = _as_object(record.get("writer")) + source = _as_object(record.get("source")) + expected_goal = directory.parent.name + expected_partition = directory.name + source_ref = record_source_ref(record) + root_digest = record.get("source_root_digest") + lineage_id = record.get("capture_lineage_id") + bound = ( + record.get("schema_version") == OUTBOX_ENTRY_SCHEMA + and record.get("entry_id") == entry_id + and record.get("seq") == seq + and type(record.get("seq")) is int + and 1 <= seq <= MAX_OUTBOX_SEQUENCE + and record.get("goal_id") == expected_goal + and record.get("partition") == expected_partition + and writer.get("runtime") in _WRITER_RUNTIMES + and isinstance(writer.get("write_class"), str) + and bool(writer.get("write_class")) + and source.get("kind") in _SOURCE_KINDS + and isinstance(root_digest, str) + and ENVELOPED_SHA256_PATTERN.match(root_digest) is not None + and isinstance(lineage_id, str) + and bool(lineage_id) + and source_ref is not None + and entry_identity( + goal_id=expected_goal, + partition=expected_partition, + seq=seq, + source_ref=source_ref, + capture_lineage_id=lineage_id, + source_root_digest=root_digest, + ) + == entry_id + ) + if not bound: + raise OutboxError( + "outbox_file_invalid", + f"{path.name} does not bind its directory, identity, and source reference", + ) + return record + + +def _load_committed_record( + path: Path | None, *, entry_id: str, capture_lineage_id: str | None = None +) -> dict[str, Any] | None: + if path is None: + return None + record = _load_json(path) + if ( + record.get("schema_version") != OUTBOX_COMMIT_SCHEMA + or record.get("entry_id") != entry_id + or not isinstance(record.get("capture_lineage_id"), str) + or not record["capture_lineage_id"] + or ( + capture_lineage_id is not None + and record["capture_lineage_id"] != capture_lineage_id + ) + ): + raise OutboxError( + "outbox_file_invalid", f"{path.name} does not match its entry" + ) + return record + + +def _retired_watermark(directory: Path) -> int: + cursor = read_cursor(directory) + return int(cursor.get("last_seq") or 0) if cursor is not None else 0 + + +def retired_residue(directory: Path) -> list[Path]: + """Entry files at or below the durable cursor. + + The cursor is written before an entry's files are unlinked, so anything it + covers is already settled in the candidate store. A crash between the + cursor write and the unlinks, or between the two unlinks, leaves these + files behind. This is a diagnostic count only: every file still requires + an exact receipt proof before it can be reclaimed. + """ + + if not directory.is_dir(): + return [] + watermark = _retired_watermark(directory) + prepared, committed = _index_entry_files(directory) + residue = [ + path + for key, path in [*prepared.items(), *committed.items()] + if key[0] <= watermark + ] + return sorted(residue) + + +def list_entries( + directory: Path, *, allow_committed_only: bool = False +) -> list[OutboxEntry]: + """All entries, including unverified residue, independent of cursor hints.""" + + prepared, committed = _index_entry_files(directory) + identities: dict[int, str] = {} + for seq, entry_id in [*prepared, *committed]: + if seq in identities and identities[seq] != entry_id: + raise OutboxError( + "outbox_file_invalid", "sequence has multiple entry identities" + ) + identities[seq] = entry_id + orphan_markers = sorted(set(committed) - set(prepared)) + if orphan_markers and not allow_committed_only: + seq, entry_id = orphan_markers[0] + raise OutboxError( + "outbox_file_invalid", + f"committed marker without prepared entry: {entry_file_name(seq, entry_id, 'committed')}", + ) + entries: list[OutboxEntry] = [] + for seq, entry_id in orphan_markers: + marker_path = committed[(seq, entry_id)] + entries.append( + OutboxEntry( + partition=directory.name, + seq=seq, + entry_id=entry_id, + prepared_path=directory / entry_file_name(seq, entry_id, "prepared"), + committed_path=marker_path, + prepared={}, + committed=_load_committed_record(marker_path, entry_id=entry_id), + ) + ) + for seq, entry_id in sorted(prepared): + key = (seq, entry_id) + prepared_record = _load_prepared_record( + prepared[key], directory=directory, seq=seq, entry_id=entry_id + ) + entries.append( + OutboxEntry( + partition=str(prepared_record.get("partition") or directory.name), + seq=seq, + entry_id=entry_id, + prepared_path=prepared[key], + committed_path=committed.get(key), + prepared=prepared_record, + committed=_load_committed_record( + committed.get(key), + entry_id=entry_id, + capture_lineage_id=prepared_record["capture_lineage_id"], + ), + ) + ) + return sorted(entries, key=lambda entry: entry.seq) + + +def cursor_path(directory: Path) -> Path: + return directory / "drain-cursor.json" + + +def read_cursor(directory: Path) -> dict[str, Any] | None: + path = cursor_path(directory) + if path.is_symlink(): + raise OutboxError( + "outbox_file_invalid", "cursor must belong to its own partition" + ) + try: + record = _load_json(path) + except FileNotFoundError: + return None + return decode_cursor(record, partition=directory.name) + + +_CURSOR_FIELDS = frozenset( + { + "schema_version", + "partition", + "last_seq", + "last_entry_id", + "last_partition_digest", + "last_cursor", + "last_provider_revision", + "updated_at", + } +) +MAX_OUTBOX_SEQUENCE = 9_999_999_999 + + +def decode_cursor(value: object, *, partition: str) -> dict[str, Any]: + """Decode the shared wire cursor. It is a hint, never a delivery receipt.""" + + def invalid() -> OutboxError: + return OutboxError("outbox_file_invalid", "drain cursor binding is invalid") + + if not isinstance(value, dict) or set(value) != _CURSOR_FIELDS: + raise invalid() + seq = value.get("last_seq") + if isinstance(seq, bool) or not isinstance(seq, (int, float)): + raise invalid() + if not 1 <= seq <= MAX_OUTBOX_SEQUENCE or not math.isfinite(seq) or seq != int(seq): + raise invalid() + entry_id = value.get("last_entry_id") + digest = value.get("last_partition_digest") + if ( + value.get("schema_version") != DRAIN_CURSOR_SCHEMA + or partition not in PARTITIONS + or value.get("partition") != partition + or not isinstance(entry_id, str) + or re.fullmatch(r"local-shadow-tx-[0-9a-f]{64}", entry_id) is None + or ( + digest is not None + and ( + not isinstance(digest, str) + or ENVELOPED_SHA256_PATTERN.fullmatch(digest) is None + ) + ) + or any( + not isinstance(value.get(key), str) or not value[key].strip() + for key in ("last_cursor", "last_provider_revision") + ) + ): + raise invalid() + timestamp = value.get("updated_at") + if ( + not isinstance(timestamp, str) + or re.fullmatch( + r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d{1,6})?(?:Z|\+00:00)", timestamp + ) + is None + ): + raise invalid() + try: + parsed = datetime.fromisoformat(timestamp.replace("Z", "+00:00")) + except ValueError as error: + raise invalid() from error + if parsed.utcoffset() != timedelta(0): + raise invalid() + return {**value, "last_seq": int(seq)} + +def runtime_root_digest(runtime_root: Path) -> str: + """Digest of the absolute, dot-normalized root as this process spells it. + + Diagnostic only (drain evidence). Outbox entries and receipts carry the + active binding's ``source_root_digest`` instead, which the TypeScript owner + derives from the resolved root, so a root reached through a symlink hashes + differently from this lexical spelling (#4892). + """ + + return text_digest(os.path.abspath(str(runtime_root))) + + +def record_source_ref(record: Mapping[str, Any]) -> str | None: + """The source reference an entry id binds: bytes digest, event id, or seed digest.""" + + source = _as_object(record.get("source")) + bytes_digest = source.get("bytes_digest") + if isinstance(bytes_digest, str) and bytes_digest: + return bytes_digest + event_id = source.get("event_id") + if isinstance(event_id, str) and event_id: + return f"event:{event_id}" + digest = record.get("partition_digest") + if isinstance(digest, str) and digest: + return f"seed:{digest}" + return None + +def outbox_summary(runtime_root: Path, goal_id: str) -> dict[str, Any]: + """Counts per partition for operator readback; never raises on an empty outbox.""" + + summary: dict[str, Any] = {} + for partition in PARTITIONS: + directory = partition_directory(runtime_root, goal_id, partition) + try: + entries = list_entries(directory) + cursor = read_cursor(directory) + residue_count = len(retired_residue(directory)) + invalid: str | None = None + except OutboxError as error: + entries, cursor, invalid = [], None, error.reason_code + residue_count = 0 + except OSError: + entries, cursor, invalid = [], None, "outbox_file_unavailable" + residue_count = 0 + summary[partition] = { + "committed_pending": sum(1 for entry in entries if entry.is_committed), + "prepared_only": sum(1 for entry in entries if not entry.is_committed), + "retired_residue": residue_count, + "next_seq": ( + max((entry.seq for entry in entries), default=0) if entries else 0 + ), + "cursor_last_seq": int(cursor.get("last_seq") or 0) if cursor else None, + "cursor_last_entry_id": cursor.get("last_entry_id") if cursor else None, + "invalid": invalid, + } + return summary diff --git a/loopx/control_plane/coordination/local_authority_shadow_outbox.py b/loopx/control_plane/coordination/local_authority_shadow_outbox.py index 99ea0f9132..b6b2297bdf 100644 --- a/loopx/control_plane/coordination/local_authority_shadow_outbox.py +++ b/loopx/control_plane/coordination/local_authority_shadow_outbox.py @@ -12,66 +12,67 @@ active-lineage write before changing primary bytes. Marker failures preserve prepared evidence; candidate delivery happens after releasing primary locks. The native drain owns cursor publication and receipt-proven cleanup; Python -retains source capture and read-only historical inventory/decoding. +retains source capture; historical inventory/decoding has an independent IO owner. """ from __future__ import annotations import json -import math import os -import re import uuid from collections.abc import Callable, Mapping from dataclasses import dataclass -from datetime import datetime, timedelta, timezone +from datetime import datetime, timezone from pathlib import Path from typing import Any from .local_authority_shadow_projection import ( - PARTITIONS, TODO_PARTITION, canonical_value, partition_digest, - sha256_digest, text_digest, ) from .coordination_state_contract_generated import ( - LOCAL_AUTHORITY_SHADOW_DRAIN_CURSOR_SCHEMA, - LOCAL_AUTHORITY_SHADOW_OUTBOX_COMMIT_SCHEMA, - LOCAL_AUTHORITY_SHADOW_OUTBOX_ENTRY_SCHEMA, LOCAL_AUTHORITY_SHADOW_READ_REQUEST_SCHEMA, LOCAL_AUTHORITY_SHADOW_READ_RESULT_SCHEMA, ) -from ..content_digest import ENVELOPED_SHA256_PATTERN from .shadow_management import ( read_shadow_capture_binding, shadow_maintenance_lock_target, ) -OUTBOX_ENTRY_SCHEMA = LOCAL_AUTHORITY_SHADOW_OUTBOX_ENTRY_SCHEMA -OUTBOX_COMMIT_SCHEMA = LOCAL_AUTHORITY_SHADOW_OUTBOX_COMMIT_SCHEMA -DRAIN_CURSOR_SCHEMA = LOCAL_AUTHORITY_SHADOW_DRAIN_CURSOR_SCHEMA -SOURCE_MARKDOWN = "markdown_active_state" -SOURCE_STATE_EVENT_LOG = "state_event_log" -SOURCE_TASK_LEASE = "task_lease_record" -WRITER_RUNTIME_PYTHON = "python" -WRITER_RUNTIME_TYPESCRIPT = "typescript" -ENTRY_ID_PREFIX = "local-shadow-tx-" -_ENTRY_FILE = re.compile( - r"^(?P\d{10})-(?Plocal-shadow-tx-[0-9a-f]{64})\.(?Pprepared|committed)\.json$" +# Retained imports preserve existing writer/tool callers; the definitions and +# read-only CLI no longer depend on this producer. +from .local_authority_shadow_history import ( + DRAIN_CURSOR_SCHEMA as DRAIN_CURSOR_SCHEMA, + OUTBOX_COMMIT_SCHEMA as OUTBOX_COMMIT_SCHEMA, + OUTBOX_ENTRY_SCHEMA as OUTBOX_ENTRY_SCHEMA, + SOURCE_MARKDOWN as SOURCE_MARKDOWN, + SOURCE_STATE_EVENT_LOG as SOURCE_STATE_EVENT_LOG, + SOURCE_TASK_LEASE as SOURCE_TASK_LEASE, + WRITER_RUNTIME_PYTHON as WRITER_RUNTIME_PYTHON, + WRITER_RUNTIME_TYPESCRIPT as WRITER_RUNTIME_TYPESCRIPT, + ENTRY_ID_PREFIX as ENTRY_ID_PREFIX, + MAX_OUTBOX_SEQUENCE as MAX_OUTBOX_SEQUENCE, + OutboxEntry as OutboxEntry, + OutboxError as OutboxError, + outbox_root as outbox_root, + partition_directory as partition_directory, + lease_directory as lease_directory, + entry_identity as entry_identity, + entry_file_name as entry_file_name, + list_entries as list_entries, + cursor_path as cursor_path, + read_cursor as read_cursor, + decode_cursor as decode_cursor, + retired_residue as retired_residue, + record_source_ref as record_source_ref, + runtime_root_digest as runtime_root_digest, + outbox_summary as outbox_summary, + _ENTRY_FILE, ) - -class OutboxError(RuntimeError): - """Typed outbox failure; never escapes into a primary write.""" - - def __init__(self, reason_code: str, message: str) -> None: - super().__init__(message) - self.reason_code = reason_code - - TodoPartitionProjector = Callable[[str], dict[str, Any]] """Active-state Markdown text -> ``{"handoff_mode": str, "todos": [compact...]}``.""" @@ -80,53 +81,10 @@ def utc_now_text() -> str: return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%fZ") -def outbox_root(runtime_root: Path, goal_id: str) -> Path: - return runtime_root / "authority-shadow" / "outbox" / goal_id - - -def partition_directory(runtime_root: Path, goal_id: str, partition: str) -> Path: - if partition not in PARTITIONS: - raise OutboxError( - "invalid_partition", f"unknown outbox partition {partition!r}" - ) - return outbox_root(runtime_root, goal_id) / partition - - def drain_lock_target(runtime_root: Path, goal_id: str) -> Path: return shadow_maintenance_lock_target(runtime_root, goal_id) -def lease_directory(runtime_root: Path, goal_id: str) -> Path: - return runtime_root / "goals" / goal_id / "task-leases" - - -def entry_identity( - *, - goal_id: str, - partition: str, - seq: int, - source_ref: str, - capture_lineage_id: str, - source_root_digest: str, -) -> str: - """Bind the entry id to the exact primary bytes (or event) it records.""" - - return ENTRY_ID_PREFIX + sha256_digest( - { - "goal_id": goal_id, - "partition": partition, - "seq": seq, - "source_ref": source_ref, - "capture_lineage_id": capture_lineage_id, - "source_root_digest": source_root_digest, - } - ).removeprefix("sha256:") - - -def entry_file_name(seq: int, entry_id: str, phase: str) -> str: - return f"{seq:010d}-{entry_id}.{phase}.json" - - def _fsync_directory(directory: Path) -> None: if os.name == "nt": return @@ -162,355 +120,6 @@ def durable_write_json(path: Path, payload: Mapping[str, Any]) -> None: temporary.unlink(missing_ok=True) -def _as_object(value: object) -> dict[str, Any]: - """Narrow an untyped JSON value to an object; anything else is empty.""" - - return dict(value) if isinstance(value, Mapping) else {} - - -def _load_json(path: Path) -> dict[str, Any]: - try: - raw = json.loads(path.read_text(encoding="utf-8")) - except (ValueError, UnicodeError) as error: - raise OutboxError( - "outbox_file_invalid", f"{path.name} is not valid JSON" - ) from error - except FileNotFoundError: - raise - except OSError as error: - raise OutboxError( - "outbox_file_unavailable", f"{path.name} cannot be read" - ) from error - if not isinstance(raw, dict): - raise OutboxError("outbox_file_invalid", f"{path.name} is not a JSON object") - return raw - - -@dataclass(frozen=True, slots=True) -class OutboxEntry: - """One two-phase entry as found on disk.""" - - partition: str - seq: int - entry_id: str - prepared_path: Path - committed_path: Path | None - prepared: dict[str, Any] - committed: dict[str, Any] | None - - @property - def is_committed(self) -> bool: - return self.committed is not None - - @property - def source_ref(self) -> str: - return str(record_source_ref(self.prepared)) - - def projection(self) -> dict[str, Any] | None: - """The partition projection recorded for this entry, if any.""" - - for record in (self.committed, self.prepared): - if isinstance(record, dict) and isinstance(record.get("projection"), dict): - return dict(record["projection"]) - return None - - def recorded_partition_digest(self) -> str | None: - for record in (self.committed, self.prepared): - if isinstance(record, dict) and isinstance( - record.get("partition_digest"), str - ): - return str(record["partition_digest"]) - return None - - -_EntryKey = tuple[int, str] - - -def _index_entry_files( - directory: Path, -) -> tuple[dict[_EntryKey, Path], dict[_EntryKey, Path]]: - """Map ``(seq, entry_id)`` to the prepared and committed files present.""" - - prepared: dict[_EntryKey, Path] = {} - committed: dict[_EntryKey, Path] = {} - try: - paths = list(directory.iterdir()) - except FileNotFoundError: - return prepared, committed - except OSError as error: - raise OutboxError( - "outbox_file_unavailable", "outbox inventory cannot be read" - ) from error - for path in paths: - if path.name == "drain-cursor.json" and not path.is_symlink(): - continue - match = _ENTRY_FILE.fullmatch(path.name) - if match is None or path.is_symlink() or not path.is_file(): - raise OutboxError( - "outbox_file_invalid", "outbox contains unclassified evidence" - ) - if int(match.group("seq")) < 1: - raise OutboxError( - "outbox_file_invalid", "outbox sequence is outside its range" - ) - key = (int(match.group("seq")), match.group("entry_id")) - target = prepared if match.group("phase") == "prepared" else committed - target[key] = path - return prepared, committed - - -_WRITER_RUNTIMES = frozenset({WRITER_RUNTIME_PYTHON, WRITER_RUNTIME_TYPESCRIPT}) -_SOURCE_KINDS = frozenset({SOURCE_MARKDOWN, SOURCE_STATE_EVENT_LOG, SOURCE_TASK_LEASE}) - - -def _load_prepared_record( - path: Path, - *, - directory: Path, - seq: int, - entry_id: str, -) -> dict[str, Any]: - """Load one prepared record and prove it binds this directory and entry id. - - The file name, the directory (``outbox//``), the record's - own goal/partition fields, and the entry id recomputed from the record's - source reference must all agree; a record that was copied, edited, or - written for another goal fails closed before it can reach the candidate. - """ - - record = _load_json(path) - writer = _as_object(record.get("writer")) - source = _as_object(record.get("source")) - expected_goal = directory.parent.name - expected_partition = directory.name - source_ref = record_source_ref(record) - root_digest = record.get("source_root_digest") - lineage_id = record.get("capture_lineage_id") - bound = ( - record.get("schema_version") == OUTBOX_ENTRY_SCHEMA - and record.get("entry_id") == entry_id - and record.get("seq") == seq - and type(record.get("seq")) is int - and 1 <= seq <= MAX_OUTBOX_SEQUENCE - and record.get("goal_id") == expected_goal - and record.get("partition") == expected_partition - and writer.get("runtime") in _WRITER_RUNTIMES - and isinstance(writer.get("write_class"), str) - and bool(writer.get("write_class")) - and source.get("kind") in _SOURCE_KINDS - and isinstance(root_digest, str) - and ENVELOPED_SHA256_PATTERN.match(root_digest) is not None - and isinstance(lineage_id, str) - and bool(lineage_id) - and source_ref is not None - and entry_identity( - goal_id=expected_goal, - partition=expected_partition, - seq=seq, - source_ref=source_ref, - capture_lineage_id=lineage_id, - source_root_digest=root_digest, - ) - == entry_id - ) - if not bound: - raise OutboxError( - "outbox_file_invalid", - f"{path.name} does not bind its directory, identity, and source reference", - ) - return record - - -def _load_committed_record( - path: Path | None, *, entry_id: str, capture_lineage_id: str | None = None -) -> dict[str, Any] | None: - if path is None: - return None - record = _load_json(path) - if ( - record.get("schema_version") != OUTBOX_COMMIT_SCHEMA - or record.get("entry_id") != entry_id - or not isinstance(record.get("capture_lineage_id"), str) - or not record["capture_lineage_id"] - or ( - capture_lineage_id is not None - and record["capture_lineage_id"] != capture_lineage_id - ) - ): - raise OutboxError( - "outbox_file_invalid", f"{path.name} does not match its entry" - ) - return record - - -def _retired_watermark(directory: Path) -> int: - cursor = read_cursor(directory) - return int(cursor.get("last_seq") or 0) if cursor is not None else 0 - - -def retired_residue(directory: Path) -> list[Path]: - """Entry files at or below the durable cursor. - - The cursor is written before an entry's files are unlinked, so anything it - covers is already settled in the candidate store. A crash between the - cursor write and the unlinks, or between the two unlinks, leaves these - files behind. This is a diagnostic count only: every file still requires - an exact receipt proof before it can be reclaimed. - """ - - if not directory.is_dir(): - return [] - watermark = _retired_watermark(directory) - prepared, committed = _index_entry_files(directory) - residue = [ - path - for key, path in [*prepared.items(), *committed.items()] - if key[0] <= watermark - ] - return sorted(residue) - - -def list_entries( - directory: Path, *, allow_committed_only: bool = False -) -> list[OutboxEntry]: - """All entries, including unverified residue, independent of cursor hints.""" - - prepared, committed = _index_entry_files(directory) - identities: dict[int, str] = {} - for seq, entry_id in [*prepared, *committed]: - if seq in identities and identities[seq] != entry_id: - raise OutboxError( - "outbox_file_invalid", "sequence has multiple entry identities" - ) - identities[seq] = entry_id - orphan_markers = sorted(set(committed) - set(prepared)) - if orphan_markers and not allow_committed_only: - seq, entry_id = orphan_markers[0] - raise OutboxError( - "outbox_file_invalid", - f"committed marker without prepared entry: {entry_file_name(seq, entry_id, 'committed')}", - ) - entries: list[OutboxEntry] = [] - for seq, entry_id in orphan_markers: - marker_path = committed[(seq, entry_id)] - entries.append( - OutboxEntry( - partition=directory.name, - seq=seq, - entry_id=entry_id, - prepared_path=directory / entry_file_name(seq, entry_id, "prepared"), - committed_path=marker_path, - prepared={}, - committed=_load_committed_record(marker_path, entry_id=entry_id), - ) - ) - for seq, entry_id in sorted(prepared): - key = (seq, entry_id) - prepared_record = _load_prepared_record( - prepared[key], directory=directory, seq=seq, entry_id=entry_id - ) - entries.append( - OutboxEntry( - partition=str(prepared_record.get("partition") or directory.name), - seq=seq, - entry_id=entry_id, - prepared_path=prepared[key], - committed_path=committed.get(key), - prepared=prepared_record, - committed=_load_committed_record( - committed.get(key), - entry_id=entry_id, - capture_lineage_id=prepared_record["capture_lineage_id"], - ), - ) - ) - return sorted(entries, key=lambda entry: entry.seq) - - -def cursor_path(directory: Path) -> Path: - return directory / "drain-cursor.json" - - -def read_cursor(directory: Path) -> dict[str, Any] | None: - path = cursor_path(directory) - if path.is_symlink(): - raise OutboxError( - "outbox_file_invalid", "cursor must belong to its own partition" - ) - try: - record = _load_json(path) - except FileNotFoundError: - return None - return decode_cursor(record, partition=directory.name) - - -_CURSOR_FIELDS = frozenset( - { - "schema_version", - "partition", - "last_seq", - "last_entry_id", - "last_partition_digest", - "last_cursor", - "last_provider_revision", - "updated_at", - } -) -MAX_OUTBOX_SEQUENCE = 9_999_999_999 - - -def decode_cursor(value: object, *, partition: str) -> dict[str, Any]: - """Decode the shared wire cursor. It is a hint, never a delivery receipt.""" - - def invalid() -> OutboxError: - return OutboxError("outbox_file_invalid", "drain cursor binding is invalid") - - if not isinstance(value, dict) or set(value) != _CURSOR_FIELDS: - raise invalid() - seq = value.get("last_seq") - if isinstance(seq, bool) or not isinstance(seq, (int, float)): - raise invalid() - if not 1 <= seq <= MAX_OUTBOX_SEQUENCE or not math.isfinite(seq) or seq != int(seq): - raise invalid() - entry_id = value.get("last_entry_id") - digest = value.get("last_partition_digest") - if ( - value.get("schema_version") != DRAIN_CURSOR_SCHEMA - or partition not in PARTITIONS - or value.get("partition") != partition - or not isinstance(entry_id, str) - or re.fullmatch(r"local-shadow-tx-[0-9a-f]{64}", entry_id) is None - or ( - digest is not None - and ( - not isinstance(digest, str) - or ENVELOPED_SHA256_PATTERN.fullmatch(digest) is None - ) - ) - or any( - not isinstance(value.get(key), str) or not value[key].strip() - for key in ("last_cursor", "last_provider_revision") - ) - ): - raise invalid() - timestamp = value.get("updated_at") - if ( - not isinstance(timestamp, str) - or re.fullmatch( - r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d{1,6})?(?:Z|\+00:00)", timestamp - ) - is None - ): - raise invalid() - try: - parsed = datetime.fromisoformat(timestamp.replace("Z", "+00:00")) - except ValueError as error: - raise invalid() from error - if parsed.utcoffset() != timedelta(0): - raise invalid() - return {**value, "last_seq": int(seq)} - - def _require_binding_goal_ref( binding: Mapping[str, Any], *, @@ -625,34 +234,6 @@ def next_seq( return highest + 1 -def runtime_root_digest(runtime_root: Path) -> str: - """Digest of the absolute, dot-normalized root as this process spells it. - - Diagnostic only (drain evidence). Outbox entries and receipts carry the - active binding's ``source_root_digest`` instead, which the TypeScript owner - derives from the resolved root, so a root reached through a symlink hashes - differently from this lexical spelling (#4892). - """ - - return text_digest(os.path.abspath(str(runtime_root))) - - -def record_source_ref(record: Mapping[str, Any]) -> str | None: - """The source reference an entry id binds: bytes digest, event id, or seed digest.""" - - source = _as_object(record.get("source")) - bytes_digest = source.get("bytes_digest") - if isinstance(bytes_digest, str) and bytes_digest: - return bytes_digest - event_id = source.get("event_id") - if isinstance(event_id, str) and event_id: - return f"event:{event_id}" - digest = record.get("partition_digest") - if isinstance(digest, str) and digest: - return f"seed:{digest}" - return None - - def _writer( write_class: str, *, runtime: str, operation_id: str | None ) -> dict[str, Any]: @@ -930,37 +511,6 @@ def committed(self) -> None: self._fail("outbox_commit_marker_failed", error) -def outbox_summary(runtime_root: Path, goal_id: str) -> dict[str, Any]: - """Counts per partition for operator readback; never raises on an empty outbox.""" - - summary: dict[str, Any] = {} - for partition in PARTITIONS: - directory = partition_directory(runtime_root, goal_id, partition) - try: - entries = list_entries(directory) - cursor = read_cursor(directory) - residue_count = len(retired_residue(directory)) - invalid: str | None = None - except OutboxError as error: - entries, cursor, invalid = [], None, error.reason_code - residue_count = 0 - except OSError: - entries, cursor, invalid = [], None, "outbox_file_unavailable" - residue_count = 0 - summary[partition] = { - "committed_pending": sum(1 for entry in entries if entry.is_committed), - "prepared_only": sum(1 for entry in entries if not entry.is_committed), - "retired_residue": residue_count, - "next_seq": ( - max((entry.seq for entry in entries), default=0) if entries else 0 - ), - "cursor_last_seq": int(cursor.get("last_seq") or 0) if cursor else None, - "cursor_last_entry_id": cursor.get("last_entry_id") if cursor else None, - "invalid": invalid, - } - return summary - - __all__ = [ "DRAIN_CURSOR_SCHEMA", "OUTBOX_COMMIT_SCHEMA", diff --git a/loopx/control_plane/coordination/local_authority_shadow_readback.py b/loopx/control_plane/coordination/local_authority_shadow_readback.py new file mode 100644 index 0000000000..c5a1214ba6 --- /dev/null +++ b/loopx/control_plane/coordination/local_authority_shadow_readback.py @@ -0,0 +1,182 @@ +"""Read-only shadow status and typed candidate-store transport. + +The operator can inspect retained bytes without any source capture producer. +No write, drain, promotion or replacement-store initialization is performed. +""" +from __future__ import annotations + +from dataclasses import asdict +from pathlib import Path +from typing import Any + +from ..projects.registry_codec import load_registry +from ...paths import resolve_runtime_root +from ...registry import find_registry_goal +from ..effect_runtime import effect_runtime_result +from . import local_authority_shadow_history as outbox +from .coordination_state_contract_generated import ( + LOCAL_AUTHORITY_SHADOW_READ_REQUEST_SCHEMA, + LOCAL_AUTHORITY_SHADOW_READ_RESULT_SCHEMA, +) +from .local_authority_shadow_projection import head_digest +from .configuration import resolve_coordination_runtime_shadow_config, local_authority_shadow_summary +from .shadow_management import read_shadow_capture_binding + +RETENTION_PRESSURE_BYTES = 8 * 1024 * 1024 + +def read_local_authority_shadow( + *, + runtime_root: Path, + goal_id: str, + store_kind: str = "runtime_shadow", + scan_after_cursor: str | None = None, + scan_limit: int = 0, + receipt_operation_id: str | None = None, + read_model: str = "full", +) -> dict[str, Any]: + """Read-only candidate view through the TypeScript store boundary.""" + + result = effect_runtime_result( + "coordination.runtime_shadow.outbox_read", + { + "schema_version": LOCAL_AUTHORITY_SHADOW_READ_REQUEST_SCHEMA, + "runtime_root": str(runtime_root), + "goal_id": goal_id, + "store_kind": store_kind, + "scan_after_cursor": scan_after_cursor, + "scan_limit": scan_limit, + "receipt_operation_id": receipt_operation_id, + "read_model": read_model, + }, + timeout=15.0, + ) + if ( + not isinstance(result, dict) + or result.get("schema_version") != LOCAL_AUTHORITY_SHADOW_READ_RESULT_SCHEMA + or result.get("goal_id") != goal_id + ): + raise RuntimeError("local authority shadow read result is invalid") + return dict(result) + +class _CandidateMissing(Exception): + """The candidate store directory does not exist yet.""" + + +def _store_bytes(runtime_root: Path, goal_id: str, *, legacy_observation: bool) -> int: + directory = ( + runtime_root / "authority-shadow" / "file" / goal_id + if legacy_observation + else runtime_root / "authority-shadow" / "file-v0" + ) + if not directory.is_dir(): + return 0 + return sum(path.stat().st_size for path in directory.iterdir() if path.is_file()) + + +def local_authority_shadow_status( + *, + registry_path: Path, + runtime_root: Path | None, + goal_id: str, +) -> dict[str, Any]: + """Operator readback: configuration, outbox backlog, and candidate head facts.""" + + registry = load_registry(registry_path) + goal = find_registry_goal(registry, goal_id) + if not isinstance(goal, dict): + raise ValueError(f"goal {goal_id!r} is not registered") + if runtime_root is None: + runtime_root = resolve_runtime_root(registry, None, registry_path=registry_path) + config = local_authority_shadow_summary(goal) + runtime_config = resolve_coordination_runtime_shadow_config(goal) + management = read_shadow_capture_binding(runtime_root, goal_id) + legacy_observation = ( + config.get("configured") is True + and not runtime_config.enabled + and management["status"] == "missing" + ) + backlog = outbox.outbox_summary(runtime_root, goal_id) + candidate: dict[str, Any] + try: + directory = ( + runtime_root / "authority-shadow" / "file" / goal_id + if legacy_observation + else runtime_root / "authority-shadow" / "file-v0" + ) + if not directory.is_dir() or not any(directory.glob("authority-store-*.json")): + # Reading through the store boundary would mint a store identity; + # a status probe must not create candidate lineage. + raise _CandidateMissing + view = read_local_authority_shadow( + runtime_root=runtime_root, + goal_id=goal_id, + store_kind=( + "legacy_observation" if legacy_observation else "runtime_shadow" + ), + ) + head = view.get("head") if isinstance(view.get("head"), dict) else None + candidate = { + "status": view.get("status"), + "reason_code": view.get("reason_code"), + "store_identity": view.get("store_identity"), + "provider_revision": view.get("provider_revision"), + "cursor": view.get("cursor"), + "head_digest": view.get("head_digest"), + "head_schema_version": head.get("schema_version") if head else None, + "partitions": view.get("partitions"), + "codec_agreement": (head_digest(head) == view.get("head_digest")) + if head + else None, + } + except _CandidateMissing: + candidate = { + "status": "missing", + "reason_code": None, + "store_identity": None, + "provider_revision": None, + "cursor": None, + "head_digest": None, + "head_schema_version": None, + "partitions": None, + "codec_agreement": None, + } + except Exception: + candidate = { + "status": "unavailable", + "reason_code": "shadow_read_failed", + "store_identity": None, + "provider_revision": None, + "cursor": None, + "head_digest": None, + "head_schema_version": None, + "partitions": None, + "codec_agreement": None, + } + candidate["store_kind"] = "legacy_observation" if legacy_observation else "runtime_shadow" + candidate["historical_only"] = legacy_observation + try: + store_bytes = _store_bytes( + runtime_root, goal_id, legacy_observation=legacy_observation + ) + storage_error = None + except OSError: + store_bytes = None + storage_error = "shadow_store_unavailable" + return { + "ok": all(item["invalid"] is None for item in backlog.values()) + and storage_error is None + and management["status"] != "hold", + "action": "status", + "goal_id": goal_id, + "config": config, + "runtime_config": asdict(runtime_config), + "management": management, + "storage_error": storage_error, + "runtime_root_digest": outbox.runtime_root_digest(runtime_root), + "outbox": backlog, + "candidate": candidate, + "store_bytes": store_bytes, + "retention_pressure": store_bytes > RETENTION_PRESSURE_BYTES + if store_bytes is not None + else None, + } diff --git a/loopx/semantics/project_registry_io_manifest_v1.json b/loopx/semantics/project_registry_io_manifest_v1.json index 73e25c99db..ddf64a7d8e 100644 --- a/loopx/semantics/project_registry_io_manifest_v1.json +++ b/loopx/semantics/project_registry_io_manifest_v1.json @@ -1118,8 +1118,8 @@ "classification": "codec_api" }, { - "site": "loopx/control_plane/coordination/local_authority_shadow_adapter.py::.local_authority_shadow_status::codec_read:load_registry#1", - "line": 252, + "site": "loopx/control_plane/coordination/local_authority_shadow_readback.py::.local_authority_shadow_status::codec_read:load_registry#1", + "line": 84, "column": 16, "kind": "codec_read", "api": "load_registry", diff --git a/tests/control_plane/test_shadow_readback_writer_isolation.py b/tests/control_plane/test_shadow_readback_writer_isolation.py new file mode 100644 index 0000000000..d1b452b819 --- /dev/null +++ b/tests/control_plane/test_shadow_readback_writer_isolation.py @@ -0,0 +1,153 @@ +"""Retained shadow diagnostics must survive source-producer retirement.""" +from __future__ import annotations + +import json +import subprocess + +import pytest + +from canonical_authority_fixture import initialize_canonical_authority +from test_canonical_todo_writer_isolation import ( + isolated_todo_distribution, retirement_wheel, # noqa: F401 -- distribution fixtures +) +from test_local_authority_shadow_outbox import ( + GOAL_ID, _capture, _fixture, _planned_todo, _record_change, +) +from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection +from loopx.todos import list_goal_todos + + +def _remove_capture_producers(package): + for name in ("runtime_shadow", "runtime_shadow_writer_adapter", + "local_authority_shadow_adapter", "local_authority_shadow_outbox"): + (package / "control_plane/coordination" / f"{name}.py").unlink() + (package / "control_plane/todos/legacy_mutation.py").unlink() + + +@pytest.mark.parametrize("configuration", [None, {"enabled": False}, {"enabled": True}]) +def test_empty_status_without_capture_producers_preserves_default_off( + tmp_path, configuration, isolated_todo_distribution, # noqa: F811 +): + package, command = isolated_todo_distribution + registry, state, runtime = _fixture(tmp_path, bootstrap=False) + if configuration is not None: + config = json.loads(registry.read_text()) + # Retired legacy observation configuration has no runtime schema. + config["goals"][0]["coordination"]["runtime_shadow"] = configuration + registry.write_text(json.dumps(config)) + original = registry.read_bytes(), state.read_bytes() + + def status(): + process = subprocess.run( + [*command, "--registry", str(registry), "--runtime-root", str(runtime), + "--format", "json", "authority-shadow", "status", "--goal-id", GOAL_ID], + cwd=tmp_path, capture_output=True, text=True, timeout=60, + ) + assert process.returncode == 0, process.stdout + process.stderr + result = json.loads(process.stdout) + assert result["runtime_config"]["enabled"] is False + assert result["candidate"]["status"] == "missing" + assert result["management"]["status"] == "missing" + assert not runtime.exists() + assert (registry.read_bytes(), state.read_bytes()) == original + return result + + expected = status() + _remove_capture_producers(package) + assert status() == expected + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_retained_status_is_read_only_with_capture_producers_absent( + tmp_path, provider, isolated_todo_distribution, # noqa: F811 +): + package, command = isolated_todo_distribution + registry, state, runtime = _fixture(tmp_path) + committed = _record_change(registry, state, runtime, "Retained committed work") + original = state.read_text() + _, planned = _planned_todo(original, "Prepared work never written") + prepared = _capture(registry, state, runtime, original_text=original) + prepared.prepare(planned) + assert prepared.outcome.failure is None + todos = list_goal_todos(registry_path=registry, goal_id=GOAL_ID)["todos"] + initialize_canonical_authority( + runtime, GOAL_ID, + build_todo_runtime_shadow_projection(goal_id=GOAL_ID, todos=todos, + handoff_mode="hard_lease"), + state_path=state, provider=provider, + ) + directory = runtime / "authority-shadow/outbox" / GOAL_ID / "todos" + candidate = next((runtime / "authority-shadow/file-v0").glob("authority-store-*.json")) + original_candidate = candidate.read_bytes() + + def files(): + return {str(p.relative_to(tmp_path)): p.read_bytes() + for root in (runtime, state.parent) for p in root.rglob("*") if p.is_file()} + + def status(): + before = files() + process = subprocess.run( + [*command, "--registry", str(registry), "--runtime-root", str(runtime), + "--format", "json", "authority-shadow", "status", "--goal-id", GOAL_ID], + cwd=tmp_path, capture_output=True, text=True, timeout=60, + ) + assert "Traceback" not in process.stderr, process.stderr + payload = json.loads(process.stdout) + assert files() == before, "a status read must not drain, repair or initialize anything" + assert process.returncode == (0 if payload["ok"] else 1), payload + return payload + + # Each oracle has independent invariant checks as well as complete + # base/head parity; original bytes and identities are never rewritten. + def scenarios(): + answers = {} + answers["pending"] = status() + assert answers["pending"]["ok"] + assert answers["pending"]["outbox"]["todos"]["committed_pending"] == 1 + assert answers["pending"]["outbox"]["todos"]["prepared_only"] == 1 + assert answers["pending"]["candidate"]["codec_agreement"] is True + assert answers["pending"]["management"]["status"] == "active" + + cursor = directory / "drain-cursor.json" + cursor.write_text(json.dumps({ + "schema_version": "loopx_local_authority_shadow_drain_cursor_v0", + "partition": "todos", "last_seq": committed.outcome.seq, + "last_entry_id": committed.outcome.entry_id, + "last_partition_digest": committed.outcome.partition_digest, + "last_cursor": "retained:1", "last_provider_revision": "retained:1", + "updated_at": "2026-09-03T00:00:00Z", + })) + answers["residue"] = status() + assert answers["residue"]["ok"] + assert answers["residue"]["outbox"]["todos"]["retired_residue"] == 2 + assert answers["residue"]["outbox"]["todos"]["committed_pending"] == 1 + cursor.write_text('{"invalid_original":true}') + answers["bad_cursor"] = status() + assert not answers["bad_cursor"]["ok"] + assert answers["bad_cursor"]["outbox"]["todos"]["invalid"] == "outbox_file_invalid" + cursor.unlink() + + extra = directory / "unclassified.json" + extra.write_text('{"retain_original":true}') + answers["bad_entry"] = status() + assert not answers["bad_entry"]["ok"] + assert answers["bad_entry"]["outbox"]["todos"]["invalid"] == "outbox_file_invalid" + extra.unlink() + + candidate.write_text('{"damaged_candidate":true}') + answers["bad_candidate"] = status() + # The typed store reader returns a failed result for invalid bytes; + # only transport exceptions become the adapter's unavailable status. + assert answers["bad_candidate"]["candidate"]["status"] == "failed" + assert answers["bad_candidate"]["candidate"]["reason_code"] is not None + candidate.unlink() + answers["missing_candidate"] = status() + assert answers["missing_candidate"]["candidate"]["status"] == "missing" + assert not candidate.exists(), "status must not mint a replacement store identity" + candidate.write_bytes(original_candidate) + return answers + + expected = scenarios() + _remove_capture_producers(package) + assert scenarios() == expected + assert scenarios() == expected, "cold repeated CLI reads preserve the original history"