Build durable close evidence outbox

Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a02b6f-7db1-7222-918b-e813a6bda38d
2026-08-23 14:05:50 +02:00 · 2026-08-23 14:05:50 +02:00 · c97e5ea92d
commit c97e5ea92d
parent 9866f35b30
5 changed files with 978 additions and 0 deletions
--- 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`
--- a/docs/close-evidence-outbox.md
+++ 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.
--- a/rein_aharness/close_outbox.py
+++ 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()
--- a/tests/test_close_outbox.py
+++ 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)
--- 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