diff --git a/README.md b/README.md index d77626d..e478eff 100644 --- a/README.md +++ b/README.md @@ -98,5 +98,6 @@ rein CLI details in an assignment. Instance manifest contract: [docs/instance-manifest.md](docs/instance-manifest.md). Example: [examples/schedule.harness.yml](examples/schedule.harness.yml). Repository mutation authority: [docs/repository-grant.md](docs/repository-grant.md). +Durable result-close design: [docs/close-evidence-outbox.md](docs/close-evidence-outbox.md). Tests: `PYTHONPATH=".:$HOME/llm-connect" python3 -m pytest tests/ -q` diff --git a/docs/close-evidence-outbox.md b/docs/close-evidence-outbox.md new file mode 100644 index 0000000..1b46d0d --- /dev/null +++ b/docs/close-evidence-outbox.md @@ -0,0 +1,69 @@ +# Close-evidence outbox + +Status: **durable core implemented, not connected to the live claim loop**. + +The outbox preserves one bounded Activity Core completion or failure intent +after repository work. Replaying an outbox entry calls only the queue close +adapter; it never dispatches a model, invokes an approach, or repeats repository +mutation. + +## Storage and identity + +The default root is: + +```text +$REIN_AHARNESS_STATE_DIR/close-outbox/ +``` + +When that variable is absent, the harness uses +`$XDG_STATE_HOME/rein-aharness` or `~/.local/state/rein-aharness`. State lives +outside target checkouts. Outbox directories are mode `0700`; lock and JSON +files are mode `0600`. + +An entry is keyed by the digest of the bounded Activity Core `run_id` and +repository `transaction_id`. Re-enqueuing the identical intent returns its +existing pending or delivered state. Reusing that identity for a different +action, result, error, or reopen decision is a conflict and is refused. + +The v1 close intent contains: + +- `action`: `complete` or `fail`; +- bounded normalized JSON `result` evidence; +- a caller-sanitized bounded error and explicit `reopen` decision for failures; +- no prompt, provider body, credential, raw tool stream, or workload callback. + +Results allow only finite JSON values, bounded depth, collection size, string +length, integer range, key length, and a total encoded size of 64 KiB. + +## Durability and replay + +Enqueue and retry-state updates use a same-directory temporary file, file +`fsync`, atomic replace, and directory `fsync` under a process-safe `flock`. +The retry attempt is durable before the delivery callback begins. + +Normal callback return marks the entry delivered and atomically moves it from +`pending/` to `delivered/`. An ordinary exception retains it in `pending/` with +only the exception class and a generic failure marker; exception text is not +persisted. Process interrupts propagate; the already-recorded attempt remains +pending. Invalid, oversized, or directory/state-mismatched records are moved +intact to `quarantine/` with a bounded reason sidecar and are never delivered +automatically. + +A network timeout may occur after Activity Core accepted a close. Retrying the +same close intent can therefore repeat the queue mutation request, but cannot +repeat repository workload. Activity Core remains authoritative for terminal +row state and must provide idempotent or reconcilable close semantics. + +## Activation gates + +The live claim loop still returns an in-memory failure when close delivery +fails. Connecting it to this outbox requires: + +1. the lease-bound repository transaction and accepted-result envelope; +2. enqueue-before-close using the same run and transaction identities; +3. Activity Core review of repeat close behavior for already-terminal rows; +4. startup/periodic replay that never calls workload code; and +5. operator status, quarantine inspection, and replay controls. + +Until those gates are implemented and reviewed, no production close behavior +changes. diff --git a/rein_aharness/close_outbox.py b/rein_aharness/close_outbox.py new file mode 100644 index 0000000..03754f6 --- /dev/null +++ b/rein_aharness/close_outbox.py @@ -0,0 +1,544 @@ +"""Durable, idempotent close-evidence outbox core. + +The outbox is intentionally not wired into the live claim loop yet. It stores +bounded Activity Core completion/failure intents outside target checkouts so a +future reconciler can retry close delivery without re-running repository work. +""" + +from __future__ import annotations + +import fcntl +import hashlib +import json +import math +import os +import re +import uuid +from contextlib import contextmanager +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Callable, Iterator, TextIO + +OUTBOX_VERSION = "1" +_MAX_PAYLOAD_BYTES = 65536 +_MAX_STRING_LENGTH = 4000 +_MAX_COLLECTION_ITEMS = 200 +_MAX_JSON_DEPTH = 8 +_IDENTIFIER = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:@+-]{0,199}") +_RECORD_KEYS = frozenset( + { + "version", + "entry_id", + "run_id", + "transaction_id", + "action", + "result", + "error", + "reopen", + "state", + "created_at", + "attempts", + "last_attempt_at", + "last_error", + } +) + + +class CloseOutboxError(RuntimeError): + """Base error for invalid records or durable outbox operations.""" + + +class InvalidCloseRequestError(CloseOutboxError): + """A close intent is ambiguous, unbounded, or unsupported.""" + + +class OutboxConflictError(CloseOutboxError): + """One run/transaction identity was reused for a different close intent.""" + + +class OutboxCorruptError(CloseOutboxError): + """A durable record cannot be safely decoded or validated.""" + + +@dataclass(frozen=True) +class CloseRequest: + """One bounded Activity Core completion or failure intent.""" + + run_id: str + transaction_id: str + action: str + result: dict[str, Any] + error: str = "" + reopen: bool = False + version: str = OUTBOX_VERSION + + def __post_init__(self) -> None: + if self.version != OUTBOX_VERSION: + raise InvalidCloseRequestError( + f"unsupported close request version {self.version!r}" + ) + _validate_identifier(self.run_id, "run_id") + _validate_identifier(self.transaction_id, "transaction_id") + if self.action not in {"complete", "fail"}: + raise InvalidCloseRequestError("action must be 'complete' or 'fail'") + if not isinstance(self.result, dict): + raise InvalidCloseRequestError("result must be an object") + normalized = _normalize_json(self.result) + object.__setattr__(self, "result", normalized) + if not isinstance(self.error, str): + raise InvalidCloseRequestError("error must be a string") + error = _bounded_text(self.error, 2000) + object.__setattr__(self, "error", error) + if not isinstance(self.reopen, bool): + raise InvalidCloseRequestError("reopen must be a boolean") + if self.action == "complete" and (error or self.reopen): + raise InvalidCloseRequestError( + "complete requests cannot carry error or reopen" + ) + if self.action == "fail" and not error: + raise InvalidCloseRequestError("fail requests require a bounded error") + if len(_json_bytes(self.intent())) > _MAX_PAYLOAD_BYTES: + raise InvalidCloseRequestError( + f"close request exceeds {_MAX_PAYLOAD_BYTES} encoded bytes" + ) + + @property + def entry_id(self) -> str: + identity = f"{self.run_id}\0{self.transaction_id}" + return hashlib.sha256(identity.encode("utf-8")).hexdigest()[:32] + + @property + def intent_digest(self) -> str: + return hashlib.sha256(_json_bytes(self.intent())).hexdigest() + + def intent(self) -> dict[str, Any]: + return { + "version": self.version, + "entry_id": self.entry_id, + "run_id": self.run_id, + "transaction_id": self.transaction_id, + "action": self.action, + "result": self.result, + "error": self.error, + "reopen": self.reopen, + } + + def detached_copy(self) -> "CloseRequest": + return CloseRequest( + run_id=self.run_id, + transaction_id=self.transaction_id, + action=self.action, + result=json.loads(json.dumps(self.result)), + error=self.error, + reopen=self.reopen, + version=self.version, + ) + + +@dataclass(frozen=True) +class EnqueueReceipt: + entry_id: str + created: bool + state: str + + +@dataclass(frozen=True) +class ReplayReport: + attempted: int = 0 + delivered: int = 0 + failed: int = 0 + quarantined: int = 0 + remaining: int = 0 + + +@dataclass(frozen=True) +class _Record: + request: CloseRequest + state: str + created_at: str + attempts: int = 0 + last_attempt_at: str | None = None + last_error: str | None = None + + def __post_init__(self) -> None: + if self.state not in {"pending", "delivered"}: + raise OutboxCorruptError("invalid outbox state") + if isinstance(self.attempts, bool) or not isinstance(self.attempts, int): + raise OutboxCorruptError("attempts must be a non-negative integer") + if self.attempts < 0: + raise OutboxCorruptError("attempts must be a non-negative integer") + _validate_timestamp(self.created_at, "created_at") + if self.last_attempt_at is not None: + _validate_timestamp(self.last_attempt_at, "last_attempt_at") + if self.last_error is not None and not isinstance(self.last_error, str): + raise OutboxCorruptError("last_error must be a string or null") + if self.last_error is not None and len(self.last_error) > 1000: + raise OutboxCorruptError("last_error exceeds the durable size bound") + + def payload(self) -> dict[str, Any]: + return { + **self.request.intent(), + "state": self.state, + "created_at": self.created_at, + "attempts": self.attempts, + "last_attempt_at": self.last_attempt_at, + "last_error": self.last_error, + } + + @classmethod + def from_payload(cls, value: Any) -> "_Record": + if not isinstance(value, dict): + raise OutboxCorruptError("outbox record must be an object") + if any(not isinstance(key, str) for key in value): + raise OutboxCorruptError("outbox record field names must be strings") + actual = set(value) + if actual != _RECORD_KEYS: + missing = sorted(_RECORD_KEYS - actual) + unknown = sorted(actual - _RECORD_KEYS) + raise OutboxCorruptError( + f"outbox record fields invalid: missing={missing} unknown={unknown}" + ) + try: + request = CloseRequest( + run_id=value["run_id"], + transaction_id=value["transaction_id"], + action=value["action"], + result=value["result"], + error=value["error"], + reopen=value["reopen"], + version=value["version"], + ) + except InvalidCloseRequestError as exc: + raise OutboxCorruptError(str(exc)) from exc + if value["entry_id"] != request.entry_id: + raise OutboxCorruptError("entry_id does not match run/transaction identity") + return cls( + request=request, + state=value["state"], + created_at=value["created_at"], + attempts=value["attempts"], + last_attempt_at=value["last_attempt_at"], + last_error=value["last_error"], + ) + + +class CloseOutbox: + """Process-safe durable store for close intents and replay outcomes.""" + + def __init__(self, *, state_dir: Path | None = None) -> None: + self.state_dir = ( + Path(state_dir).expanduser().resolve() if state_dir else _state_dir() + ) + self.root = self.state_dir / "close-outbox" + self.pending_dir = self.root / "pending" + self.delivered_dir = self.root / "delivered" + self.quarantine_dir = self.root / "quarantine" + self.lock_path = self.root / ".lock" + for directory in ( + self.root, + self.pending_dir, + self.delivered_dir, + self.quarantine_dir, + ): + _ensure_private_dir(directory) + + def enqueue(self, request: CloseRequest) -> EnqueueReceipt: + """Persist one intent or return its existing identical durable state.""" + if not isinstance(request, CloseRequest): + raise InvalidCloseRequestError("request must be a CloseRequest") + with self._locked(): + for state, directory in ( + ("delivered", self.delivered_dir), + ("pending", self.pending_dir), + ): + path = directory / f"{request.entry_id}.json" + if path.exists(): + record = self._load(path) + if record.state != state: + raise OutboxCorruptError( + f"outbox record directory/state mismatch: expected {state}" + ) + if record.request.intent_digest != request.intent_digest: + raise OutboxConflictError( + "run/transaction identity already has a different close intent" + ) + return EnqueueReceipt(request.entry_id, False, state) + if tuple(self.quarantine_dir.glob(f"{request.entry_id}.*.json")): + raise OutboxCorruptError( + "run/transaction identity has quarantined outbox material" + ) + record = _Record( + request=request, + state="pending", + created_at=_now(), + ) + _atomic_write_json( + self.pending_dir / f"{request.entry_id}.json", + record.payload(), + ) + return EnqueueReceipt(request.entry_id, True, "pending") + + def replay( + self, + deliver: Callable[[CloseRequest], Any], + *, + limit: int = 100, + ) -> ReplayReport: + """Attempt pending deliveries once each without executing workload code.""" + if isinstance(limit, bool) or not isinstance(limit, int) or not 1 <= limit <= 1000: + raise ValueError("limit must be an integer between 1 and 1000") + attempted = delivered = failed = quarantined = 0 + with self._locked(): + paths = sorted(self.pending_dir.glob("*.json"))[:limit] + for path in paths: + try: + record = self._load(path) + except OutboxCorruptError as exc: + self._quarantine(path, str(exc)) + quarantined += 1 + continue + if record.state != "pending": + self._quarantine( + path, + f"pending directory contains state {record.state!r}", + ) + quarantined += 1 + continue + + destination = self.delivered_dir / path.name + if destination.exists(): + self._quarantine( + path, + "pending entry conflicts with existing delivered material", + ) + quarantined += 1 + continue + + attempted += 1 + attempted_at = _now() + trying = _Record( + request=record.request, + state="pending", + created_at=record.created_at, + attempts=record.attempts + 1, + last_attempt_at=attempted_at, + last_error=None, + ) + _atomic_write_json(path, trying.payload()) + try: + deliver(record.request.detached_copy()) + except Exception as exc: + retained = _Record( + request=record.request, + state="pending", + created_at=record.created_at, + attempts=trying.attempts, + last_attempt_at=attempted_at, + last_error=( + f"{type(exc).__name__}: close delivery failed" + )[:1000], + ) + _atomic_write_json(path, retained.payload()) + failed += 1 + continue + + completed = _Record( + request=record.request, + state="delivered", + created_at=record.created_at, + attempts=trying.attempts, + last_attempt_at=attempted_at, + last_error=None, + ) + _atomic_write_json(path, completed.payload()) + os.replace(path, destination) + _fsync_dir(self.pending_dir) + _fsync_dir(self.delivered_dir) + delivered += 1 + + remaining = len(tuple(self.pending_dir.glob("*.json"))) + return ReplayReport( + attempted=attempted, + delivered=delivered, + failed=failed, + quarantined=quarantined, + remaining=remaining, + ) + + def pending_count(self) -> int: + with self._locked(): + return len(tuple(self.pending_dir.glob("*.json"))) + + def _load(self, path: Path) -> _Record: + try: + if path.stat().st_size > _MAX_PAYLOAD_BYTES + 8192: + raise OutboxCorruptError("outbox record exceeds the durable size bound") + raw = path.read_text(encoding="utf-8") + value = json.loads(raw) + return _Record.from_payload(value) + except OutboxCorruptError: + raise + except (OSError, UnicodeError, ValueError, TypeError) as exc: + raise OutboxCorruptError( + f"cannot decode outbox record {path.name}: {type(exc).__name__}" + ) from exc + + def _quarantine(self, path: Path, reason: str) -> None: + suffix = uuid.uuid4().hex[:12] + target = self.quarantine_dir / f"{path.stem}.{suffix}.json" + os.replace(path, target) + _fsync_dir(self.pending_dir) + _fsync_dir(self.quarantine_dir) + _atomic_write_json( + self.quarantine_dir / f"{path.stem}.{suffix}.error.json", + { + "entry_id": path.stem, + "quarantined_at": _now(), + "reason": _bounded_text(reason, 1000), + }, + ) + + @contextmanager + def _locked(self) -> Iterator[None]: + lock_file = _open_private_file(self.lock_path) + try: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX) + yield + finally: + try: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN) + finally: + lock_file.close() + + +def _state_dir() -> Path: + explicit = os.environ.get("REIN_AHARNESS_STATE_DIR", "").strip() + if explicit: + return Path(explicit).expanduser().resolve() + xdg = os.environ.get("XDG_STATE_HOME", "").strip() + if xdg: + return (Path(xdg).expanduser() / "rein-aharness").resolve() + return (Path.home() / ".local" / "state" / "rein-aharness").resolve() + + +def _validate_identifier(value: Any, name: str) -> None: + if not isinstance(value, str) or _IDENTIFIER.fullmatch(value) is None: + raise InvalidCloseRequestError( + f"{name} must be a bounded, path-safe identifier" + ) + + +def _normalize_json(value: Any, *, depth: int = 0) -> Any: + if depth > _MAX_JSON_DEPTH: + raise InvalidCloseRequestError("result exceeds the JSON depth bound") + if value is None or isinstance(value, bool): + return value + if isinstance(value, str): + if len(value) > _MAX_STRING_LENGTH: + raise InvalidCloseRequestError( + f"result string exceeds {_MAX_STRING_LENGTH} characters" + ) + return value + if isinstance(value, int): + if abs(value) > 2**63 - 1: + raise InvalidCloseRequestError("result integer exceeds signed 64-bit range") + return value + if isinstance(value, float): + if not math.isfinite(value): + raise InvalidCloseRequestError("result floats must be finite") + return value + if isinstance(value, list): + if len(value) > _MAX_COLLECTION_ITEMS: + raise InvalidCloseRequestError("result array exceeds the item bound") + return [_normalize_json(item, depth=depth + 1) for item in value] + if isinstance(value, dict): + if len(value) > _MAX_COLLECTION_ITEMS: + raise InvalidCloseRequestError("result object exceeds the field bound") + normalized: dict[str, Any] = {} + for key, item in value.items(): + if not isinstance(key, str) or not key or len(key) > 200: + raise InvalidCloseRequestError( + "result field names must be non-empty strings of at most 200 characters" + ) + normalized[key] = _normalize_json(item, depth=depth + 1) + return normalized + raise InvalidCloseRequestError( + f"result contains unsupported JSON type: {type(value).__name__}" + ) + + +def _validate_timestamp(value: Any, name: str) -> None: + if not isinstance(value, str): + raise OutboxCorruptError(f"{name} must be a timestamp string") + try: + parsed = datetime.fromisoformat(value) + except ValueError as exc: + raise OutboxCorruptError(f"{name} is not an ISO timestamp") from exc + if parsed.tzinfo is None: + raise OutboxCorruptError(f"{name} must include a timezone") + + +def _ensure_private_dir(path: Path) -> None: + path.mkdir(parents=True, exist_ok=True) + os.chmod(path, 0o700) + + +def _open_private_file(path: Path) -> TextIO: + fd = os.open(path, os.O_RDWR | os.O_CREAT, 0o600) + try: + os.fchmod(fd, 0o600) + return os.fdopen(fd, "r+", encoding="utf-8") + except BaseException: + os.close(fd) + raise + + +def _atomic_write_json(path: Path, value: dict[str, Any]) -> None: + data = _json_bytes(value) + b"\n" + temp = path.parent / f".{path.name}.{uuid.uuid4().hex}.tmp" + fd = os.open(temp, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + try: + os.fchmod(fd, 0o600) + with os.fdopen(fd, "wb") as handle: + fd = -1 + handle.write(data) + handle.flush() + os.fsync(handle.fileno()) + os.replace(temp, path) + os.chmod(path, 0o600) + _fsync_dir(path.parent) + except BaseException: + if fd >= 0: + os.close(fd) + try: + temp.unlink() + except FileNotFoundError: + pass + raise + + +def _fsync_dir(path: Path) -> None: + fd = os.open(path, os.O_RDONLY) + try: + os.fsync(fd) + finally: + os.close(fd) + + +def _json_bytes(value: Any) -> bytes: + return json.dumps( + value, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + allow_nan=False, + ).encode("utf-8") + + +def _bounded_text(value: str, limit: int) -> str: + return " ".join(value.split())[:limit] + + +def _now() -> str: + return datetime.now(timezone.utc).isoformat() diff --git a/tests/test_close_outbox.py b/tests/test_close_outbox.py new file mode 100644 index 0000000..ffabc8b --- /dev/null +++ b/tests/test_close_outbox.py @@ -0,0 +1,339 @@ +from __future__ import annotations + +import json +import math +import os +import subprocess +import sys +from pathlib import Path + +import pytest + +import rein_aharness.close_outbox as close_outbox_module +from rein_aharness.close_outbox import ( + CloseOutbox, + CloseRequest, + InvalidCloseRequestError, + OutboxConflictError, + OutboxCorruptError, +) + + +def _request( + *, + run_id: str = "run-1", + transaction_id: str = "tx-1", + action: str = "complete", + result: dict[str, object] | None = None, + error: str = "", + reopen: bool = False, +) -> CloseRequest: + return CloseRequest( + run_id=run_id, + transaction_id=transaction_id, + action=action, + result=result or {"ok": True, "accepted_commit": "a" * 40}, + error=error, + reopen=reopen, + ) + + +def test_enqueue_uses_private_external_atomic_record(tmp_path: Path) -> None: + checkout = tmp_path / "checkout" + checkout.mkdir() + outbox = CloseOutbox(state_dir=tmp_path / "state") + + receipt = outbox.enqueue(_request()) + + path = outbox.pending_dir / f"{receipt.entry_id}.json" + assert receipt.created is True + assert receipt.state == "pending" + assert path.is_file() + assert not path.is_relative_to(checkout) + assert path.stat().st_mode & 0o777 == 0o600 + assert outbox.root.stat().st_mode & 0o777 == 0o700 + assert outbox.pending_dir.stat().st_mode & 0o777 == 0o700 + payload = json.loads(path.read_text(encoding="utf-8")) + assert payload["state"] == "pending" + assert payload["attempts"] == 0 + assert payload["entry_id"] == receipt.entry_id + assert not tuple(outbox.pending_dir.glob("*.tmp")) + + +def test_identical_enqueue_is_suppressed_without_rewrite(tmp_path: Path) -> None: + outbox = CloseOutbox(state_dir=tmp_path / "state") + request = _request() + first = outbox.enqueue(request) + path = outbox.pending_dir / f"{first.entry_id}.json" + before = path.read_bytes() + + second = outbox.enqueue(request) + + assert second.created is False + assert second.state == "pending" + assert path.read_bytes() == before + assert outbox.pending_count() == 1 + + +def test_identity_reuse_with_different_intent_is_rejected(tmp_path: Path) -> None: + outbox = CloseOutbox(state_dir=tmp_path / "state") + outbox.enqueue(_request(result={"ok": True})) + + with pytest.raises(OutboxConflictError, match="different close intent"): + outbox.enqueue(_request(result={"ok": False})) + + +def test_successful_replay_moves_entry_and_never_redelivers(tmp_path: Path) -> None: + outbox = CloseOutbox(state_dir=tmp_path / "state") + receipt = outbox.enqueue(_request()) + delivered: list[CloseRequest] = [] + + report = outbox.replay(delivered.append) + + assert report.attempted == 1 + assert report.delivered == 1 + assert report.failed == 0 + assert report.remaining == 0 + assert len(delivered) == 1 + path = outbox.delivered_dir / f"{receipt.entry_id}.json" + payload = json.loads(path.read_text(encoding="utf-8")) + assert payload["state"] == "delivered" + assert payload["attempts"] == 1 + assert payload["last_error"] is None + + second = outbox.replay(delivered.append) + duplicate = outbox.enqueue(_request()) + assert second.attempted == 0 + assert len(delivered) == 1 + assert duplicate.created is False + assert duplicate.state == "delivered" + + +def test_failed_delivery_stays_pending_then_replays_same_intent(tmp_path: Path) -> None: + outbox = CloseOutbox(state_dir=tmp_path / "state") + request = _request( + action="fail", + result={"ok": False}, + error="adapter failed", + reopen=False, + ) + receipt = outbox.enqueue(request) + + def unavailable(close_request: CloseRequest) -> None: + close_request.result["mutated-by-callback"] = True + raise RuntimeError("API down\n" + ("x" * 2000)) + + first = outbox.replay(unavailable) + path = outbox.pending_dir / f"{receipt.entry_id}.json" + retained = json.loads(path.read_text(encoding="utf-8")) + assert first.failed == 1 + assert first.remaining == 1 + assert retained["attempts"] == 1 + assert retained["last_error"] == "RuntimeError: close delivery failed" + assert "API down" not in retained["last_error"] + assert "x" * 20 not in retained["last_error"] + assert "mutated-by-callback" not in retained["result"] + + delivered: list[CloseRequest] = [] + second = outbox.replay(delivered.append) + final = json.loads( + (outbox.delivered_dir / path.name).read_text(encoding="utf-8") + ) + assert second.delivered == 1 + assert final["attempts"] == 2 + assert final["last_error"] is None + assert delivered[0].intent_digest == request.intent_digest + + +def test_process_interrupt_leaves_attempt_durable_and_pending(tmp_path: Path) -> None: + outbox = CloseOutbox(state_dir=tmp_path / "state") + receipt = outbox.enqueue(_request()) + + def interrupted(_: CloseRequest) -> None: + raise KeyboardInterrupt + + with pytest.raises(KeyboardInterrupt): + outbox.replay(interrupted) + + payload = json.loads( + (outbox.pending_dir / f"{receipt.entry_id}.json").read_text(encoding="utf-8") + ) + assert payload["state"] == "pending" + assert payload["attempts"] == 1 + assert payload["last_attempt_at"] + assert payload["last_error"] is None + + +def test_corrupt_pending_record_is_preserved_in_quarantine(tmp_path: Path) -> None: + outbox = CloseOutbox(state_dir=tmp_path / "state") + request = _request() + receipt = outbox.enqueue(request) + path = outbox.pending_dir / f"{receipt.entry_id}.json" + path.write_text("{not-json\n", encoding="utf-8") + called = False + + def deliver(_: CloseRequest) -> None: + nonlocal called + called = True + + report = outbox.replay(deliver) + + assert report.quarantined == 1 + assert report.attempted == 0 + assert report.remaining == 0 + assert called is False + quarantined = tuple(outbox.quarantine_dir.glob(f"{receipt.entry_id}.*.json")) + assert len(quarantined) == 2 + assert any(item.name.endswith(".error.json") for item in quarantined) + assert any(item.read_text(encoding="utf-8") == "{not-json\n" for item in quarantined) + with pytest.raises(OutboxCorruptError, match="quarantined"): + outbox.enqueue(request) + + +def test_pending_record_with_delivered_state_is_quarantined(tmp_path: Path) -> None: + outbox = CloseOutbox(state_dir=tmp_path / "state") + receipt = outbox.enqueue(_request()) + path = outbox.pending_dir / f"{receipt.entry_id}.json" + payload = json.loads(path.read_text(encoding="utf-8")) + payload["state"] = "delivered" + path.write_text(json.dumps(payload), encoding="utf-8") + + report = outbox.replay(lambda _: pytest.fail("must not deliver corrupt state")) + + assert report.quarantined == 1 + assert report.attempted == 0 + assert report.remaining == 0 + + +def test_atomic_enqueue_failure_leaves_no_partial_or_temp_record( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + outbox = CloseOutbox(state_dir=tmp_path / "state") + + def fail_replace(source: Path, destination: Path) -> None: + raise OSError("simulated rename failure") + + monkeypatch.setattr(close_outbox_module.os, "replace", fail_replace) + with pytest.raises(OSError, match="rename failure"): + outbox.enqueue(_request()) + + assert not tuple(outbox.pending_dir.iterdir()) + + +def test_two_processes_enqueue_one_durable_intent(tmp_path: Path) -> None: + state_dir = tmp_path / "state" + child = """ +import json +import sys +from pathlib import Path +from rein_aharness.close_outbox import CloseOutbox, CloseRequest +outbox = CloseOutbox(state_dir=Path(sys.argv[1])) +receipt = outbox.enqueue(CloseRequest( + run_id="run-shared", + transaction_id="tx-shared", + action="complete", + result={"ok": True}, +)) +print(json.dumps({"created": receipt.created, "entry_id": receipt.entry_id})) +""" + env = os.environ.copy() + processes = [ + subprocess.Popen( + [sys.executable, "-c", child, str(state_dir)], + cwd=Path(__file__).resolve().parents[1], + env=env, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + for _ in range(2) + ] + outputs = [process.communicate(timeout=30) for process in processes] + + assert [process.returncode for process in processes] == [0, 0] + receipts = [json.loads(stdout) for stdout, _ in outputs] + assert sorted(item["created"] for item in receipts) == [False, True] + assert receipts[0]["entry_id"] == receipts[1]["entry_id"] + outbox = CloseOutbox(state_dir=state_dir) + assert outbox.pending_count() == 1 + assert len(tuple(outbox.pending_dir.glob("*.json"))) == 1 + + +def test_replay_limit_is_bounded_and_leaves_remaining_entries(tmp_path: Path) -> None: + outbox = CloseOutbox(state_dir=tmp_path / "state") + for number in range(3): + outbox.enqueue(_request(run_id=f"run-{number}", transaction_id=f"tx-{number}")) + + report = outbox.replay(lambda _: None, limit=2) + + assert report.attempted == 2 + assert report.delivered == 2 + assert report.remaining == 1 + with pytest.raises(ValueError, match="between 1 and 1000"): + outbox.replay(lambda _: None, limit=0) + + +@pytest.mark.parametrize( + "updates,match", + [ + ({"run_id": "bad/id"}, "run_id"), + ({"transaction_id": "with space"}, "transaction_id"), + ({"action": "cancel"}, "action"), + ({"action": "complete", "error": "not allowed"}, "cannot carry"), + ({"action": "complete", "reopen": True}, "cannot carry"), + ({"action": "fail", "error": ""}, "require"), + ], +) +def test_close_request_rejects_ambiguous_identity_or_action( + updates: dict[str, object], + match: str, +) -> None: + values: dict[str, object] = { + "run_id": "run-1", + "transaction_id": "tx-1", + "action": "complete", + "result": {"ok": True}, + "error": "", + "reopen": False, + } + values.update(updates) + with pytest.raises(InvalidCloseRequestError, match=match): + CloseRequest(**values) # type: ignore[arg-type] + + +@pytest.mark.parametrize( + "result,match", + [ + ({"text": "x" * 4001}, "string exceeds"), + ({"items": list(range(201))}, "array exceeds"), + ({"value": math.nan}, "finite"), + ({"value": 2**63}, "64-bit"), + ({"value": {1, 2}}, "unsupported JSON type"), + ({1: "not-string"}, "field names"), + ( + {f"field-{number}": "x" * 4000 for number in range(20)}, + "encoded bytes", + ), + ], +) +def test_close_request_rejects_unbounded_or_non_json_result( + result: dict[object, object], + match: str, +) -> None: + with pytest.raises(InvalidCloseRequestError, match=match): + CloseRequest( + run_id="run-1", + transaction_id="tx-1", + action="complete", + result=result, # type: ignore[arg-type] + ) + + +def test_close_request_rejects_excessive_json_depth() -> None: + value: dict[str, object] = {"leaf": True} + for _ in range(10): + value = {"nested": value} + + with pytest.raises(InvalidCloseRequestError, match="depth bound"): + _request(result=value) diff --git a/workplans/HARNESS-WP-0003-governed-runtime-integrity.md b/workplans/HARNESS-WP-0003-governed-runtime-integrity.md index fdb9cc5..0db34ed 100644 --- a/workplans/HARNESS-WP-0003-governed-runtime-integrity.md +++ b/workplans/HARNESS-WP-0003-governed-runtime-integrity.md @@ -274,6 +274,31 @@ authoritative-carriage review from Activity Core Glas (`e149c359-36bf-4867-ac10-f58a8f248666`). No execution enablement was requested. +### Close-evidence outbox preparation — 2026-08-23 + +Added a production-inert `CloseOutbox` core for Activity Core completion and +failure intents. It stores private mode-`0600` JSON outside target checkouts, +keys entries by bounded run/transaction identity, uses a process-safe lock and +file/directory `fsync` around atomic replacement, suppresses identical pending +or delivered entries, and rejects conflicting identity reuse. Result evidence +is normalized and bounded by type, depth, collection/string/integer limits, and +a 64-KiB encoded ceiling. + +Replay durably increments the attempt before calling a close-only callback, +moves success to delivered state, retains ordinary API failure with a bounded +error, and leaves process interrupts pending. Corrupt, oversized, or +directory/state-mismatched material is preserved with a reason in quarantine +and never delivered. Twenty-five focused cases cover private/atomic storage, +duplicate/conflict behavior, successful and failed replay, callback mutation, +interrupt persistence, corruption, atomic-write failure, two-process enqueue, +replay limits, and payload bounds. + +`docs/close-evidence-outbox.md` records the storage/replay contract and the +network-ambiguity residual: a close request may be repeated after a timeout, +but replay never calls workload code and therefore cannot duplicate the +repository commit. The live claim loop remains unchanged pending transaction +wiring, Activity Core repeat-close review, and operator status/replay controls. + ## Remove tenant logic from the shared runtime ```task