"""Run-scoped artifact submission receipts — the single source of truth for
"was this artifact submitted in this run?".
A receipt decouples *completion detection* from *artifact storage*. The
submission handler writes a receipt the moment it has durably persisted an
artifact; the completion gate reads receipts to decide whether the required
artifact is present. The gate never recomputes a storage path, so a receipt
keyed on ``(run_id, artifact_type)`` — both stable identities, never paths —
cannot drift away from where the artifact actually landed (``.agent/tmp`` vs
``.agent/artifacts``, a per-worker namespace, or any future layout change).
Storage (RFC-013 P3): receipts are stored in a single WAL-mode SQLite
database at ``<workspace>/.agent/state.db`` via ``RunStateDB`` (one row
per ``(run_id, artifact_type)``). This eliminates one-file-per-event
state churn under ``.agent/receipts/<run_id>/`` (a measurable share of
macOS fseventsd activity under long multi-instance runs). The legacy
file path is preserved as a read-fallback during the dual-read rollout
window so an in-flight run that was upgraded mid-run still passes its
completion gate. Production writes go to the DB only; the file path is
read-only fallback.
"""
from __future__ import annotations
import contextlib
import hashlib
import hmac
import json
import sqlite3
from pathlib import Path
from typing import cast
from ralph.mcp.artifacts.file_backend import DEFAULT_FILE_BACKEND, FileBackend
from ralph.mcp.artifacts.state_db import MISSING, RunStateDB, _Missing
#: Directory (workspace-relative) holding every receipt for a single run.
RECEIPT_DIR_RELPATH_FMT = ".agent/receipts/{run_id}"
[docs]
class ReceiptPersistenceError(RuntimeError):
"""Raised when both the RunStateDB and legacy-file paths fail to persist a receipt.
Without this guard, ``write_artifact_receipt`` returns successfully even
when no durable receipt was written \u2014 letting artifact submission
continue against a missing receipt and producing a silent failure
downstream when the completion gate reads it. ``execute_ops_with_rollback``
already propagates this exception upward to rollback the in-flight
submit (the receipt op is the last step, so the artifact and its
handoff would also be unrolled).
"""
def _receipt_dir(workspace_root: Path, run_id: str) -> Path:
return workspace_root / RECEIPT_DIR_RELPATH_FMT.format(run_id=run_id)
def _receipt_path(workspace_root: Path, run_id: str, artifact_type: str) -> Path:
return _receipt_dir(workspace_root, run_id) / f"{artifact_type}.json"
def _receipt_hmac(secret: str, run_id: str, artifact_type: str) -> str:
"""Compute the HMAC-SHA256 of ``run_id`` and ``artifact_type`` with ``secret``.
The HMAC binds the receipt to the broker-owned ``secret`` so a model
that can write under ``.agent/`` (workspace write capabilities) cannot
forge a valid receipt without the secret. The secret is never exposed
via the agent's environment (notably not via ``MCP_RUN_ID_ENV`` or
any other broker-exposed variable).
"""
msg = f"{run_id}\n{artifact_type}".encode()
return hmac.new(secret.encode(), msg, hashlib.sha256).hexdigest()
def _open_db(workspace_root: Path) -> RunStateDB:
return RunStateDB(Path(workspace_root))
def _legacy_file_receipt_present(
workspace_root: Path,
run_id: str,
artifact_type: str,
*,
backend: FileBackend,
receipt_secret: str | None,
) -> bool:
"""Read the legacy ``.agent/receipts/<run_id>/<type>.json`` file path."""
path = _receipt_path(workspace_root, run_id, artifact_type)
if not backend.exists(path):
return False
if receipt_secret is None:
return True
try:
raw = backend.read_text(path, encoding="utf-8")
parsed = cast("object", json.loads(raw))
except (OSError, json.JSONDecodeError):
return False
if not isinstance(parsed, dict):
return False
stored = cast("dict[str, object]", parsed).get("hmac")
if not isinstance(stored, str):
return False
expected = _receipt_hmac(receipt_secret, run_id, artifact_type)
return hmac.compare_digest(stored, expected)
[docs]
def write_artifact_receipt(
workspace_root: Path,
run_id: str,
artifact_type: str,
*,
backend: FileBackend = DEFAULT_FILE_BACKEND,
receipt_secret: str | None = None,
) -> None:
"""Record that ``artifact_type`` was durably persisted during ``run_id``.
Must be called only after the artifact itself is committed to storage so the
receipt and the artifact appear together (or, on rollback, not at all).
When ``receipt_secret`` is provided the receipt includes a ``hmac``
field that binds it to the broker-owned secret. A model that can
write under ``.agent/`` cannot forge a receipt with a valid HMAC
because the secret is never exposed to the agent.
Storage (RFC-013 P3): the canonical store is the per-workspace
``.agent/state.db``. Production writes go to the DB ONLY when the
DB write succeeds; the legacy ``.agent/receipts/<run_id>/<artifact_type>.json``
file path is then read-only fallback during the dual-read rollout
window so receipts left behind by the pre-upgrade release are still
honored.
Durable-fallback: when ``RunStateDB`` raises ``sqlite3.Error``
(locked / corrupt / unsupported WAL) on either open or upsert,
this function falls back to writing the legacy file path so the
completion gate always has durable evidence. Atomic-rollback for
tests and callers using explicit ``backend`` kwargs still works
because ``backend`` continues to control where the legacy bytes
land (see ``FailingBackend`` pattern). The HMAC is included in
both stores when ``receipt_secret`` is provided.
"""
hmac_hex: str | None
if receipt_secret is not None:
hmac_hex = _receipt_hmac(receipt_secret, run_id, artifact_type)
else:
hmac_hex = None
db_written = False
db: RunStateDB | None = None
try:
db = _open_db(workspace_root)
except (OSError, RuntimeError, sqlite3.Error):
db = None
if db is not None:
try:
db.upsert_receipt(run_id, artifact_type, hmac_hex)
db_written = True
except sqlite3.Error:
pass # Will fall through to legacy-file durable fallback below.
finally:
with contextlib.suppress(OSError, RuntimeError, sqlite3.Error):
db.close()
if db_written:
return
legacy_written = _write_legacy_receipt_fallback(
workspace_root,
run_id,
artifact_type,
hmac_hex=hmac_hex,
backend=backend,
)
if not legacy_written:
raise ReceiptPersistenceError(
f"Both DB and legacy paths failed to persist receipt for "
f"run_id={run_id!r} artifact_type={artifact_type!r}"
)
def _write_legacy_receipt_fallback(
workspace_root: Path,
run_id: str,
artifact_type: str,
*,
hmac_hex: str | None,
backend: FileBackend,
) -> bool:
"""Write the legacy ``.agent/receipts/<run_id>/<artifact_type>.json`` fallback.
Used by ``write_artifact_receipt`` only when the RunStateDB write
fails (sqlite3.Error on open or upsert). The HMAC is included in
the payload when one was provided so a subsequent read with the
same secret verifies and a mismatching secret rejects.
Returns:
``True`` when the legacy receipt file was written;
``False`` when ``OSError`` blocked the write (either the
``.agent/receipts/<run_id>/`` mkdir or the file write).
A ``False`` return is the fail-closed signal that
``write_artifact_receipt`` escalates into a
``ReceiptPersistenceError`` so the artifact submit also
fails closed.
"""
path = _receipt_path(workspace_root, run_id, artifact_type)
payload: dict[str, object] = {"run_id": run_id, "artifact_type": artifact_type}
if hmac_hex is not None:
payload["hmac"] = hmac_hex
try:
backend.mkdir(_receipt_dir(workspace_root, run_id), parents=True, exist_ok=True)
backend.write_text(path, json.dumps(payload), encoding="utf-8")
except OSError:
return False # Both DB and legacy paths failed - nothing durable to write.
return True
[docs]
def artifact_receipt_present(
workspace_root: Path,
run_id: str,
artifact_type: str,
*,
backend: FileBackend = DEFAULT_FILE_BACKEND,
receipt_secret: str | None = None,
) -> bool:
"""Return True when a valid receipt for ``(run_id, artifact_type)`` exists.
Reads the per-workspace ``.agent/state.db`` first (RFC-013 P3). When
the DB has no row, falls back to the legacy file path
``.agent/receipts/<run_id>/<artifact_type>.json`` so receipts left
behind by the pre-upgrade release are still honored during the
dual-read window.
When ``receipt_secret`` is provided the stored HMAC is verified
against ``(run_id, artifact_type)``; a receipt that exists but
fails HMAC verification returns ``False``. This pins the receipt
to the broker-owned secret so a model with workspace write
capabilities cannot forge a valid receipt.
"""
try:
db = _open_db(workspace_root)
except (OSError, RuntimeError, sqlite3.Error):
# DB unavailable — fall back to the legacy file path.
return _legacy_file_receipt_present(
workspace_root,
run_id,
artifact_type,
backend=backend,
receipt_secret=receipt_secret,
)
stored: str | None | _Missing
try:
stored = db.get_receipt_hmac(run_id, artifact_type)
except (OSError, RuntimeError, sqlite3.Error):
# DB read failed; close and fall back to legacy file path.
with contextlib.suppress(OSError, RuntimeError, sqlite3.Error):
db.close()
return _legacy_file_receipt_present(
workspace_root,
run_id,
artifact_type,
backend=backend,
receipt_secret=receipt_secret,
)
with contextlib.suppress(OSError, RuntimeError, sqlite3.Error):
db.close()
if stored is not MISSING:
if receipt_secret is None:
return True
if not isinstance(stored, str):
return False
expected = _receipt_hmac(receipt_secret, run_id, artifact_type)
return hmac.compare_digest(stored, expected)
# DB has no row — read the legacy file path so an in-flight run that
# was upgraded mid-run still passes its completion gate.
return _legacy_file_receipt_present(
workspace_root,
run_id,
artifact_type,
backend=backend,
receipt_secret=receipt_secret,
)
[docs]
def delete_artifact_receipt(
workspace_root: Path,
run_id: str,
artifact_type: str,
*,
backend: FileBackend = DEFAULT_FILE_BACKEND,
) -> None:
"""Remove one receipt (no-op when absent) — the undo for ``write_artifact_receipt``.
Deletes both the DB row and the legacy file path (dual-target)
so a stale file from the pre-upgrade release cannot leave a
receipt in place after the DB row is gone.
"""
try:
db = _open_db(workspace_root)
except (OSError, RuntimeError, sqlite3.Error):
db = None
if db is not None:
try:
db.delete_receipt(run_id, artifact_type)
finally:
with contextlib.suppress(OSError, RuntimeError, sqlite3.Error):
db.close()
backend.unlink(_receipt_path(workspace_root, run_id, artifact_type), missing_ok=True)
[docs]
def clear_run_receipts(
workspace_root: Path,
run_id: str,
*,
backend: FileBackend = DEFAULT_FILE_BACKEND,
) -> None:
"""Remove every receipt for ``run_id`` (no-op when none exist).
Called at the start of each (re)invocation so a resumed session
with a reused ``run_id`` never inherits a stale "already submitted"
signal. Clears both the DB rows and the legacy file paths.
Best-effort: a missing or read-only ``.agent/state.db`` does not
block the call (the legacy file cleanup still proceeds). The DB
clear itself is also best-effort — a transient ``sqlite3.Error``
during ``RunStateDB.clear_run_receipts`` is suppressed so the
legacy-file cleanup below always runs, matching the RFC-013
retention contract that a single failure mode cannot abort
rerun / session cleanup.
"""
try:
db = _open_db(workspace_root)
except (OSError, RuntimeError, sqlite3.Error):
db = None
if db is not None:
try:
with contextlib.suppress(OSError, RuntimeError, sqlite3.Error):
db.clear_run_receipts(run_id)
finally:
with contextlib.suppress(OSError, RuntimeError, sqlite3.Error):
db.close()
receipt_dir = _receipt_dir(workspace_root, run_id)
for path in backend.glob(receipt_dir, "*.json"):
backend.unlink(path, missing_ok=True)
__all__ = [
"RECEIPT_DIR_RELPATH_FMT",
"ReceiptPersistenceError",
"artifact_receipt_present",
"clear_run_receipts",
"delete_artifact_receipt",
"write_artifact_receipt",
]