From 08d6b43b2a8e9243371e7d2cb47481d6d56846ee Mon Sep 17 00:00:00 2001 From: now-ing Date: Sun, 6 Sep 2026 10:48:56 +0800 Subject: [PATCH 1/7] fix(post-writeback): preserve composition failure identity with retryable receipt A failed post-writeback composition projection no longer collapses its durable failure to the generic composition/unknown identity. The failure dispatch now reports each registered hook's concrete public-safe hook_id/capability_id, records one retryable receipt bound to the successful primary writeback in an append-only per-goal journal (goals//post_writeback_hooks/composition-retry-receipts.jsonl), and settles that receipt after a later replay composes the projection cleanly. The primary writeback is never rolled back or repeated, and no job queue, retry scheduler, or second writeback transaction is added. Closes #3917 Signed-off-by: now-ing --- loopx/cli_commands/post_writeback.py | 181 +++++++- .../post_writeback_composition_retry.py | 288 +++++++++++++ .../test_post_writeback_composition_retry.py | 404 ++++++++++++++++++ 3 files changed, 854 insertions(+), 19 deletions(-) create mode 100644 loopx/control_plane/post_writeback_composition_retry.py create mode 100644 tests/control_plane/test_post_writeback_composition_retry.py diff --git a/loopx/cli_commands/post_writeback.py b/loopx/cli_commands/post_writeback.py index cd1472bafa..8342dbb38a 100644 --- a/loopx/cli_commands/post_writeback.py +++ b/loopx/cli_commands/post_writeback.py @@ -9,12 +9,122 @@ PostWritebackHookRegistration, dispatch_post_writeback_hooks, ) +from ..control_plane.post_writeback_composition_retry import ( + append_composition_retry_receipt, + build_composition_retry_receipt, + composition_retry_receipt_id, + composition_retry_receipt_log_path, + composition_retry_receipt_ref, + settle_composition_retry_receipt, +) from ..history import load_registry from ..paths import resolve_runtime_root PostWritebackProjectionBuilder = Callable[..., Mapping[str, object]] +_COMPOSITION_HOOK_FIELDS = ("hook_id", "capability_id") + + +def _composition_hook_identities( + hooks: Sequence[PostWritebackHookRegistration], +) -> list[dict[str, str]]: + return [ + {field: str(getattr(registration, field) or "") for field in _COMPOSITION_HOOK_FIELDS} + for registration in hooks + ] + + +def _composition_failure_dispatch( + hooks: Sequence[PostWritebackHookRegistration], + *, + error_code: str, + receipt_ref: str | None, +) -> dict[str, Any]: + """Report the concrete public-safe hook identities behind one composition failure.""" + + failures: list[dict[str, str]] = [] + for registration in hooks: + failure = { + "hook_id": str(registration.hook_id or ""), + "capability_id": str(registration.capability_id or ""), + "error_code": error_code, + } + if receipt_ref is not None: + failure["durable_receipt_ref"] = receipt_ref + failures.append(failure) + return { + "schema_version": POST_WRITEBACK_HOOK_DISPATCH_SCHEMA_VERSION, + "phase": "post_writeback", + "registered_count": len(hooks), + "invoked_count": 0, + "replayed_hooks": [], + "retried_hooks": [], + "intent_count": 0, + "intents": [], + "failures": failures, + "primary_writeback_preserved": True, + "external_writes_performed": False, + } + + +def _recorded_composition_failure( + journal_path: Path, + *, + hooks: Sequence[PostWritebackHookRegistration], + goal_id: str, + event_kind: str, + identity: Mapping[str, Any], + state_version: str, + committed_at: str, + error_code: str, +) -> dict[str, Any]: + """Persist one retryable receipt, degrading to identity-only on journal errors.""" + + receipt = build_composition_retry_receipt( + goal_id=goal_id, + event_kind=event_kind, + identity=identity, + state_version=state_version, + committed_at=committed_at, + hook_identities=_composition_hook_identities(hooks), + error_code=error_code, + ) + receipt_ref: str | None = composition_retry_receipt_ref( + str(receipt["receipt_id"]) + ) + try: + append_composition_retry_receipt(journal_path, receipt) + except (OSError, ValueError): + receipt_ref = None + return _composition_failure_dispatch( + hooks, error_code=error_code, receipt_ref=receipt_ref + ) + + +def _settle_composition_retry_quietly( + journal_path: Path, + *, + goal_id: str, + event_kind: str, + identity: Mapping[str, Any], + state_version: str, + committed_at: str, + hooks: Sequence[PostWritebackHookRegistration], +) -> None: + try: + settle_composition_retry_receipt( + journal_path, + goal_id=goal_id, + event_kind=event_kind, + identity=identity, + state_version=state_version, + committed_at=committed_at, + hook_identities=_composition_hook_identities(hooks), + ) + except (OSError, ValueError): + return + def dispatch_committed_cli_post_writeback_hooks( *, @@ -31,14 +141,30 @@ def dispatch_committed_cli_post_writeback_hooks( ) -> dict[str, Any]: """Bridge one committed CLI mutation into the TS-owned hook lifecycle. - Projection and provider failures are isolated from the primary mutation. - The helper intentionally owns no capability policy and grants no effects. + Projection and provider failures are isolated from the primary mutation: + the primary writeback is never rolled back or repeated, a failed + composition records one retryable receipt bound to that writeback, and a + later replay of the same writeback settles the receipt once the + projection composes cleanly. The helper intentionally owns no capability + policy and grants no effects. """ try: runtime_root = resolve_runtime_root( load_registry(registry_path), runtime_root_arg ) + journal_path = composition_retry_receipt_log_path(runtime_root, goal_id) + composition_retry_receipt_id( + goal_id=goal_id, + event_kind=event_kind, + identity=identity, + state_version=state_version, + ) + except Exception: # Optional hooks never alter primary truth. + return _composition_failure_dispatch( + hooks, error_code="source_projection_failed", receipt_ref=None + ) + try: projection = ( dict( projection_builder( @@ -52,7 +178,19 @@ def dispatch_committed_cli_post_writeback_hooks( if projection_builder is not None else {} ) - return dispatch_post_writeback_hooks( + except Exception: # The source projection stays retryable, never primary. + return _recorded_composition_failure( + journal_path, + hooks=hooks, + goal_id=goal_id, + event_kind=event_kind, + identity=identity, + state_version=state_version, + committed_at=committed_at, + error_code="source_projection_failed", + ) + try: + result = dispatch_post_writeback_hooks( hooks, source={ "schema_version": "loopx_post_writeback_hook_source_v0", @@ -74,22 +212,27 @@ def dispatch_committed_cli_post_writeback_hooks( }, runtime_root=runtime_root, ) - except Exception: # Optional hooks never alter primary truth. - return { - "schema_version": POST_WRITEBACK_HOOK_DISPATCH_SCHEMA_VERSION, - "phase": "post_writeback", - "registered_count": len(hooks), - "intent_count": 0, - "failures": [ - { - "hook_id": "composition", - "capability_id": "unknown", - "error_code": "source_projection_failed", - } - ], - "primary_writeback_preserved": True, - "external_writes_performed": False, - } + except Exception: # An unexpected transport collapse stays retryable. + return _recorded_composition_failure( + journal_path, + hooks=hooks, + goal_id=goal_id, + event_kind=event_kind, + identity=identity, + state_version=state_version, + committed_at=committed_at, + error_code="dispatch_failed", + ) + _settle_composition_retry_quietly( + journal_path, + goal_id=goal_id, + event_kind=event_kind, + identity=identity, + state_version=state_version, + committed_at=committed_at, + hooks=hooks, + ) + return result __all__ = [ diff --git a/loopx/control_plane/post_writeback_composition_retry.py b/loopx/control_plane/post_writeback_composition_retry.py new file mode 100644 index 0000000000..1dec233188 --- /dev/null +++ b/loopx/control_plane/post_writeback_composition_retry.py @@ -0,0 +1,288 @@ +from __future__ import annotations + +import hashlib +import json +from collections.abc import Mapping, Sequence +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +from ..file_lock import exclusive_file_lock +from ..history import validate_goal_id_path_segment + + +POST_WRITEBACK_COMPOSITION_RETRY_RECEIPT_SCHEMA_VERSION = ( + "loopx_post_writeback_composition_retry_receipt_v0" +) +POST_WRITEBACK_COMPOSITION_RETRY_LOG_NAME = "composition-retry-receipts.jsonl" +POST_WRITEBACK_COMPOSITION_RETRY_REF_PREFIX = "post-writeback-composition:" +POST_WRITEBACK_COMPOSITION_RETRY_ERROR_CODES = ( + "source_projection_failed", + "dispatch_failed", +) +_COMPOSITION_RETRY_JOURNAL_ROW_LIMIT = 512 + + +def _now_iso() -> str: + return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") + + +def composition_retry_receipt_log_path(runtime_root: Path, goal_id: str) -> Path: + """Resolve the per-goal append-only journal for composition retry receipts.""" + + return ( + runtime_root.expanduser() + / "goals" + / validate_goal_id_path_segment(goal_id) + / "post_writeback_hooks" + / POST_WRITEBACK_COMPOSITION_RETRY_LOG_NAME + ) + + +def composition_retry_receipt_id( + *, + goal_id: str, + event_kind: str, + identity: Mapping[str, Any], + state_version: str, +) -> str: + """Bind one receipt to the exact primary writeback it observes.""" + + stable = { + "goal_id": str(goal_id or ""), + "event_kind": str(event_kind or ""), + "agent_id": str(identity.get("agent_id") or ""), + "todo_id": str(identity.get("todo_id") or ""), + "turn_instance_id": str(identity.get("turn_instance_id") or ""), + "effect_id": str(identity.get("effect_id") or ""), + "state_version": str(state_version or ""), + } + encoded = json.dumps(stable, sort_keys=True, ensure_ascii=True).encode("utf-8") + return "pwcr_" + hashlib.sha256(encoded).hexdigest()[:16] + + +def composition_retry_receipt_ref(receipt_id: str) -> str: + return f"{POST_WRITEBACK_COMPOSITION_RETRY_REF_PREFIX}{receipt_id}" + + +def build_composition_retry_receipt( + *, + goal_id: str, + event_kind: str, + identity: Mapping[str, Any], + state_version: str, + committed_at: str, + hook_identities: Sequence[Mapping[str, str]], + error_code: str | None = None, + status: str = "retryable", + recorded_at: str | None = None, +) -> dict[str, Any]: + """Build one public-safe retry receipt bound to a committed primary writeback. + + The receipt records lifecycle identity only: no projection payload, task + text, local paths, or provider output is ever embedded. + """ + + if status not in {"retryable", "settled"}: + raise ValueError("composition retry receipt status is invalid") + if status == "retryable" and error_code not in ( + POST_WRITEBACK_COMPOSITION_RETRY_ERROR_CODES + ): + raise ValueError("composition retry receipt error_code is invalid") + bounded_identities: list[dict[str, str]] = [] + seen_hook_ids: set[str] = set() + for raw_identity in hook_identities: + hook_id = str(raw_identity.get("hook_id") or "")[:200] + capability_id = str(raw_identity.get("capability_id") or "")[:200] + if not hook_id or hook_id in seen_hook_ids: + continue + seen_hook_ids.add(hook_id) + bounded_identities.append( + {"hook_id": hook_id, "capability_id": capability_id} + ) + return { + "schema_version": POST_WRITEBACK_COMPOSITION_RETRY_RECEIPT_SCHEMA_VERSION, + "receipt_id": composition_retry_receipt_id( + goal_id=goal_id, + event_kind=event_kind, + identity=identity, + state_version=state_version, + ), + "status": status, + "error_code": error_code, + "event_kind": str(event_kind or ""), + "identity": { + "goal_id": str(goal_id or ""), + "agent_id": str(identity.get("agent_id") or ""), + "todo_id": str(identity.get("todo_id") or ""), + "turn_instance_id": str(identity.get("turn_instance_id") or ""), + "effect_id": str(identity.get("effect_id") or ""), + }, + "state_version": str(state_version or ""), + "committed_at": str(committed_at or ""), + "hooks": bounded_identities, + "primary_writeback_preserved": True, + "external_writes_performed": False, + "recorded_at": recorded_at or _now_iso(), + } + + +def _iter_composition_retry_rows( + log_path: Path, *, row_limit: int = _COMPOSITION_RETRY_JOURNAL_ROW_LIMIT +) -> list[dict[str, Any]]: + """Read a bounded suffix of valid journal rows, oldest first.""" + + try: + lines = log_path.read_text(encoding="utf-8").splitlines() + except OSError: + return [] + rows: list[dict[str, Any]] = [] + for line in lines[-max(0, row_limit) :]: + text = line.strip() + if not text: + continue + try: + row = json.loads(text) + except json.JSONDecodeError: + continue + if not isinstance(row, dict): + continue + if ( + row.get("schema_version") + != POST_WRITEBACK_COMPOSITION_RETRY_RECEIPT_SCHEMA_VERSION + or not isinstance(row.get("receipt_id"), str) + or row.get("status") not in {"retryable", "settled"} + ): + continue + rows.append(row) + return rows + + +def _current_composition_retry_row( + handle: Any, receipt_id: str +) -> dict[str, Any] | None: + handle.seek(0) + current: dict[str, Any] | None = None + for line in handle: + text = line.strip() + if not text or receipt_id not in text: + continue + try: + row = json.loads(text) + except json.JSONDecodeError: + continue + if ( + isinstance(row, dict) + and row.get("schema_version") + == POST_WRITEBACK_COMPOSITION_RETRY_RECEIPT_SCHEMA_VERSION + and row.get("receipt_id") == receipt_id + and row.get("status") in {"retryable", "settled"} + ): + current = row + return current + + +def append_composition_retry_receipt( + log_path: Path, receipt: Mapping[str, Any] +) -> tuple[dict[str, Any], bool]: + """Append one receipt row once, never regressing a settled receipt. + + Returns the durable row and whether this call appended it. Re-recording a + still-retryable receipt appends the newest observation; a settled receipt + is terminal and is returned unchanged. + """ + + payload = dict(receipt) + if ( + payload.get("schema_version") + != POST_WRITEBACK_COMPOSITION_RETRY_RECEIPT_SCHEMA_VERSION + ): + raise ValueError("unsupported composition retry receipt schema") + receipt_id = str(payload.get("receipt_id") or "") + if not receipt_id: + raise ValueError("composition retry receipt_id is required") + log_path.parent.mkdir(parents=True, exist_ok=True) + with exclusive_file_lock(log_path): + with log_path.open("a+", encoding="utf-8") as handle: + current = _current_composition_retry_row(handle, receipt_id) + if current is not None and current.get("status") == "settled": + return current, False + handle.write( + json.dumps(payload, sort_keys=True, ensure_ascii=False) + "\n" + ) + return payload, True + + +def settle_composition_retry_receipt( + log_path: Path, + *, + goal_id: str, + event_kind: str, + identity: Mapping[str, Any], + state_version: str, + committed_at: str, + hook_identities: Sequence[Mapping[str, str]], +) -> tuple[dict[str, Any], bool]: + """Supersede one retryable receipt after its projection composed cleanly.""" + + settled = build_composition_retry_receipt( + goal_id=goal_id, + event_kind=event_kind, + identity=identity, + state_version=state_version, + committed_at=committed_at, + hook_identities=hook_identities, + error_code=None, + status="settled", + ) + receipt_id = str(settled["receipt_id"]) + if not log_path.is_file(): + return {}, False + with exclusive_file_lock(log_path): + with log_path.open("r+", encoding="utf-8") as handle: + current = _current_composition_retry_row(handle, receipt_id) + if current is not None and current.get("status") == "settled": + return current, False + if current is None: + return {}, False + handle.seek(0, 2) + handle.write( + json.dumps(settled, sort_keys=True, ensure_ascii=False) + "\n" + ) + return settled, True + + +def pending_composition_retry_receipts( + runtime_root: Path, goal_id: str +) -> list[dict[str, Any]]: + """Read unconsumed retryable receipts for one goal, newest row per receipt.""" + + rows = _iter_composition_retry_rows( + composition_retry_receipt_log_path(runtime_root, goal_id) + ) + latest: dict[str, dict[str, Any]] = {} + for row in rows: + latest[str(row.get("receipt_id"))] = row + return sorted( + ( + dict(row) + for row in latest.values() + if row.get("status") == "retryable" + ), + key=lambda row: str(row.get("receipt_id") or ""), + ) + + +__all__ = [ + "POST_WRITEBACK_COMPOSITION_RETRY_ERROR_CODES", + "POST_WRITEBACK_COMPOSITION_RETRY_LOG_NAME", + "POST_WRITEBACK_COMPOSITION_RETRY_RECEIPT_SCHEMA_VERSION", + "POST_WRITEBACK_COMPOSITION_RETRY_REF_PREFIX", + "append_composition_retry_receipt", + "build_composition_retry_receipt", + "composition_retry_receipt_id", + "composition_retry_receipt_log_path", + "composition_retry_receipt_ref", + "pending_composition_retry_receipts", + "settle_composition_retry_receipt", +] diff --git a/tests/control_plane/test_post_writeback_composition_retry.py b/tests/control_plane/test_post_writeback_composition_retry.py new file mode 100644 index 0000000000..0decef8fa9 --- /dev/null +++ b/tests/control_plane/test_post_writeback_composition_retry.py @@ -0,0 +1,404 @@ +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +from loopx.cli_commands.post_writeback import ( + dispatch_committed_cli_post_writeback_hooks, +) +from loopx.control_plane.capability_hooks import ( + POST_WRITEBACK_HOOK_RESULT_SCHEMA_VERSION, + PostWritebackHookRegistration, +) +from loopx.control_plane.post_writeback_composition_retry import ( + POST_WRITEBACK_COMPOSITION_RETRY_RECEIPT_SCHEMA_VERSION, + append_composition_retry_receipt, + build_composition_retry_receipt, + composition_retry_receipt_id, + composition_retry_receipt_log_path, + composition_retry_receipt_ref, + pending_composition_retry_receipts, + settle_composition_retry_receipt, +) + + +def _write_registry(tmp_path: Path) -> tuple[Path, Path]: + runtime_root = tmp_path / "runtime" + runtime_root.mkdir(exist_ok=True) + registry_path = tmp_path / "registry.global.json" + registry_path.write_text( + json.dumps( + { + "common_runtime_root": str(runtime_root), + "goals": [{"id": "goal-1"}], + } + ), + encoding="utf-8", + ) + return registry_path, runtime_root + + +def _identity() -> dict[str, str]: + return { + "agent_id": "agent-1", + "todo_id": "todo-1", + "turn_instance_id": "turn-1", + "effect_id": "goal-1:agent-1:todo-1:turn-1", + } + + +def _hook( + *, producer_calls: list[int] | None = None +) -> PostWritebackHookRegistration: + def producer(value: object) -> dict[str, object]: + if producer_calls is not None: + producer_calls.append(1) + assert isinstance(value, dict) + receipt = value["receipt"] + assert isinstance(receipt, dict) + return { + "schema_version": POST_WRITEBACK_HOOK_RESULT_SCHEMA_VERSION, + "hook_id": "periodic_report.stage_completion", + "capability_id": "periodic-report", + "phase": "post_writeback", + "status": "intent", + "intent": { + "schema_version": "loopx_capability_intent_v0", + "intent_kind": "periodic_report.trigger_evaluation", + "idempotency_key": "periodic-report:stage-123", + "source_receipt_id": receipt["event_id"], + "payload": {"stage_identity": "stage-123"}, + "requested_write_scope": [], + }, + } + + return PostWritebackHookRegistration( + hook_id="periodic_report.stage_completion", + capability_id="periodic-report", + event_kinds=("todo_complete",), + intent_kinds=("periodic_report.trigger_evaluation",), + requested_read_scope=("stage_completion",), + producer=producer, + ) + + +def _stage_projection() -> dict[str, object]: + return { + "stage_completion": { + "schema_version": "periodic_report_stage_completion_receipt_v0", + "stage_identity": "stage-123", + } + } + + +def _dispatch( + registry_path: Path, + *, + hooks: tuple[PostWritebackHookRegistration, ...], + projection_builder: Any, +) -> dict[str, Any]: + return dispatch_committed_cli_post_writeback_hooks( + payload={"ok": True, "completed": True}, + registry_path=registry_path, + runtime_root_arg=None, + goal_id="goal-1", + event_kind="todo_complete", + identity=_identity(), + state_version="2026-09-06T00:00:00Z", + committed_at="2026-09-06T00:00:00Z", + hooks=hooks, + projection_builder=projection_builder, + ) + + +def _journal_rows(runtime_root: Path) -> list[dict[str, Any]]: + journal_path = composition_retry_receipt_log_path(runtime_root, "goal-1") + return [ + json.loads(line) + for line in journal_path.read_text(encoding="utf-8").splitlines() + if line.strip() + ] + + +def test_failed_projection_preserves_actionable_hook_identity( + tmp_path: Path, +) -> None: + """Acceptance 1: the failed projection keeps its concrete identity.""" + + registry_path, runtime_root = _write_registry(tmp_path) + + def failing_builder(**_kwargs: object) -> dict[str, object]: + raise RuntimeError("transient projection failure") + + result = _dispatch( + registry_path, hooks=(_hook(),), projection_builder=failing_builder + ) + + assert result["registered_count"] == 1 + assert result["intent_count"] == 0 + assert result["primary_writeback_preserved"] is True + assert result["external_writes_performed"] is False + (failure,) = result["failures"] + assert failure["hook_id"] == "periodic_report.stage_completion" + assert failure["capability_id"] == "periodic-report" + assert failure["error_code"] == "source_projection_failed" + assert failure["durable_receipt_ref"].startswith( + "post-writeback-composition:pwcr_" + ) + + pending = pending_composition_retry_receipts(runtime_root, "goal-1") + (receipt,) = pending + assert receipt["schema_version"] == ( + POST_WRITEBACK_COMPOSITION_RETRY_RECEIPT_SCHEMA_VERSION + ) + assert receipt["status"] == "retryable" + assert receipt["error_code"] == "source_projection_failed" + assert receipt["event_kind"] == "todo_complete" + assert receipt["identity"]["effect_id"] == "goal-1:agent-1:todo-1:turn-1" + assert receipt["identity"]["todo_id"] == "todo-1" + assert receipt["state_version"] == "2026-09-06T00:00:00Z" + assert receipt["committed_at"] == "2026-09-06T00:00:00Z" + assert receipt["hooks"] == [ + { + "hook_id": "periodic_report.stage_completion", + "capability_id": "periodic-report", + } + ] + assert receipt["primary_writeback_preserved"] is True + assert receipt["external_writes_performed"] is False + + +def test_replay_after_transient_recovery_projects_once_and_settles_receipt( + tmp_path: Path, +) -> None: + """Acceptance 2: recovery replays the projection once and settles the receipt.""" + + registry_path, runtime_root = _write_registry(tmp_path) + projection_calls: list[int] = [] + producer_calls: list[int] = [] + + def flaky_builder(**_kwargs: object) -> dict[str, object]: + projection_calls.append(1) + if len(projection_calls) == 1: + raise RuntimeError("transient projection failure") + return _stage_projection() + + hooks = (_hook(producer_calls=producer_calls),) + + failed = _dispatch( + registry_path, hooks=hooks, projection_builder=flaky_builder + ) + assert failed["failures"][0]["error_code"] == "source_projection_failed" + assert [receipt["status"] for receipt in pending_composition_retry_receipts(runtime_root, "goal-1")] == ["retryable"] + + recovered = _dispatch( + registry_path, hooks=hooks, projection_builder=flaky_builder + ) + assert recovered["failures"] == [] + assert recovered["intent_count"] == 1 + assert len(producer_calls) == 1 + assert len(projection_calls) == 2 + assert pending_composition_retry_receipts(runtime_root, "goal-1") == [] + rows = _journal_rows(runtime_root) + assert [row["status"] for row in rows if row["receipt_id"] == rows[-1]["receipt_id"]][-1] == "settled" + + replayed = _dispatch( + registry_path, hooks=hooks, projection_builder=flaky_builder + ) + assert replayed["intent_count"] == 1 + assert replayed["replayed_hooks"] == ["periodic_report.stage_completion"] + assert len(producer_calls) == 1 + assert len(projection_calls) == 3 + assert pending_composition_retry_receipts(runtime_root, "goal-1") == [] + settled_ids = { + row["receipt_id"] for row in _journal_rows(runtime_root) + } + assert len(settled_ids) == 1 + + +def test_composition_replay_keeps_primary_writeback_idempotent( + tmp_path: Path, +) -> None: + """Acceptance 3: replaying the primary writeback never repeats its effect.""" + + registry_path, runtime_root = _write_registry(tmp_path) + primary_effects: list[str] = ["goal-1:agent-1:todo-1:turn-1"] + projection_calls: list[int] = [] + producer_calls: list[int] = [] + + def flaky_builder(**_kwargs: object) -> dict[str, object]: + projection_calls.append(1) + if len(projection_calls) <= 2: + raise RuntimeError("transient projection failure") + return _stage_projection() + + hooks = (_hook(producer_calls=producer_calls),) + first_failure = _dispatch( + registry_path, hooks=hooks, projection_builder=flaky_builder + ) + second_failure = _dispatch( + registry_path, hooks=hooks, projection_builder=flaky_builder + ) + recovered = _dispatch( + registry_path, hooks=hooks, projection_builder=flaky_builder + ) + + for result in (first_failure, second_failure, recovered): + assert result["primary_writeback_preserved"] is True + assert result["external_writes_performed"] is False + assert primary_effects == ["goal-1:agent-1:todo-1:turn-1"] + assert len(producer_calls) == 1 + + rows = _journal_rows(runtime_root) + assert len({row["receipt_id"] for row in rows}) == 1 + statuses = [row["status"] for row in rows] + assert statuses == ["retryable", "retryable", "settled"] + assert pending_composition_retry_receipts(runtime_root, "goal-1") == [] + + +def test_composition_retry_receipt_id_binds_primary_writeback_identity() -> None: + base = composition_retry_receipt_id( + goal_id="goal-1", + event_kind="todo_complete", + identity=_identity(), + state_version="vision-revision-2", + ) + assert base.startswith("pwcr_") + assert base == composition_retry_receipt_id( + goal_id="goal-1", + event_kind="todo_complete", + identity=_identity(), + state_version="vision-revision-2", + ) + assert base != composition_retry_receipt_id( + goal_id="goal-1", + event_kind="todo_complete", + identity=_identity(), + state_version="vision-revision-3", + ) + assert base != composition_retry_receipt_id( + goal_id="goal-2", + event_kind="todo_complete", + identity=_identity(), + state_version="vision-revision-2", + ) + assert composition_retry_receipt_ref(base) == ( + f"post-writeback-composition:{base}" + ) + + +def test_composition_retry_journal_append_is_idempotent_and_terminal( + tmp_path: Path, +) -> None: + journal_path = tmp_path / "goals" / "goal-1" / ( + "post_writeback_hooks" + ) / "composition-retry-receipts.jsonl" + receipt = build_composition_retry_receipt( + goal_id="goal-1", + event_kind="todo_complete", + identity=_identity(), + state_version="vision-revision-2", + committed_at="2026-09-06T00:00:00Z", + hook_identities=[ + {"hook_id": "periodic_report.stage_completion", "capability_id": "periodic-report"} + ], + error_code="source_projection_failed", + ) + + appended_first, appended_flag_first = append_composition_retry_receipt( + journal_path, receipt + ) + appended_again, appended_flag_again = append_composition_retry_receipt( + journal_path, receipt + ) + assert appended_flag_first is True + assert appended_flag_again is True + + settled, settled_flag = settle_composition_retry_receipt( + journal_path, + goal_id="goal-1", + event_kind="todo_complete", + identity=_identity(), + state_version="vision-revision-2", + committed_at="2026-09-06T00:00:00Z", + hook_identities=[ + {"hook_id": "periodic_report.stage_completion", "capability_id": "periodic-report"} + ], + ) + assert settled_flag is True + assert settled["status"] == "settled" + assert settled["error_code"] is None + + regressed, regressed_flag = append_composition_retry_receipt( + journal_path, receipt + ) + assert regressed_flag is False + assert regressed["status"] == "settled" + rows = [ + json.loads(line) + for line in journal_path.read_text(encoding="utf-8").splitlines() + if line.strip() + ] + assert [row["status"] for row in rows] == ["retryable", "retryable", "settled"] + + resettle, resettle_flag = settle_composition_retry_receipt( + journal_path, + goal_id="goal-1", + event_kind="todo_complete", + identity=_identity(), + state_version="vision-revision-2", + committed_at="2026-09-06T00:00:00Z", + hook_identities=[], + ) + assert resettle_flag is False + assert resettle["status"] == "settled" + assert len(rows) == 3 + + +def test_settle_without_pending_receipt_is_a_noop(tmp_path: Path) -> None: + journal_path = composition_retry_receipt_log_path(tmp_path, "goal-1") + settled, appended = settle_composition_retry_receipt( + journal_path, + goal_id="goal-1", + event_kind="todo_complete", + identity=_identity(), + state_version="vision-revision-2", + committed_at="2026-09-06T00:00:00Z", + hook_identities=[], + ) + assert appended is False + assert settled == {} + assert not journal_path.exists() + + +def test_pending_receipts_skip_foreign_and_malformed_rows( + tmp_path: Path, +) -> None: + runtime_root = tmp_path + journal_path = composition_retry_receipt_log_path(runtime_root, "goal-1") + journal_path.parent.mkdir(parents=True, exist_ok=True) + pending_receipt = build_composition_retry_receipt( + goal_id="goal-1", + event_kind="todo_complete", + identity=_identity(), + state_version="vision-revision-2", + committed_at="2026-09-06T00:00:00Z", + hook_identities=[{"hook_id": "h.one", "capability_id": "cap-one"}], + error_code="source_projection_failed", + ) + journal_path.write_text( + "\n".join( + [ + "not-json", + json.dumps({"schema_version": "unrelated_v0"}), + json.dumps(pending_receipt), + ] + ) + + "\n", + encoding="utf-8", + ) + pending = pending_composition_retry_receipts(runtime_root, "goal-1") + assert [receipt["receipt_id"] for receipt in pending] == [ + pending_receipt["receipt_id"] + ] From 774df713a17f0e1ab8ab54d95bd8ed4aabebf002 Mon Sep 17 00:00:00 2001 From: now-ing Date: Sun, 6 Sep 2026 12:53:09 +0800 Subject: [PATCH 2/7] test(post-writeback): pin composition receipt degradation and settle semantics Add the two counterfactuals a strict reviewer asks for next: a broken composition journal must degrade to identity-only dispatch output (no durable_receipt_ref, no journal file, concrete hook identity kept), and a composed projection must settle the composition receipt even when a hook-level producer fails, because the receipt tracks projection composition while hook failures keep their own per-hook trail. Document that division on the settle call. Signed-off-by: now-ing --- loopx/cli_commands/post_writeback.py | 3 + .../test_post_writeback_composition_retry.py | 86 +++++++++++++++++++ 2 files changed, 89 insertions(+) diff --git a/loopx/cli_commands/post_writeback.py b/loopx/cli_commands/post_writeback.py index 8342dbb38a..ddd24ab846 100644 --- a/loopx/cli_commands/post_writeback.py +++ b/loopx/cli_commands/post_writeback.py @@ -223,6 +223,9 @@ def dispatch_committed_cli_post_writeback_hooks( committed_at=committed_at, error_code="dispatch_failed", ) + # The composition receipt tracks projection composition only: it settles + # once the projection composed and the hook lifecycle returned, while any + # hook-level failure keeps its own per-hook failure trail in `result`. _settle_composition_retry_quietly( journal_path, goal_id=goal_id, diff --git a/tests/control_plane/test_post_writeback_composition_retry.py b/tests/control_plane/test_post_writeback_composition_retry.py index 0decef8fa9..fddc21214e 100644 --- a/tests/control_plane/test_post_writeback_composition_retry.py +++ b/tests/control_plane/test_post_writeback_composition_retry.py @@ -4,6 +4,8 @@ from pathlib import Path from typing import Any +import pytest + from loopx.cli_commands.post_writeback import ( dispatch_committed_cli_post_writeback_hooks, ) @@ -402,3 +404,87 @@ def test_pending_receipts_skip_foreign_and_malformed_rows( assert [receipt["receipt_id"] for receipt in pending] == [ pending_receipt["receipt_id"] ] + + +def test_journal_unavailable_degrades_to_identity_without_receipt_ref( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A broken journal must not erase the concrete failure identity either.""" + + import loopx.cli_commands.post_writeback as post_writeback_module + + registry_path, runtime_root = _write_registry(tmp_path) + + def failing_builder(**_kwargs: object) -> dict[str, object]: + raise RuntimeError("transient projection failure") + + def broken_append(*_args: object, **_kwargs: object) -> None: + raise OSError("journal unavailable") + + monkeypatch.setattr( + post_writeback_module, "append_composition_retry_receipt", broken_append + ) + result = _dispatch( + registry_path, + hooks=(_hook(),), + projection_builder=failing_builder, + ) + assert result["registered_count"] == 1 + assert result["intent_count"] == 0 + assert result["primary_writeback_preserved"] is True + failure = result["failures"][0] + assert failure["hook_id"] == "periodic_report.stage_completion" + assert failure["capability_id"] == "periodic-report" + assert failure["error_code"] == "source_projection_failed" + assert "durable_receipt_ref" not in failure + assert not composition_retry_receipt_log_path( + runtime_root, "goal-1" + ).exists() + + +def test_producer_failure_settles_composition_receipt_with_hook_level_trail( + tmp_path: Path, +) -> None: + """Composition settles on a composed projection; hook failures keep their own trail.""" + + registry_path, runtime_root = _write_registry(tmp_path) + + def failing_builder(**_kwargs: object) -> dict[str, object]: + raise RuntimeError("transient projection failure") + + first = _dispatch( + registry_path, + hooks=(_hook(),), + projection_builder=failing_builder, + ) + assert first["intent_count"] == 0 + pending = pending_composition_retry_receipts(runtime_root, "goal-1") + assert len(pending) == 1 + + def broken_producer(_value: object) -> dict[str, object]: + raise RuntimeError("hook producer collapsed") + + def hook_with_broken_producer() -> PostWritebackHookRegistration: + registration = _hook() + object.__setattr__(registration, "producer", broken_producer) + return registration + + second = _dispatch( + registry_path, + hooks=(hook_with_broken_producer(),), + projection_builder=lambda **_kwargs: _stage_projection(), + ) + hook_failure = next( + ( + item + for item in second.get("failures", []) + if item.get("hook_id") == "periodic_report.stage_completion" + ), + None, + ) + assert hook_failure is not None + rows = _journal_rows(runtime_root) + assert any(row.get("status") == "settled" for row in rows), ( + "a composed projection must settle the composition receipt even when a " + "hook-level producer fails: hook failures carry their own durable trail" + ) From 353e5164a3db168efcc24721fcebd7d53f07b27e Mon Sep 17 00:00:00 2001 From: now-ing Date: Sun, 6 Sep 2026 18:56:03 +0800 Subject: [PATCH 3/7] fix(post-writeback): bind composition receipts to the hook set and surface pending retries Three durability gaps from the review are closed: - The receipt identity now includes a normalized, order-insensitive digest of the registered hook set, so a composition that later succeeds with a different hook set settles a different receipt (or none at all). A replaced, removed, or reordered hook can no longer mark the original failure resolved; the cross-hook-set counterexample is pinned as a test. - Pending readback scans every journal row before folding to the newest row per receipt id, so no unresolved receipt can fall out of the view past a row bound. Storage governance moved to append-time compaction under the lock, which folds repeated observations of one receipt while keeping every distinct unresolved receipt durable. - The pending view gained a production readback consumer: status attaches a bounded pending-composition-retry projection (only when something is pending, so healthy status output is unchanged) with the replay action, and a later-turn test discovers a pending receipt through that projection, replays the committed mutation, and observes it cleared. Signed-off-by: now-ing --- loopx/cli_commands/post_writeback.py | 1 + loopx/cli_commands/status.py | 10 + .../post_writeback_composition_retry.py | 189 ++++++++++-- .../test_post_writeback_composition_retry.py | 283 +++++++++++++++--- 4 files changed, 420 insertions(+), 63 deletions(-) diff --git a/loopx/cli_commands/post_writeback.py b/loopx/cli_commands/post_writeback.py index ddd24ab846..eb19f0354c 100644 --- a/loopx/cli_commands/post_writeback.py +++ b/loopx/cli_commands/post_writeback.py @@ -159,6 +159,7 @@ def dispatch_committed_cli_post_writeback_hooks( event_kind=event_kind, identity=identity, state_version=state_version, + hook_identities=_composition_hook_identities(hooks), ) except Exception: # Optional hooks never alter primary truth. return _composition_failure_dispatch( diff --git a/loopx/cli_commands/status.py b/loopx/cli_commands/status.py index 6637f6b423..733649b2e3 100644 --- a/loopx/cli_commands/status.py +++ b/loopx/cli_commands/status.py @@ -11,6 +11,9 @@ resolve_status_projection_cache_runtime_root, write_status_projection_cache, ) +from ..control_plane.post_writeback_composition_retry import ( + collect_pending_composition_retry_projection, +) from ..control_plane.status.agent_lane_projection import ( compact_agent_lane_status_payload_for_display, ) @@ -232,6 +235,13 @@ def handle_status_command( agent_id=args.agent_id, ) compact_agent_lane_todo_index_for_status_display(payload) + pending_composition_retries = collect_pending_composition_retry_projection( + runtime_root, args.goal_id + ) + if pending_composition_retries is not None: + payload["pending_composition_retry_receipts"] = ( + pending_composition_retries + ) except Exception as exc: payload = { "ok": False, diff --git a/loopx/control_plane/post_writeback_composition_retry.py b/loopx/control_plane/post_writeback_composition_retry.py index 1dec233188..2ff3b1343a 100644 --- a/loopx/control_plane/post_writeback_composition_retry.py +++ b/loopx/control_plane/post_writeback_composition_retry.py @@ -20,7 +20,19 @@ "source_projection_failed", "dispatch_failed", ) -_COMPOSITION_RETRY_JOURNAL_ROW_LIMIT = 512 +POST_WRITEBACK_COMPOSITION_RETRY_PROJECTION_SCHEMA_VERSION = ( + "loopx_post_writeback_composition_retry_projection_v0" +) +POST_WRITEBACK_COMPOSITION_RETRY_REPLAY_ACTION = ( + "Replay the committed CLI mutation that recorded this receipt with the " + "same goal/event/todo/turn/effect identity and state_version: the primary " + "writeback is idempotent, hook sidecars dedupe provider work, and a clean " + "projection composition settles the receipt." +) +# Journals fold back to one row per receipt id (under the append lock) once +# they cross this bound; reads scan every row so no unresolved receipt is +# ever silently dropped from the pending view. +_COMPOSITION_RETRY_JOURNAL_COMPACT_ROW_LIMIT = 512 def _now_iso() -> str: @@ -39,14 +51,40 @@ def composition_retry_receipt_log_path(runtime_root: Path, goal_id: str) -> Path ) +def _normalized_hook_set_digest( + hook_identities: Sequence[Mapping[str, str]], +) -> str: + """Render the order-insensitive hook set so identities stay comparable.""" + + pairs = sorted( + { + ( + str(item.get("hook_id") or "")[:200], + str(item.get("capability_id") or "")[:200], + ) + for item in hook_identities + if str(item.get("hook_id") or "") + } + ) + encoded = json.dumps(pairs, sort_keys=True, ensure_ascii=True).encode("utf-8") + return hashlib.sha256(encoded).hexdigest()[:16] + + def composition_retry_receipt_id( *, goal_id: str, event_kind: str, identity: Mapping[str, Any], state_version: str, + hook_identities: Sequence[Mapping[str, str]], ) -> str: - """Bind one receipt to the exact primary writeback it observes.""" + """Bind one receipt to the exact primary writeback and hook set it observes. + + The hook set is part of the identity: a composition that later succeeds + with a different registered hook set settles a different receipt (or no + receipt at all), so a replaced or removed hook can never mark the original + failure resolved. + """ stable = { "goal_id": str(goal_id or ""), @@ -56,6 +94,7 @@ def composition_retry_receipt_id( "turn_instance_id": str(identity.get("turn_instance_id") or ""), "effect_id": str(identity.get("effect_id") or ""), "state_version": str(state_version or ""), + "hook_set_digest": _normalized_hook_set_digest(hook_identities), } encoded = json.dumps(stable, sort_keys=True, ensure_ascii=True).encode("utf-8") return "pwcr_" + hashlib.sha256(encoded).hexdigest()[:16] @@ -97,9 +136,7 @@ def build_composition_retry_receipt( if not hook_id or hook_id in seen_hook_ids: continue seen_hook_ids.add(hook_id) - bounded_identities.append( - {"hook_id": hook_id, "capability_id": capability_id} - ) + bounded_identities.append({"hook_id": hook_id, "capability_id": capability_id}) return { "schema_version": POST_WRITEBACK_COMPOSITION_RETRY_RECEIPT_SCHEMA_VERSION, "receipt_id": composition_retry_receipt_id( @@ -107,6 +144,7 @@ def build_composition_retry_receipt( event_kind=event_kind, identity=identity, state_version=state_version, + hook_identities=bounded_identities, ), "status": status, "error_code": error_code, @@ -128,16 +166,23 @@ def build_composition_retry_receipt( def _iter_composition_retry_rows( - log_path: Path, *, row_limit: int = _COMPOSITION_RETRY_JOURNAL_ROW_LIMIT + log_path: Path, *, row_limit: int | None = None ) -> list[dict[str, Any]]: - """Read a bounded suffix of valid journal rows, oldest first.""" + """Read every valid journal row, oldest first. + + The read is untruncated on purpose: folding to the newest row per receipt + id must observe the latest state of *every* receipt, so a bounded suffix + that silently drops an unresolved row is never acceptable here. Storage + governance happens at append time via compaction instead. + """ try: lines = log_path.read_text(encoding="utf-8").splitlines() except OSError: return [] + selected = lines if row_limit is None else lines[-max(0, row_limit) :] rows: list[dict[str, Any]] = [] - for line in lines[-max(0, row_limit) :]: + for line in selected: text = line.strip() if not text: continue @@ -207,12 +252,55 @@ def append_composition_retry_receipt( current = _current_composition_retry_row(handle, receipt_id) if current is not None and current.get("status") == "settled": return current, False - handle.write( - json.dumps(payload, sort_keys=True, ensure_ascii=False) + "\n" - ) + handle.write(json.dumps(payload, sort_keys=True, ensure_ascii=False) + "\n") + _compact_composition_retry_journal(handle) return payload, True +def _compact_composition_retry_journal(handle: Any) -> None: + """Fold the journal back to one row per receipt id once it grows. + + Runs under the append lock, so compaction never races another writer. + Keeping only the newest row per receipt id preserves every receipt's + latest lifecycle state (the only fact readers consume) while bounding + storage on high-churn goals. + """ + + handle.seek(0) + lines = handle.read().splitlines() + if len(lines) <= _COMPOSITION_RETRY_JOURNAL_COMPACT_ROW_LIMIT: + handle.seek(0, 2) + return + latest: dict[str, dict[str, Any]] = {} + order: list[str] = [] + for line in lines: + text = line.strip() + if not text: + continue + try: + row = json.loads(text) + except json.JSONDecodeError: + continue + if ( + not isinstance(row, dict) + or row.get("schema_version") + != POST_WRITEBACK_COMPOSITION_RETRY_RECEIPT_SCHEMA_VERSION + or not isinstance(row.get("receipt_id"), str) + or row.get("status") not in {"retryable", "settled"} + ): + continue + receipt_id = str(row["receipt_id"]) + if receipt_id not in latest: + order.append(receipt_id) + latest[receipt_id] = row + handle.seek(0) + handle.truncate() + for receipt_id in order: + handle.write( + json.dumps(latest[receipt_id], sort_keys=True, ensure_ascii=False) + "\n" + ) + + def settle_composition_retry_receipt( log_path: Path, *, @@ -246,9 +334,7 @@ def settle_composition_retry_receipt( if current is None: return {}, False handle.seek(0, 2) - handle.write( - json.dumps(settled, sort_keys=True, ensure_ascii=False) + "\n" - ) + handle.write(json.dumps(settled, sort_keys=True, ensure_ascii=False) + "\n") return settled, True @@ -257,20 +343,9 @@ def pending_composition_retry_receipts( ) -> list[dict[str, Any]]: """Read unconsumed retryable receipts for one goal, newest row per receipt.""" - rows = _iter_composition_retry_rows( + return pending_composition_retry_receipts_for_path( composition_retry_receipt_log_path(runtime_root, goal_id) ) - latest: dict[str, dict[str, Any]] = {} - for row in rows: - latest[str(row.get("receipt_id"))] = row - return sorted( - ( - dict(row) - for row in latest.values() - if row.get("status") == "retryable" - ), - key=lambda row: str(row.get("receipt_id") or ""), - ) __all__ = [ @@ -283,6 +358,68 @@ def pending_composition_retry_receipts( "composition_retry_receipt_id", "composition_retry_receipt_log_path", "composition_retry_receipt_ref", + "collect_pending_composition_retry_projection", "pending_composition_retry_receipts", + "pending_composition_retry_receipts_for_path", "settle_composition_retry_receipt", ] + + +def pending_composition_retry_receipts_for_path( + journal_path: Path, +) -> list[dict[str, Any]]: + """Read unconsumed retryable receipts from one journal path.""" + + latest: dict[str, dict[str, Any]] = {} + for row in _iter_composition_retry_rows(journal_path): + latest[str(row.get("receipt_id"))] = row + return sorted( + (dict(row) for row in latest.values() if row.get("status") == "retryable"), + key=lambda row: str(row.get("receipt_id") or ""), + ) + + +def collect_pending_composition_retry_projection( + runtime_root: Path, + goal_id: str | None, + *, + max_items: int = 20, +) -> dict[str, Any] | None: + """Collect pending composition retry receipts for status/doctor readback. + + Returns None when nothing is pending so the status output stays unchanged + on healthy goals; otherwise returns a bounded public-safe projection with + the replay guidance, letting a later turn or operator discover and clear + outstanding composition retries through the normal read model. + """ + + root = runtime_root.expanduser() + goals_root = root / "goals" + if goal_id: + journal_paths = [composition_retry_receipt_log_path(root, goal_id)] + elif goals_root.is_dir(): + journal_paths = sorted(path for path in goals_root.iterdir() if path.is_dir()) + journal_paths = [ + path / "post_writeback_hooks" / POST_WRITEBACK_COMPOSITION_RETRY_LOG_NAME + for path in journal_paths + ] + else: + journal_paths = [] + pending: list[dict[str, Any]] = [] + for journal_path in journal_paths: + if not journal_path.is_file(): + continue + for row in pending_composition_retry_receipts_for_path(journal_path): + pending.append(row) + if len(pending) >= max_items: + break + if len(pending) >= max_items: + break + if not pending: + return None + return { + "schema_version": POST_WRITEBACK_COMPOSITION_RETRY_PROJECTION_SCHEMA_VERSION, + "pending_count": len(pending), + "pending": pending, + "replay_action": POST_WRITEBACK_COMPOSITION_RETRY_REPLAY_ACTION, + } diff --git a/tests/control_plane/test_post_writeback_composition_retry.py b/tests/control_plane/test_post_writeback_composition_retry.py index fddc21214e..cf948dd1fb 100644 --- a/tests/control_plane/test_post_writeback_composition_retry.py +++ b/tests/control_plane/test_post_writeback_composition_retry.py @@ -14,9 +14,12 @@ PostWritebackHookRegistration, ) from loopx.control_plane.post_writeback_composition_retry import ( + _COMPOSITION_RETRY_JOURNAL_COMPACT_ROW_LIMIT, + POST_WRITEBACK_COMPOSITION_RETRY_PROJECTION_SCHEMA_VERSION, POST_WRITEBACK_COMPOSITION_RETRY_RECEIPT_SCHEMA_VERSION, append_composition_retry_receipt, build_composition_retry_receipt, + collect_pending_composition_retry_projection, composition_retry_receipt_id, composition_retry_receipt_log_path, composition_retry_receipt_ref, @@ -50,9 +53,7 @@ def _identity() -> dict[str, str]: } -def _hook( - *, producer_calls: list[int] | None = None -) -> PostWritebackHookRegistration: +def _hook(*, producer_calls: list[int] | None = None) -> PostWritebackHookRegistration: def producer(value: object) -> dict[str, object]: if producer_calls is not None: producer_calls.append(1) @@ -145,9 +146,7 @@ def failing_builder(**_kwargs: object) -> dict[str, object]: assert failure["hook_id"] == "periodic_report.stage_completion" assert failure["capability_id"] == "periodic-report" assert failure["error_code"] == "source_projection_failed" - assert failure["durable_receipt_ref"].startswith( - "post-writeback-composition:pwcr_" - ) + assert failure["durable_receipt_ref"].startswith("post-writeback-composition:pwcr_") pending = pending_composition_retry_receipts(runtime_root, "goal-1") (receipt,) = pending @@ -188,34 +187,31 @@ def flaky_builder(**_kwargs: object) -> dict[str, object]: hooks = (_hook(producer_calls=producer_calls),) - failed = _dispatch( - registry_path, hooks=hooks, projection_builder=flaky_builder - ) + failed = _dispatch(registry_path, hooks=hooks, projection_builder=flaky_builder) assert failed["failures"][0]["error_code"] == "source_projection_failed" - assert [receipt["status"] for receipt in pending_composition_retry_receipts(runtime_root, "goal-1")] == ["retryable"] + assert [ + receipt["status"] + for receipt in pending_composition_retry_receipts(runtime_root, "goal-1") + ] == ["retryable"] - recovered = _dispatch( - registry_path, hooks=hooks, projection_builder=flaky_builder - ) + recovered = _dispatch(registry_path, hooks=hooks, projection_builder=flaky_builder) assert recovered["failures"] == [] assert recovered["intent_count"] == 1 assert len(producer_calls) == 1 assert len(projection_calls) == 2 assert pending_composition_retry_receipts(runtime_root, "goal-1") == [] rows = _journal_rows(runtime_root) - assert [row["status"] for row in rows if row["receipt_id"] == rows[-1]["receipt_id"]][-1] == "settled" + assert [ + row["status"] for row in rows if row["receipt_id"] == rows[-1]["receipt_id"] + ][-1] == "settled" - replayed = _dispatch( - registry_path, hooks=hooks, projection_builder=flaky_builder - ) + replayed = _dispatch(registry_path, hooks=hooks, projection_builder=flaky_builder) assert replayed["intent_count"] == 1 assert replayed["replayed_hooks"] == ["periodic_report.stage_completion"] assert len(producer_calls) == 1 assert len(projection_calls) == 3 assert pending_composition_retry_receipts(runtime_root, "goal-1") == [] - settled_ids = { - row["receipt_id"] for row in _journal_rows(runtime_root) - } + settled_ids = {row["receipt_id"] for row in _journal_rows(runtime_root)} assert len(settled_ids) == 1 @@ -242,9 +238,7 @@ def flaky_builder(**_kwargs: object) -> dict[str, object]: second_failure = _dispatch( registry_path, hooks=hooks, projection_builder=flaky_builder ) - recovered = _dispatch( - registry_path, hooks=hooks, projection_builder=flaky_builder - ) + recovered = _dispatch(registry_path, hooks=hooks, projection_builder=flaky_builder) for result in (first_failure, second_failure, recovered): assert result["primary_writeback_preserved"] is True @@ -260,11 +254,18 @@ def flaky_builder(**_kwargs: object) -> dict[str, object]: def test_composition_retry_receipt_id_binds_primary_writeback_identity() -> None: + hooks = [ + { + "hook_id": "periodic_report.stage_completion", + "capability_id": "periodic-report", + } + ] base = composition_retry_receipt_id( goal_id="goal-1", event_kind="todo_complete", identity=_identity(), state_version="vision-revision-2", + hook_identities=hooks, ) assert base.startswith("pwcr_") assert base == composition_retry_receipt_id( @@ -272,30 +273,42 @@ def test_composition_retry_receipt_id_binds_primary_writeback_identity() -> None event_kind="todo_complete", identity=_identity(), state_version="vision-revision-2", + hook_identities=list(reversed(hooks)), ) assert base != composition_retry_receipt_id( goal_id="goal-1", event_kind="todo_complete", identity=_identity(), state_version="vision-revision-3", + hook_identities=hooks, ) assert base != composition_retry_receipt_id( goal_id="goal-2", event_kind="todo_complete", identity=_identity(), state_version="vision-revision-2", + hook_identities=hooks, ) - assert composition_retry_receipt_ref(base) == ( - f"post-writeback-composition:{base}" + assert base != composition_retry_receipt_id( + goal_id="goal-1", + event_kind="todo_complete", + identity=_identity(), + state_version="vision-revision-2", + hook_identities=[], ) + assert composition_retry_receipt_ref(base) == (f"post-writeback-composition:{base}") def test_composition_retry_journal_append_is_idempotent_and_terminal( tmp_path: Path, ) -> None: - journal_path = tmp_path / "goals" / "goal-1" / ( - "post_writeback_hooks" - ) / "composition-retry-receipts.jsonl" + journal_path = ( + tmp_path + / "goals" + / "goal-1" + / ("post_writeback_hooks") + / "composition-retry-receipts.jsonl" + ) receipt = build_composition_retry_receipt( goal_id="goal-1", event_kind="todo_complete", @@ -303,7 +316,10 @@ def test_composition_retry_journal_append_is_idempotent_and_terminal( state_version="vision-revision-2", committed_at="2026-09-06T00:00:00Z", hook_identities=[ - {"hook_id": "periodic_report.stage_completion", "capability_id": "periodic-report"} + { + "hook_id": "periodic_report.stage_completion", + "capability_id": "periodic-report", + } ], error_code="source_projection_failed", ) @@ -325,16 +341,17 @@ def test_composition_retry_journal_append_is_idempotent_and_terminal( state_version="vision-revision-2", committed_at="2026-09-06T00:00:00Z", hook_identities=[ - {"hook_id": "periodic_report.stage_completion", "capability_id": "periodic-report"} + { + "hook_id": "periodic_report.stage_completion", + "capability_id": "periodic-report", + } ], ) assert settled_flag is True assert settled["status"] == "settled" assert settled["error_code"] is None - regressed, regressed_flag = append_composition_retry_receipt( - journal_path, receipt - ) + regressed, regressed_flag = append_composition_retry_receipt(journal_path, receipt) assert regressed_flag is False assert regressed["status"] == "settled" rows = [ @@ -351,12 +368,34 @@ def test_composition_retry_journal_append_is_idempotent_and_terminal( identity=_identity(), state_version="vision-revision-2", committed_at="2026-09-06T00:00:00Z", - hook_identities=[], + hook_identities=[ + { + "hook_id": "periodic_report.stage_completion", + "capability_id": "periodic-report", + } + ], ) assert resettle_flag is False assert resettle["status"] == "settled" assert len(rows) == 3 + # A different registered hook set computes a different receipt identity, + # so it must not touch (let alone settle) the original failure's receipt. + foreign, foreign_flag = settle_composition_retry_receipt( + journal_path, + goal_id="goal-1", + event_kind="todo_complete", + identity=_identity(), + state_version="vision-revision-2", + committed_at="2026-09-06T00:00:00Z", + hook_identities=[ + {"hook_id": "other.hook", "capability_id": "other-capability"} + ], + ) + assert foreign_flag is False + assert foreign == {} + assert len(rows) == 3 + def test_settle_without_pending_receipt_is_a_noop(tmp_path: Path) -> None: journal_path = composition_retry_receipt_log_path(tmp_path, "goal-1") @@ -437,9 +476,7 @@ def broken_append(*_args: object, **_kwargs: object) -> None: assert failure["capability_id"] == "periodic-report" assert failure["error_code"] == "source_projection_failed" assert "durable_receipt_ref" not in failure - assert not composition_retry_receipt_log_path( - runtime_root, "goal-1" - ).exists() + assert not composition_retry_receipt_log_path(runtime_root, "goal-1").exists() def test_producer_failure_settles_composition_receipt_with_hook_level_trail( @@ -488,3 +525,175 @@ def hook_with_broken_producer() -> PostWritebackHookRegistration: "a composed projection must settle the composition receipt even when a " "hook-level producer fails: hook failures carry their own durable trail" ) + + +def test_foreign_hook_set_cannot_settle_original_failure_receipt( + tmp_path: Path, +) -> None: + """P1 probe: a changed hook set must not resolve the original failure.""" + + registry_path, runtime_root = _write_registry(tmp_path) + + def failing_builder(**_kwargs: object) -> dict[str, object]: + raise RuntimeError("transient projection failure") + + first = _dispatch( + registry_path, + hooks=(_hook(),), + projection_builder=failing_builder, + ) + assert first["failures"][0]["hook_id"] == "periodic_report.stage_completion" + pending = pending_composition_retry_receipts(runtime_root, "goal-1") + assert len(pending) == 1 + + def other_hook() -> PostWritebackHookRegistration: + return PostWritebackHookRegistration( + hook_id="periodic_report.other_stage", + capability_id="periodic-report", + event_kinds=("todo_complete",), + intent_kinds=("periodic_report.trigger_evaluation",), + requested_read_scope=("stage_completion",), + producer=lambda value: { + "schema_version": POST_WRITEBACK_HOOK_RESULT_SCHEMA_VERSION, + "hook_id": "periodic_report.other_stage", + "capability_id": "periodic-report", + "phase": "post_writeback", + "status": "not_applicable", + "intent": None, + "error_code": None, + }, + ) + + def other_producer_fails(value: object) -> dict[str, object]: + raise RuntimeError("hook producer collapsed") + + def other_registration() -> PostWritebackHookRegistration: + registration = other_hook() + object.__setattr__(registration, "producer", other_producer_fails) + return registration + + second = _dispatch( + registry_path, + hooks=(other_registration(),), + projection_builder=lambda **_kwargs: _stage_projection(), + ) + assert second["failures"][0]["hook_id"] == "periodic_report.other_stage" + still_pending = pending_composition_retry_receipts(runtime_root, "goal-1") + assert [row["receipt_id"] for row in still_pending] == [ + row["receipt_id"] for row in pending + ], "the original hook-a receipt must survive a foreign hook-b composition" + + +def test_pending_view_keeps_unresolved_receipts_beyond_the_compaction_bound( + tmp_path: Path, +) -> None: + """P1 probe: bounded suffix reads must never hide an unresolved receipt.""" + + registry_path, runtime_root = _write_registry(tmp_path) + + def failing_builder(**_kwargs: object) -> dict[str, object]: + raise RuntimeError("transient projection failure") + + assert _dispatch( + registry_path, hooks=(_hook(),), projection_builder=failing_builder + )["failures"] + original = pending_composition_retry_receipts(runtime_root, "goal-1") + assert len(original) == 1 + journal_path = composition_retry_receipt_log_path(runtime_root, "goal-1") + + filler_identity = dict(_identity()) + for index in range(_COMPOSITION_RETRY_JOURNAL_COMPACT_ROW_LIMIT + 8): + filler_identity["todo_id"] = f"todo_filler_{index:04d}" + append_composition_retry_receipt( + journal_path, + build_composition_retry_receipt( + goal_id="goal-1", + event_kind="todo_complete", + identity=filler_identity, + state_version=f"vision-revision-{index}", + committed_at="2026-09-06T00:00:00Z", + hook_identities=[ + { + "hook_id": "periodic_report.stage_completion", + "capability_id": "periodic-report", + } + ], + error_code="source_projection_failed", + ), + ) + + after = pending_composition_retry_receipts(runtime_root, "goal-1") + assert original[0]["receipt_id"] in {row["receipt_id"] for row in after} + assert len(after) == _COMPOSITION_RETRY_JOURNAL_COMPACT_ROW_LIMIT + 9 + + # Distinct unresolved receipts are all durable state, so compaction must + # not collapse them; re-observing one receipt, however, folds its rows. + filler_identity["todo_id"] = "todo_filler_0000" + append_composition_retry_receipt( + journal_path, + build_composition_retry_receipt( + goal_id="goal-1", + event_kind="todo_complete", + identity=filler_identity, + state_version="vision-revision-0", + committed_at="2026-09-06T00:00:00Z", + hook_identities=[ + { + "hook_id": "periodic_report.stage_completion", + "capability_id": "periodic-report", + } + ], + error_code="source_projection_failed", + ), + ) + rows = [ + json.loads(line) + for line in journal_path.read_text(encoding="utf-8").splitlines() + if line.strip() + ] + ids = [row["receipt_id"] for row in rows] + assert len(ids) == len(set(ids)) == len(after) + still_visible = pending_composition_retry_receipts(runtime_root, "goal-1") + assert original[0]["receipt_id"] in {row["receipt_id"] for row in still_visible} + assert len(still_visible) == len(after) + + +def test_later_turn_discovers_and_clears_pending_composition_retries( + tmp_path: Path, +) -> None: + """P1 probe: the read model surfaces pending retries and replay clears them.""" + + registry_path, runtime_root = _write_registry(tmp_path) + + def failing_builder(**_kwargs: object) -> dict[str, object]: + raise RuntimeError("transient projection failure") + + assert _dispatch( + registry_path, hooks=(_hook(),), projection_builder=failing_builder + )["failures"] + + # A later turn (fresh process state) discovers the pending retry through + # the status readback projection instead of the original command response. + discovered = collect_pending_composition_retry_projection(runtime_root, None) + assert discovered is not None + assert discovered["pending_count"] == 1 + assert discovered["pending"][0]["identity"]["goal_id"] == "goal-1" + assert discovered["schema_version"] == ( + POST_WRITEBACK_COMPOSITION_RETRY_PROJECTION_SCHEMA_VERSION + ) + assert "Replay the committed CLI mutation" in discovered["replay_action"] + healthy = collect_pending_composition_retry_projection( + tmp_path / "missing-runtime", None + ) + assert healthy is None + + # Replaying the same committed mutation composes cleanly and settles the + # receipt; the next readback no longer reports anything pending. + replay = _dispatch( + registry_path, + hooks=(_hook(),), + projection_builder=lambda **_kwargs: _stage_projection(), + ) + assert replay["intent_count"] == 1 + assert collect_pending_composition_retry_projection(runtime_root, None) is None + assert pending_composition_retry_receipts(runtime_root, "goal-1") == [] From 9305d9a03e739924812a66aa684ae1763029ab78 Mon Sep 17 00:00:00 2001 From: now-ing Date: Sun, 6 Sep 2026 19:45:59 +0800 Subject: [PATCH 4/7] fix(post-writeback): separate composition receipts by hook policy version The hook-set digest keyed only (hook_id, capability_id), so replacing a hook adapter with a new policy version (the manifest-discovered adapter shape) computed the same receipt identity and could settle a failure the replacement never actually resolved. The digest and the receipt's hook rows now carry policy_version, and a version-replacement regression pins that only the original version's composition settles its own receipt. Signed-off-by: now-ing --- loopx/cli_commands/post_writeback.py | 11 +++-- .../post_writeback_composition_retry.py | 10 ++++- .../test_post_writeback_composition_retry.py | 40 +++++++++++++++++++ 3 files changed, 56 insertions(+), 5 deletions(-) diff --git a/loopx/cli_commands/post_writeback.py b/loopx/cli_commands/post_writeback.py index eb19f0354c..730f272e3a 100644 --- a/loopx/cli_commands/post_writeback.py +++ b/loopx/cli_commands/post_writeback.py @@ -23,14 +23,17 @@ PostWritebackProjectionBuilder = Callable[..., Mapping[str, object]] -_COMPOSITION_HOOK_FIELDS = ("hook_id", "capability_id") - - def _composition_hook_identities( hooks: Sequence[PostWritebackHookRegistration], ) -> list[dict[str, str]]: return [ - {field: str(getattr(registration, field) or "") for field in _COMPOSITION_HOOK_FIELDS} + { + "hook_id": str(registration.hook_id or ""), + "capability_id": str(registration.capability_id or ""), + "policy_version": str( + getattr(registration, "policy_version", "") or "" + ), + } for registration in hooks ] diff --git a/loopx/control_plane/post_writeback_composition_retry.py b/loopx/control_plane/post_writeback_composition_retry.py index 2ff3b1343a..a056a78bdf 100644 --- a/loopx/control_plane/post_writeback_composition_retry.py +++ b/loopx/control_plane/post_writeback_composition_retry.py @@ -61,6 +61,7 @@ def _normalized_hook_set_digest( ( str(item.get("hook_id") or "")[:200], str(item.get("capability_id") or "")[:200], + str(item.get("policy_version") or "")[:64], ) for item in hook_identities if str(item.get("hook_id") or "") @@ -133,10 +134,17 @@ def build_composition_retry_receipt( for raw_identity in hook_identities: hook_id = str(raw_identity.get("hook_id") or "")[:200] capability_id = str(raw_identity.get("capability_id") or "")[:200] + policy_version = str(raw_identity.get("policy_version") or "")[:64] if not hook_id or hook_id in seen_hook_ids: continue seen_hook_ids.add(hook_id) - bounded_identities.append({"hook_id": hook_id, "capability_id": capability_id}) + bounded_identities.append( + { + "hook_id": hook_id, + "capability_id": capability_id, + "policy_version": policy_version, + } + ) return { "schema_version": POST_WRITEBACK_COMPOSITION_RETRY_RECEIPT_SCHEMA_VERSION, "receipt_id": composition_retry_receipt_id( diff --git a/tests/control_plane/test_post_writeback_composition_retry.py b/tests/control_plane/test_post_writeback_composition_retry.py index cf948dd1fb..be913a59ab 100644 --- a/tests/control_plane/test_post_writeback_composition_retry.py +++ b/tests/control_plane/test_post_writeback_composition_retry.py @@ -164,6 +164,7 @@ def failing_builder(**_kwargs: object) -> dict[str, object]: { "hook_id": "periodic_report.stage_completion", "capability_id": "periodic-report", + "policy_version": "v0", } ] assert receipt["primary_writeback_preserved"] is True @@ -697,3 +698,42 @@ def failing_builder(**_kwargs: object) -> dict[str, object]: assert replay["intent_count"] == 1 assert collect_pending_composition_retry_projection(runtime_root, None) is None assert pending_composition_retry_receipts(runtime_root, "goal-1") == [] + + +def test_receipt_identity_separates_policy_versions(tmp_path: Path) -> None: + """A policy-version replacement must not settle the original failure.""" + + registry_path, runtime_root = _write_registry(tmp_path) + + def failing_builder(**_kwargs: object) -> dict[str, object]: + raise RuntimeError("transient projection failure") + + assert _dispatch( + registry_path, hooks=(_hook(),), projection_builder=failing_builder + )["failures"] + + def upgraded_hook() -> PostWritebackHookRegistration: + registration = _hook() + object.__setattr__(registration, "policy_version", "v1") + return registration + + upgraded, upgraded_flag = settle_composition_retry_receipt( + composition_retry_receipt_log_path(runtime_root, "goal-1"), + goal_id="goal-1", + event_kind="todo_complete", + identity=_identity(), + state_version="2026-09-06T00:00:00Z", + committed_at="2026-09-06T00:00:00Z", + hook_identities=[ + { + "hook_id": "periodic_report.stage_completion", + "capability_id": "periodic-report", + "policy_version": "v1", + } + ], + ) + assert upgraded_flag is False + assert upgraded == {} + pending = pending_composition_retry_receipts(runtime_root, "goal-1") + assert len(pending) == 1 + assert pending[0]["hooks"][0]["policy_version"] == "v0" From c7efbbeb25754c8e53192921b70a9086cfaa03b4 Mon Sep 17 00:00:00 2001 From: now-ing Date: Sun, 6 Sep 2026 23:50:12 +0800 Subject: [PATCH 5/7] test(post-writeback): settle idempotency after a policy-version replacement Signed-off-by: now-ing --- .../test_post_writeback_composition_retry.py | 21 +++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/tests/control_plane/test_post_writeback_composition_retry.py b/tests/control_plane/test_post_writeback_composition_retry.py index be913a59ab..a51d1ed509 100644 --- a/tests/control_plane/test_post_writeback_composition_retry.py +++ b/tests/control_plane/test_post_writeback_composition_retry.py @@ -737,3 +737,24 @@ def upgraded_hook() -> PostWritebackHookRegistration: pending = pending_composition_retry_receipts(runtime_root, "goal-1") assert len(pending) == 1 assert pending[0]["hooks"][0]["policy_version"] == "v0" + + # The original version's own composition still settles its receipt + # idempotently after a replacement tried and failed to claim it. + settled, settled_flag = settle_composition_retry_receipt( + composition_retry_receipt_log_path(runtime_root, "goal-1"), + goal_id="goal-1", + event_kind="todo_complete", + identity=_identity(), + state_version="2026-09-06T00:00:00Z", + committed_at="2026-09-06T00:00:00Z", + hook_identities=[ + { + "hook_id": "periodic_report.stage_completion", + "capability_id": "periodic-report", + "policy_version": "v0", + } + ], + ) + assert settled_flag is True + assert settled["status"] == "settled" + assert pending_composition_retry_receipts(runtime_root, "goal-1") == [] From 731ff3574aec77b7d9d0437fe293f8381c364a28 Mon Sep 17 00:00:00 2001 From: now-ing Date: Mon, 7 Sep 2026 06:03:01 +0800 Subject: [PATCH 6/7] fix(control-plane): compact composition retry journals atomically In-place journal compaction truncated the only durable copy before the folded replacement existed, so a crash or I/O failure between truncate and rewrite lost every pending receipt (owner probe: 512 pending -> 0 after an OSError injected right after the real truncate). Materialize the folded journal in a sibling temporary file, fsync and verify it, then swap it in with one atomic replace: the failure propagates, the previous journal survives untouched, and lock-free readers only ever observe a complete old or complete folded file. Reuse the existing NamedTemporaryFile + os.replace atomic-write boundary under the append lock; add regressions for interrupted compaction and concurrent readers (red on the in-place truncation, green here). Signed-off-by: now-ing --- .../post_writeback_composition_retry.py | 52 +++-- .../test_post_writeback_composition_retry.py | 184 ++++++++++++++++++ 2 files changed, 225 insertions(+), 11 deletions(-) diff --git a/loopx/control_plane/post_writeback_composition_retry.py b/loopx/control_plane/post_writeback_composition_retry.py index a056a78bdf..0f0fb9df1b 100644 --- a/loopx/control_plane/post_writeback_composition_retry.py +++ b/loopx/control_plane/post_writeback_composition_retry.py @@ -2,6 +2,8 @@ import hashlib import json +import os +import tempfile from collections.abc import Mapping, Sequence from datetime import datetime, timezone from pathlib import Path @@ -261,19 +263,24 @@ def append_composition_retry_receipt( if current is not None and current.get("status") == "settled": return current, False handle.write(json.dumps(payload, sort_keys=True, ensure_ascii=False) + "\n") - _compact_composition_retry_journal(handle) + _compact_composition_retry_journal(log_path, handle) return payload, True -def _compact_composition_retry_journal(handle: Any) -> None: +def _compact_composition_retry_journal(log_path: Path, handle: Any) -> None: """Fold the journal back to one row per receipt id once it grows. Runs under the append lock, so compaction never races another writer. - Keeping only the newest row per receipt id preserves every receipt's - latest lifecycle state (the only fact readers consume) while bounding - storage on high-churn goals. + The folded replacement is materialized, synced, and verified in a + sibling temporary file before one atomic replace swaps it in: a + process exit or I/O failure at any point leaves the previous journal + fully intact instead of truncating the only durable copy, and + lock-free readers only ever observe the complete old or the complete + folded file. Compaction failures propagate; the journal is never + recovered by swallowing the error or truncating again. """ + handle.flush() handle.seek(0) lines = handle.read().splitlines() if len(lines) <= _COMPOSITION_RETRY_JOURNAL_COMPACT_ROW_LIMIT: @@ -301,12 +308,35 @@ def _compact_composition_retry_journal(handle: Any) -> None: if receipt_id not in latest: order.append(receipt_id) latest[receipt_id] = row - handle.seek(0) - handle.truncate() - for receipt_id in order: - handle.write( - json.dumps(latest[receipt_id], sort_keys=True, ensure_ascii=False) + "\n" - ) + replacement = "".join( + json.dumps(latest[receipt_id], sort_keys=True, ensure_ascii=False) + "\n" + for receipt_id in order + ) + temporary_name = "" + try: + with tempfile.NamedTemporaryFile( + "w", + encoding="utf-8", + dir=log_path.parent, + prefix=f"{log_path.name}.", + suffix=".tmp", + delete=False, + ) as temporary_handle: + temporary_name = temporary_handle.name + temporary_handle.write(replacement) + temporary_handle.flush() + os.fsync(temporary_handle.fileno()) + verified = _iter_composition_retry_rows(Path(temporary_name)) + if [str(row.get("receipt_id")) for row in verified] != order: + raise OSError("composition retry journal replacement is unverifiable") + os.replace(temporary_name, log_path) + except BaseException: + if temporary_name: + try: + os.unlink(temporary_name) + except OSError: + pass + raise def settle_composition_retry_receipt( diff --git a/tests/control_plane/test_post_writeback_composition_retry.py b/tests/control_plane/test_post_writeback_composition_retry.py index a51d1ed509..692d0e4b51 100644 --- a/tests/control_plane/test_post_writeback_composition_retry.py +++ b/tests/control_plane/test_post_writeback_composition_retry.py @@ -1,6 +1,8 @@ from __future__ import annotations import json +import os +import threading from pathlib import Path from typing import Any @@ -24,6 +26,7 @@ composition_retry_receipt_log_path, composition_retry_receipt_ref, pending_composition_retry_receipts, + pending_composition_retry_receipts_for_path, settle_composition_retry_receipt, ) @@ -659,6 +662,187 @@ def failing_builder(**_kwargs: object) -> dict[str, object]: assert len(still_visible) == len(after) +def _filler_append(journal_path: Path, index: int) -> None: + identity = _identity() + identity["todo_id"] = f"todo_filler_{index:04d}" + append_composition_retry_receipt( + journal_path, + build_composition_retry_receipt( + goal_id="goal-1", + event_kind="todo_complete", + identity=identity, + state_version=f"vision-revision-{index}", + committed_at="2026-09-06T00:00:00Z", + hook_identities=[ + { + "hook_id": "periodic_report.stage_completion", + "capability_id": "periodic-report", + } + ], + error_code="source_projection_failed", + ), + ) + + +def test_failed_compaction_keeps_the_journal_intact( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """P1 regression: interrupted compaction must never lose pending receipts. + + Owner fault injection: 512 pending receipts, the next append triggers + compaction, and an OSError lands right after the destructive in-place + step. The legacy truncate-then-rewrite path left a 0-byte journal and + every unhandled failure vanished from the pending readback; the folded + replacement must be swapped in atomically so the failure propagates + while the pre-compaction journal stays fully readable. + """ + + registry_path, runtime_root = _write_registry(tmp_path) + journal_path = composition_retry_receipt_log_path(runtime_root, "goal-1") + + for index in range(_COMPOSITION_RETRY_JOURNAL_COMPACT_ROW_LIMIT): + _filler_append(journal_path, index) + assert ( + len(pending_composition_retry_receipts(runtime_root, "goal-1")) + == _COMPOSITION_RETRY_JOURNAL_COMPACT_ROW_LIMIT + ) + + real_path_open = Path.open + + class _JournalHandle: + def __init__(self, handle: Any) -> None: + self._handle = handle + + def __getattr__(self, name: str) -> Any: + return getattr(self._handle, name) + + def __iter__(self) -> Any: + return iter(self._handle) + + def __enter__(self) -> _JournalHandle: + self._handle.__enter__() + return self + + def __exit__(self, *exc: object) -> None: + self._handle.__exit__(*exc) # type: ignore[func-returns-value] + + def truncate(self, *args: object, **kwargs: object) -> object: + self._handle.truncate(*args, **kwargs) + raise OSError("injected after journal truncation") + + def exploding_journal_open(self: Path, *args: object, **kwargs: object) -> object: + handle = real_path_open(self, *args, **kwargs) # type: ignore[arg-type] + mode = str(args[0] if args else kwargs.get("mode", "r")) + if self == journal_path and "a" in mode: + return _JournalHandle(handle) + return handle + + monkeypatch.setattr(Path, "open", exploding_journal_open) + monkeypatch.setattr( + os, + "replace", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + OSError("injected at journal replacement") + ), + ) + + with pytest.raises(OSError, match="injected"): + _filler_append(journal_path, _COMPOSITION_RETRY_JOURNAL_COMPACT_ROW_LIMIT) + + surviving = pending_composition_retry_receipts(runtime_root, "goal-1") + assert len(surviving) == _COMPOSITION_RETRY_JOURNAL_COMPACT_ROW_LIMIT + 1 + journal_lines = [ + line + for line in journal_path.read_text(encoding="utf-8").splitlines() + if line.strip() + ] + assert len(journal_lines) == _COMPOSITION_RETRY_JOURNAL_COMPACT_ROW_LIMIT + 1 + assert not list(journal_path.parent.glob("*.tmp")) + + +def test_concurrent_reader_never_observes_a_partial_journal( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """P1 regression: readers only ever see the complete old or folded file.""" + + registry_path, runtime_root = _write_registry(tmp_path) + journal_path = composition_retry_receipt_log_path(runtime_root, "goal-1") + + bound = _COMPOSITION_RETRY_JOURNAL_COMPACT_ROW_LIMIT + for _round in range(2): + for index in range(bound): + _filler_append(journal_path, index) + + real_replace = os.replace + swap_observations: dict[str, int] = {} + + def observing_replace(src: object, dst: object, *args: object) -> object: + destination = Path(str(dst)) + swap_observations["before"] = len( + [ + line + for line in destination.read_text(encoding="utf-8").splitlines() + if line.strip() + ] + ) + result = real_replace(src, dst, *args) + swap_observations["after"] = len( + [ + line + for line in destination.read_text(encoding="utf-8").splitlines() + if line.strip() + ] + ) + return result + + monkeypatch.setattr(os, "replace", observing_replace) + + _filler_append(journal_path, 0) + + # The swap point observes the complete unfolded journal (the second + # append round already folds at every row, so the journal holds one + # extra row per swap) and the complete folded replacement. + assert swap_observations == {"before": bound + 1, "after": bound} + pending = pending_composition_retry_receipts(runtime_root, "goal-1") + assert len(pending) == bound + + +def test_live_readers_during_repeated_compaction_see_complete_journals( + tmp_path: Path, +) -> None: + """P1 regression: lock-free readers never observe a truncated journal.""" + + registry_path, runtime_root = _write_registry(tmp_path) + journal_path = composition_retry_receipt_log_path(runtime_root, "goal-1") + + bound = _COMPOSITION_RETRY_JOURNAL_COMPACT_ROW_LIMIT + for _round in range(2): + for index in range(bound): + _filler_append(journal_path, index) + + stop = threading.Event() + observations: list[int] = [] + + def reader() -> None: + while not stop.is_set(): + observations.append( + len(pending_composition_retry_receipts_for_path(journal_path)) + ) + + reader_thread = threading.Thread(target=reader) + reader_thread.start() + try: + for index in range(32): + _filler_append(journal_path, index) + finally: + stop.set() + reader_thread.join(timeout=10.0) + assert not reader_thread.is_alive() + + assert observations + assert set(observations) == {bound} + + def test_later_turn_discovers_and_clears_pending_composition_retries( tmp_path: Path, ) -> None: From 47f65fe695653c0d5b0d86d36b0d3dc25ee8635e Mon Sep 17 00:00:00 2001 From: now-ing Date: Mon, 7 Sep 2026 07:24:27 +0800 Subject: [PATCH 7/7] fix(status): scope pending composition retries to the requested agent The pending composition-retry projection ignored the status command's agent selector, so one agent's --agent-id status answer surfaced other peers' pending retry receipts with generic replay guidance, and peer rows could fill the display cap before the agent's own work. The collector now filters by the receipt's identity agent before counting and truncating; a selector with no pending work stays absent from the output, and the global view keeps every agent's rows. Signed-off-by: now-ing --- loopx/cli_commands/status.py | 2 +- .../post_writeback_composition_retry.py | 21 ++++-- .../test_post_writeback_composition_retry.py | 71 +++++++++++++++++++ 3 files changed, 87 insertions(+), 7 deletions(-) diff --git a/loopx/cli_commands/status.py b/loopx/cli_commands/status.py index 733649b2e3..169805be72 100644 --- a/loopx/cli_commands/status.py +++ b/loopx/cli_commands/status.py @@ -236,7 +236,7 @@ def handle_status_command( ) compact_agent_lane_todo_index_for_status_display(payload) pending_composition_retries = collect_pending_composition_retry_projection( - runtime_root, args.goal_id + runtime_root, args.goal_id, agent_id=args.agent_id ) if pending_composition_retries is not None: payload["pending_composition_retry_receipts"] = ( diff --git a/loopx/control_plane/post_writeback_composition_retry.py b/loopx/control_plane/post_writeback_composition_retry.py index 0f0fb9df1b..c83087668d 100644 --- a/loopx/control_plane/post_writeback_composition_retry.py +++ b/loopx/control_plane/post_writeback_composition_retry.py @@ -421,6 +421,7 @@ def collect_pending_composition_retry_projection( runtime_root: Path, goal_id: str | None, *, + agent_id: str | None = None, max_items: int = 20, ) -> dict[str, Any] | None: """Collect pending composition retry receipts for status/doctor readback. @@ -443,21 +444,29 @@ def collect_pending_composition_retry_projection( ] else: journal_paths = [] + selector_agent_id = str(agent_id or "") if agent_id else "" pending: list[dict[str, Any]] = [] + total_matching = 0 for journal_path in journal_paths: if not journal_path.is_file(): continue for row in pending_composition_retry_receipts_for_path(journal_path): - pending.append(row) - if len(pending) >= max_items: - break - if len(pending) >= max_items: - break + if ( + selector_agent_id + and str((row.get("identity") or {}).get("agent_id") or "") + != selector_agent_id + ): + continue + total_matching += 1 + if len(pending) < max_items: + pending.append(row) + if not total_matching: + return None if not pending: return None return { "schema_version": POST_WRITEBACK_COMPOSITION_RETRY_PROJECTION_SCHEMA_VERSION, - "pending_count": len(pending), + "pending_count": total_matching, "pending": pending, "replay_action": POST_WRITEBACK_COMPOSITION_RETRY_REPLAY_ACTION, } diff --git a/tests/control_plane/test_post_writeback_composition_retry.py b/tests/control_plane/test_post_writeback_composition_retry.py index 692d0e4b51..4f9e34dffe 100644 --- a/tests/control_plane/test_post_writeback_composition_retry.py +++ b/tests/control_plane/test_post_writeback_composition_retry.py @@ -942,3 +942,74 @@ def upgraded_hook() -> PostWritebackHookRegistration: assert settled_flag is True assert settled["status"] == "settled" assert pending_composition_retry_receipts(runtime_root, "goal-1") == [] + + +def test_agent_scoped_projection_excludes_peer_work_and_caps_after_filter( + tmp_path: Path, +) -> None: + """An agent selector must filter before counting and display truncation.""" + + _registry, runtime_root = _write_registry(tmp_path) + journal_path = composition_retry_receipt_log_path(runtime_root, "goal-1") + peer_identity = dict(_identity()) + peer_identity["agent_id"] = "agent-peer" + for index in range(6): + peer_identity["todo_id"] = f"todo_peer_{index:03d}" + append_composition_retry_receipt( + journal_path, + build_composition_retry_receipt( + goal_id="goal-1", + event_kind="todo_complete", + identity=peer_identity, + state_version=f"peer-{index}", + committed_at="2026-09-06T00:00:00Z", + hook_identities=[ + { + "hook_id": "periodic_report.stage_completion", + "capability_id": "periodic-report", + "policy_version": "v0", + } + ], + error_code="source_projection_failed", + ), + ) + own_identity = dict(_identity()) + own_identity["agent_id"] = "agent-own" + own_identity["todo_id"] = "todo_own_000" + append_composition_retry_receipt( + journal_path, + build_composition_retry_receipt( + goal_id="goal-1", + event_kind="todo_complete", + identity=own_identity, + state_version="own-0", + committed_at="2026-09-06T00:00:00Z", + hook_identities=[ + { + "hook_id": "periodic_report.stage_completion", + "capability_id": "periodic-report", + "policy_version": "v0", + } + ], + error_code="source_projection_failed", + ), + ) + + own = collect_pending_composition_retry_projection( + runtime_root, "goal-1", agent_id="agent-own", max_items=5 + ) + assert own is not None + assert own["pending_count"] == 1 + assert [row["identity"]["agent_id"] for row in own["pending"]] == ["agent-own"] + + global_view = collect_pending_composition_retry_projection( + runtime_root, "goal-1", max_items=5 + ) + assert global_view is not None + assert global_view["pending_count"] == 7 + assert len(global_view["pending"]) == 5 + + lonely = collect_pending_composition_retry_projection( + runtime_root, "goal-1", agent_id="agent-nobody" + ) + assert lonely is None