#!/usr/bin/env python3 """Durable private operations recording, Python 3.9+, standard library only. Explicitly enable a client and choose its private storage directory. enqueue() is local only; flush() and reconcile() make bounded requests when called. No action is executed or repeated by this module. A record is a caller's claim, not proof that a tool ran or that someone authorized it. Evidence bytes remain local: only their exact-byte SHA-256 digest is queued or uploaded. Keep the bytes separately. The outbox is scoped to one API origin, workspace, and stream. Pending records survive restarts. Ambiguous delivery is reconciled by stable client event keys. Terminal per-record rejections are quarantined; retry/drop require a local audit reason. An accepted record cannot be changed or dropped. No background delivery. CLI: python rightshold_operations.py /absolute/config.json [/absolute/input.json] Configuration excludes credentials and enabled. Supply RIGHTSHOLD_API_KEY only in the process environment. Local exports contain private record metadata. """ from __future__ import annotations import argparse import asyncio import contextlib import hashlib import http.client import json import math import os import re import sqlite3 import stat import sys import threading import time import urllib.error import urllib.parse import urllib.request from datetime import datetime, timezone from pathlib import Path from typing import Any, Callable, Dict, Optional _HTTP_CODES = frozenset(( "operations_invalid", "operations_forbidden", "operations_not_found", "operations_conflict", "operations_limit", "operations_unavailable", "invalid_api_key", "malformed_api_key", "api_key_required", "account_closed", "account_frozen", "sandbox_account_expired", "rate_limited", "too_many_requests", "invalid_json", "not_found", "internal_error", "unavailable", "service_unavailable", )) _LOCAL_CODES = frozenset(( "invalid_options", "invalid_scope", "invalid_base_url", "disabled", "closed", "outbox_locked", "outbox_busy", "storage_error", "storage_corrupt", "insecure_storage", "scope_mismatch", "backpressure", "invalid_event", "event_conflict", "invalid_evidence", "ambiguous_evidence", "credential_in_record", "invalid_response", "response_too_large", "network_error", "http_error", "invalid_reason", "event_not_found", "immutable_receipt", "invalid_identifier", "invalid_config", "invalid_command", )) class OperationsError(Exception): """Only allowlisted protocol codes are safe to log; remote text is discarded.""" def __init__(self, code: str, status: Optional[int] = None): self.code = code if code in _HTTP_CODES or code in _LOCAL_CODES else "http_error" self.status = status self.retryable = status == 429 or (status is not None and status >= 500) or self.code in ( "network_error", "invalid_response", "response_too_large") super().__init__("Rightshold operations request failed (" + self.code + (", HTTP " + str(status) if status is not None else "") + ").") def _json(value: Any) -> str: return json.dumps(value, sort_keys=True, ensure_ascii=False, separators=(",", ":"), allow_nan=False) def encode_operation_evidence(value: Any) -> bytes: """Exact strings/bytes; objects use sorted-key JSON. Retain these local bytes. Cross-language numeric serialization is intentionally not implied.""" try: if isinstance(value, (bytes, bytearray, memoryview)): return bytes(value) if isinstance(value, str): return value.encode("utf-8") return _json(value).encode("utf-8") except (ValueError, TypeError, UnicodeError, OverflowError): raise OperationsError("invalid_evidence") from None def hash_operation_evidence(value: Any) -> str: return hashlib.sha256(encode_operation_evidence(value)).hexdigest() def _digest(value: Any) -> str: return hashlib.sha256(_json(value).encode("utf-8")).hexdigest() def _now() -> str: return datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z") def _identifier(value: Any, pattern: str) -> str: if not isinstance(value, str) or not re.fullmatch(pattern, value): raise OperationsError("invalid_identifier") return value def _private_path(file: Path, directory: bool) -> None: info = file.lstat() if stat.S_ISLNK(info.st_mode) or (not stat.S_ISDIR(info.st_mode) if directory else not stat.S_ISREG(info.st_mode)): raise OperationsError("insecure_storage") if os.name != "nt" and (info.st_mode & 0o077 or hasattr(os, "getuid") and info.st_uid != os.getuid()): raise OperationsError("insecure_storage") def _safe_read(file: Path, maximum: int) -> str: _private_path(file, False) if file.stat().st_size > maximum: raise OperationsError("backpressure") return file.read_text(encoding="utf-8") def _origin(value: str) -> str: try: parsed = urllib.parse.urlsplit(value) if parsed.username or parsed.password or parsed.query or parsed.fragment or parsed.path not in ("", "/") or not parsed.hostname: raise ValueError() if parsed.scheme != "https" and not (parsed.scheme == "http" and parsed.hostname in ("localhost", "127.0.0.1", "::1")): raise ValueError() parsed.port return urllib.parse.urlunsplit((parsed.scheme, parsed.netloc, "", "", "")) except (ValueError, TypeError): raise OperationsError("invalid_base_url") from None class _NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): return None class _WriterLock: """Kernel-released advisory lock; a terminated process never strands the outbox.""" def __init__(self, file: Path): self.file = None try: if file.exists() or file.is_symlink(): _private_path(file, False) descriptor = os.open(str(file), os.O_RDWR | os.O_CREAT | getattr(os, "O_NOFOLLOW", 0), 0o600) self.file = os.fdopen(descriptor, "r+b", buffering=0) if os.name == "nt": import msvcrt if file.stat().st_size == 0: self.file.write(b"\0") self.file.seek(0) msvcrt.locking(self.file.fileno(), msvcrt.LK_NBLCK, 1) else: import fcntl fcntl.flock(self.file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) except OperationsError: self.close() raise except OSError: self.close() raise OperationsError("outbox_locked") from None def close(self): if self.file is not None: self.file.close() self.file = None _SCHEMA = """ CREATE TABLE IF NOT EXISTS metadata (id INTEGER PRIMARY KEY CHECK(id=1),scope TEXT NOT NULL,revision INTEGER NOT NULL,created_at TEXT NOT NULL); CREATE TABLE IF NOT EXISTS records ( client_event_id TEXT PRIMARY KEY, request_sha256 TEXT NOT NULL, body TEXT NOT NULL, status TEXT NOT NULL CHECK(status IN ('pending','accepted','quarantined','dropped')), attempts INTEGER NOT NULL DEFAULT 0, next_attempt_at INTEGER NOT NULL DEFAULT 0, uncertain INTEGER NOT NULL DEFAULT 0, last_error TEXT, receipt TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS pending_records ON records(status,next_attempt_at,created_at); CREATE TABLE IF NOT EXISTS audit ( sequence INTEGER PRIMARY KEY AUTOINCREMENT, at TEXT NOT NULL, action TEXT NOT NULL, client_event_id TEXT, reason TEXT ); """ class RightsholdOperations: """SQLite-backed, single-writer outbox. Key material lives only in memory. enabled=False is inert: no files, evidence inspection, or network. An enabled client may enqueue offline without a key. flush/reconcile need a key. Local storage errors are raised because claiming durability without it loses work. Network errors are returned by flush/reconcile while records remain pending. """ def __init__(self, directory: str, workspace_id: str, stream_id: str, *, api_key: Optional[str] = None, base_url: str = "https://www.rightshold.com", enabled: bool = False, timeout_seconds: float = 5, batch_size: int = 50, max_records: int = 10000, max_record_bytes: int = 16384, max_disk_bytes: int = 64 * 1024 * 1024, max_audit_entries: int = 50000, backoff_ms: int = 1000, max_backoff_ms: int = 300000): if not isinstance(directory, str) or not os.path.isabs(directory): raise OperationsError("invalid_options") self.directory = Path(os.path.abspath(directory)) self.scope = {"workspace_id": _identifier(workspace_id, r"opw_[a-f0-9]{24}"), "stream_id": _identifier(stream_id, r"opst_[a-f0-9]{24}"), "base_url": _origin(base_url)} self.enabled = enabled self._key = os.environ.get("RIGHTSHOLD_API_KEY", "") if api_key is None else api_key if not isinstance(enabled, bool) or not isinstance(self._key, str): raise OperationsError("invalid_options") if enabled and self._key and self._key in _json(self.scope): raise OperationsError("credential_in_record") self.timeout_seconds = timeout_seconds self.batch_size = batch_size self.max_records = max_records self.max_record_bytes = max_record_bytes self.max_disk_bytes = max_disk_bytes self.max_audit_entries = max_audit_entries self.backoff_ms, self.max_backoff_ms = backoff_ms, max_backoff_ms ranges = ((batch_size, 1, 100), (max_records, 1, 1000000), (max_record_bytes, 256, 1024 * 1024), (max_disk_bytes, 65536, 1024 * 1024 * 1024), (max_audit_entries, 1, 1000000), (backoff_ms, 1, 300000), (max_backoff_ms, backoff_ms, 86400000)) if any(isinstance(n, bool) or not isinstance(n, int) or n < low or n > high for n, low, high in ranges) or \ isinstance(timeout_seconds, bool) or not isinstance(timeout_seconds, (int, float)) or not math.isfinite(timeout_seconds) or not 0 < timeout_seconds <= 30: raise OperationsError("invalid_options") self._mutex = threading.RLock() self._db = None self._writer = None self._closed = False self._busy = False self._opener = urllib.request.build_opener(_NoRedirect()) if not enabled: return try: if not self.directory.exists(): self.directory.mkdir(mode=0o700) _private_path(self.directory, True) self._writer = _WriterLock(self.directory / ".python-writer.lock") file = self.directory / "outbox.sqlite3" if file.exists() or file.is_symlink(): _private_path(file, False) else: descriptor = os.open(str(file), os.O_RDWR | os.O_CREAT | os.O_EXCL, 0o600) os.close(descriptor) self._db = sqlite3.connect(str(file), isolation_level=None, timeout=0.25, check_same_thread=False) self._db.row_factory = sqlite3.Row self._db.execute("PRAGMA journal_mode=DELETE") self._db.execute("PRAGMA synchronous=FULL") self._db.execute("PRAGMA foreign_keys=ON") page_size = self._db.execute("PRAGMA page_size").fetchone()[0] page_budget = max(1, (self.max_disk_bytes - 8192) // (2 * (page_size + 8))) self._db.execute("PRAGMA max_page_count=" + str(page_budget)) if self._db.execute("PRAGMA quick_check").fetchone()[0] != "ok": raise OperationsError("storage_corrupt") self._db.executescript(_SCHEMA) with self._transaction(): metadata = self._db.execute("SELECT * FROM metadata WHERE id=1").fetchone() if metadata is None: self._db.execute("INSERT INTO metadata VALUES (1,?,0,?)", (_json(self.scope), _now())) elif metadata["scope"] != _json(self.scope): raise OperationsError("scope_mismatch") for row in self._db.execute("SELECT * FROM records"): record = self._row(row) if record["request_sha256"] != _digest(record["body"]): raise OperationsError("storage_corrupt") if self._key and self._key in _json(record): raise OperationsError("credential_in_record") self._limits() except Exception as error: self.close() if isinstance(error, OperationsError): raise code = "backpressure" if isinstance(error, sqlite3.DatabaseError) and getattr(error, "sqlite_errorcode", None) == getattr(sqlite3, "SQLITE_FULL", 13) else \ "storage_corrupt" if isinstance(error, sqlite3.DatabaseError) else "storage_error" raise OperationsError(code) from None def __enter__(self): return self def __exit__(self, *args): self.close() def close(self): with self._mutex: if self._busy: raise OperationsError("outbox_busy") if self._db is not None: self._db.close() self._db = None if self._writer is not None: self._writer.close() self._writer = None self._closed = True def _guard(self) -> bool: if self._closed: raise OperationsError("closed") return self.enabled @contextlib.contextmanager def _transaction(self): try: self._db.execute("BEGIN IMMEDIATE") except sqlite3.DatabaseError: raise OperationsError("storage_error") from None try: yield self._db.execute("COMMIT") except BaseException as error: try: self._db.execute("ROLLBACK") except sqlite3.DatabaseError: pass if isinstance(error, sqlite3.DatabaseError): raise OperationsError("backpressure" if getattr(error, "sqlite_errorcode", None) == getattr(sqlite3, "SQLITE_FULL", 13) else "storage_error") from None raise def _limits(self): if self._db.execute("SELECT COUNT(*) FROM records").fetchone()[0] > self.max_records or \ self._db.execute("SELECT COUNT(*) FROM audit").fetchone()[0] > self.max_audit_entries: raise OperationsError("backpressure") pages = self._db.execute("PRAGMA page_count").fetchone()[0] size = self._db.execute("PRAGMA page_size").fetchone()[0] # Reserve the database, a full rollback journal, its per-page framing, # and lock/header bytes before acknowledging any durable mutation. if pages * (size + 8) * 2 + 8192 > self.max_disk_bytes: raise OperationsError("backpressure") def _audit(self, action: str, client_event_id: Optional[str] = None, reason: Optional[str] = None): self._db.execute("INSERT INTO audit(at,action,client_event_id,reason) VALUES (?,?,?,?)", (_now(), action, client_event_id, reason)) self._db.execute("UPDATE metadata SET revision=revision+1 WHERE id=1") self._limits() def _reason(self, reason: str): if not isinstance(reason, str) or not reason.strip() or len(reason) > 2048: raise OperationsError("invalid_reason") if self._key and self._key in reason: raise OperationsError("credential_in_record") @staticmethod def _row(row) -> dict: result = dict(row) for key in ("body", "last_error", "receipt"): result[key] = None if result[key] is None else json.loads(result[key]) result["uncertain"] = bool(result["uncertain"]) return result def _input(self, client_event_id: str, action_type: str, occurred_at: Optional[str], outcome: Optional[str], severity: Optional[str], summary: Optional[str], tags: Optional[list], evidence: Any, action_sha256: Optional[str]) -> dict: if not isinstance(client_event_id, str) or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:@/\-]{0,159}", client_event_id) or \ not isinstance(action_type, str) or len(action_type) > 96 or not re.fullmatch(r"[a-z][a-z0-9]*(?:[._:-][a-z0-9]+)*", action_type): raise OperationsError("invalid_event") if evidence is not None and action_sha256 is not None: raise OperationsError("ambiguous_evidence") existing = self._db.execute("SELECT body FROM records WHERE client_event_id=?", (client_event_id,)).fetchone() occurred_at = occurred_at or (json.loads(existing[0])["occurred_at"] if existing else _now()) try: if not isinstance(occurred_at, str) or not re.fullmatch(r"[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}(?:\.[0-9]{1,3})?(?:Z|[+-][0-9]{2}:[0-9]{2})", occurred_at): raise ValueError() datetime.fromisoformat(occurred_at.replace("Z", "+00:00")) except (ValueError, TypeError): raise OperationsError("invalid_event") from None digest = hash_operation_evidence(evidence) if evidence is not None else action_sha256 if digest is not None and (not isinstance(digest, str) or not re.fullmatch(r"[a-f0-9]{64}", digest)) or \ outcome is not None and outcome not in ("success", "failure", "unknown") or \ severity is not None and severity not in ("debug", "info", "warning", "error", "critical") or \ summary is not None and (not isinstance(summary, str) or len(summary) > 2048 or re.search(r"[\x00-\x08\x0b\x0c\x0e-\x1f]", summary)) or \ tags is not None and (not isinstance(tags, list) or len(tags) > 16 or any(not isinstance(t, str) or not t.strip() or len(t) > 64 or re.search(r"[\x00-\x08\x0b\x0c\x0e-\x1f]", t) for t in tags)): raise OperationsError("invalid_event") body = {"client_event_id": client_event_id, "action_type": action_type, "occurred_at": occurred_at} body.update({k: v for k, v in {"outcome": outcome, "severity": severity, "summary": summary, "tags": tags, "action_sha256": digest}.items() if v is not None}) encoded = _json(body) if len(encoded.encode("utf-8")) > self.max_record_bytes: raise OperationsError("backpressure") if self._key and self._key in encoded: raise OperationsError("credential_in_record") return body def enqueue(self, client_event_id: str, action_type: str, *, occurred_at: Optional[str] = None, outcome: Optional[str] = None, severity: Optional[str] = None, summary: Optional[str] = None, tags: Optional[list] = None, evidence: Any = None, action_sha256: Optional[str] = None): """Durably queue a caller-owned intent/completion claim without networking. Reusing an id with identical fields returns its current local state. An id with changed fields is rejected. Evidence itself is never stored here.""" with self._mutex: if not self._guard(): return None body = self._input(client_event_id, action_type, occurred_at, outcome, severity, summary, tags, evidence, action_sha256) digest = _digest(body) previous = self._db.execute("SELECT * FROM records WHERE client_event_id=?", (client_event_id,)).fetchone() if previous: if previous["request_sha256"] != digest: raise OperationsError("event_conflict") return self._row(previous) stamp = _now() try: with self._transaction(): self._db.execute("INSERT INTO records(client_event_id,request_sha256,body,status,created_at,updated_at) VALUES (?,?,?,'pending',?,?)", (client_event_id, digest, _json(body), stamp, stamp)) self._audit("enqueue", client_event_id) except sqlite3.DatabaseError: raise OperationsError("storage_error") from None return self.local(client_event_id) def local(self, client_event_id: str): with self._mutex: if not self._guard(): return None row = self._db.execute("SELECT * FROM records WHERE client_event_id=?", (client_event_id,)).fetchone() return self._row(row) if row else None def status(self) -> dict: with self._mutex: counts = dict(pending=0, accepted=0, quarantined=0, dropped=0, uncertain=0) if not self._guard(): return dict(enabled=False, **counts, bytes=0, revision=0, audit_entries=0, oldest_pending_at=None) for row in self._db.execute("SELECT status,COUNT(*) AS n FROM records GROUP BY status"): counts[row["status"]] = row["n"] counts["uncertain"] = self._db.execute("SELECT COUNT(*) FROM records WHERE status='pending' AND uncertain=1").fetchone()[0] oldest = self._db.execute("SELECT MIN(created_at) FROM records WHERE status='pending'").fetchone()[0] return dict(enabled=True, **counts, bytes=(self.directory / "outbox.sqlite3").stat().st_size, revision=self._db.execute("SELECT revision FROM metadata WHERE id=1").fetchone()[0], audit_entries=self._db.execute("SELECT COUNT(*) FROM audit").fetchone()[0], oldest_pending_at=oldest) def local_export(self): with self._mutex: if not self._guard(): return None return {"spec": "rs-operations-local-export/1", "scope": dict(self.scope), "records": [self._row(r) for r in self._db.execute("SELECT * FROM records ORDER BY created_at,client_event_id")], "audit": [dict(r) for r in self._db.execute("SELECT * FROM audit ORDER BY sequence")]} def _manual(self, client_event_id: str, action: str, reason: str): with self._mutex: if not self._guard(): return if self._busy: raise OperationsError("outbox_busy") self._reason(reason) current = self.local(client_event_id) if current is None: raise OperationsError("event_not_found") if current["status"] == "accepted": raise OperationsError("immutable_receipt") with self._transaction(): self._db.execute("UPDATE records SET status=?,next_attempt_at=0,last_error=NULL,updated_at=? WHERE client_event_id=?", ("dropped" if action == "drop" else "pending", _now(), client_event_id)) self._audit(action, client_event_id, reason) def retry(self, client_event_id: str, reason: str): """Retry recording only. An ambiguous prior delivery is reconciled first.""" self._manual(client_event_id, "retry", reason) def drop(self, client_event_id: str, reason: str): """Stop delivery locally; this never deletes a server event or ledger entry.""" self._manual(client_event_id, "drop", reason) @property def _prefix(self): return "/v1/operations/workspaces/" + self.scope["workspace_id"] @property def _stream_prefix(self): return self._prefix + "/streams/" + self.scope["stream_id"] def _request(self, route: str, body: Optional[dict] = None): if not self._key: raise OperationsError("api_key_required") data = None if body is None else _json(body).encode("utf-8") headers = {"accept": "application/json", "authorization": "Bearer " + self._key} if data is not None: headers["content-type"] = "application/json" request = urllib.request.Request(self.scope["base_url"] + route, data=data, headers=headers, method="GET" if body is None else "POST") try: with self._opener.open(request, timeout=self.timeout_seconds) as response: content = response.read(4 * 1024 * 1024 + 1) if len(content) > 4 * 1024 * 1024: raise OperationsError("response_too_large") try: value = json.loads(content) except (ValueError, UnicodeError): raise OperationsError("invalid_response") from None if not isinstance(value, dict): raise OperationsError("invalid_response") return value except urllib.error.HTTPError as error: try: value = json.loads(error.read(65536)) code = value.get("error") if isinstance(value, dict) else None except (ValueError, UnicodeError, OSError, http.client.HTTPException): code = None finally: error.close() code = code if isinstance(code, str) and code in _HTTP_CODES and not (self._key and self._key in code) else "http_error" raise OperationsError(code, error.code) from None except OperationsError: raise except (OSError, urllib.error.URLError, http.client.HTTPException): raise OperationsError("network_error") from None def _receipt(self, value: Any, client_id: str) -> dict: if not isinstance(value, dict) or value.get("client_event_id") != client_id or \ not isinstance(value.get("id"), str) or not re.fullmatch(r"act_[a-f0-9]{20}", value["id"]) or \ isinstance(value.get("sequence"), bool) or not isinstance(value.get("sequence"), int) or value["sequence"] < 1 or \ not isinstance(value.get("event_hash"), str) or not re.fullmatch(r"[a-f0-9]{64}", value["event_hash"]) or \ value.get("workspace_id", self.scope["workspace_id"]) != self.scope["workspace_id"] or \ value.get("stream_id", self.scope["stream_id"]) != self.scope["stream_id"]: raise OperationsError("invalid_response") seq = value.get("ledger_seq") if seq is not None and (isinstance(seq, bool) or not isinstance(seq, int) or seq < 1): raise OperationsError("invalid_response") receipt = {k: value[k] for k in ("id", "client_event_id", "sequence", "event_hash")} if "ledger_seq" in value: receipt["ledger_seq"] = seq if self._key and self._key in _json(receipt): raise OperationsError("invalid_response") return receipt @staticmethod def _error_view(error: OperationsError): return {"code": error.code, "status": error.status, "retryable": error.retryable} def _deferred(self, ids: list, error: OperationsError): with self._transaction(): for client_id in ids: current = self.local(client_id) if current and current["status"] == "pending": delay = min(self.max_backoff_ms, self.backoff_ms * 2 ** min(20, max(0, current["attempts"] - 1))) self._db.execute("UPDATE records SET last_error=?,next_attempt_at=?,updated_at=? WHERE client_event_id=?", (_json(self._error_view(error)), int(time.time() * 1000) + delay, _now(), client_id)) self._audit("delivery_deferred") def _pending(self, limit: int, force: bool = True): if isinstance(limit, bool) or not isinstance(limit, int) or not 1 <= limit <= 100: raise OperationsError("invalid_options") sql = "SELECT * FROM records WHERE status='pending'" + ("" if force else " AND next_attempt_at<=?") + " ORDER BY created_at,client_event_id LIMIT ?" values = (limit,) if force else (int(time.time() * 1000), limit) return [self._row(r) for r in self._db.execute(sql, values)] def _reconcile_records(self, records: list) -> int: if not records: return 0 body = {"client_event_ids": [r["client_event_id"] for r in records], "expected_events": [r["body"] for r in records]} route = self._stream_prefix + "/reconcile" if self._key.startswith("rso_"): route = "/v1/operations/reconcile" body.update(workspace_id=self.scope["workspace_id"], stream_id=self.scope["stream_id"]) response = self._request(route, body) results = response.get("results") if not isinstance(results, list) or len(results) != len(records): raise OperationsError("invalid_response") updates = {} expected = {r["client_event_id"] for r in records} for result in results: if not isinstance(result, dict) or result.get("client_event_id") not in expected or result["client_event_id"] in updates: raise OperationsError("invalid_response") if result.get("status") == "missing" and result.get("content_matches", "absent") is None: updates[result["client_event_id"]] = None elif result.get("status") == "conflict" and result.get("content_matches") is False: updates[result["client_event_id"]] = False elif result.get("status") == "recorded" and result.get("content_matches") is True: updates[result["client_event_id"]] = self._receipt(dict(result, id=result.get("event_id")), result["client_event_id"]) else: raise OperationsError("invalid_response") accepted = 0 with self._transaction(): for client_id, receipt in updates.items(): if receipt is False: self._db.execute("UPDATE records SET status='quarantined',uncertain=0,last_error=?,updated_at=? WHERE client_event_id=? AND status='pending'", (_json(self._error_view(OperationsError("operations_conflict", 409))), _now(), client_id)) elif receipt: self._db.execute("UPDATE records SET status='accepted',receipt=?,uncertain=0,last_error=NULL,updated_at=? WHERE client_event_id=? AND status='pending'", (_json(receipt), _now(), client_id)) accepted += 1 else: self._db.execute("UPDATE records SET uncertain=0,last_error=NULL,updated_at=? WHERE client_event_id=? AND status='pending'", (_now(), client_id)) self._audit("reconcile") return accepted def reconcile(self, *, limit: Optional[int] = None) -> dict: """Read server keys for a bounded pending set. Never submits new events.""" with self._mutex: report = dict(attempted=0, accepted=0, quarantined=0, pending=self.status()["pending"], reconciled=0, error=None) if not self._guard(): return report if self._busy: raise OperationsError("outbox_busy") self._busy = True try: records = self._pending(self.batch_size if limit is None else limit) quarantined_before = self.status()["quarantined"] report["attempted"] = len(records) try: report["accepted"] = report["reconciled"] = self._reconcile_records(records) except OperationsError as error: report["error"] = self._error_view(error) self._deferred([r["client_event_id"] for r in records], error) report["pending"] = self.status()["pending"] report["quarantined"] = self.status()["quarantined"] - quarantined_before return report finally: self._busy = False def flush(self, *, limit: Optional[int] = None, force: bool = False) -> dict: """Submit one partial batch. Transient/ambiguous failures remain pending. Persisted backoff gates retry; force bypasses time, never idempotency or reconciliation. Receipt failures never repeat the underlying work.""" with self._mutex: report = dict(attempted=0, accepted=0, quarantined=0, pending=self.status()["pending"], reconciled=0, error=None) if not self._guard(): return report if self._busy: raise OperationsError("outbox_busy") self._busy = True try: records = self._pending(self.batch_size if limit is None else limit, force) if not records: return report if any(r["uncertain"] for r in records): quarantined_before = self.status()["quarantined"] try: report["accepted"] = report["reconciled"] = self._reconcile_records([r for r in records if r["uncertain"]]) except OperationsError as error: report["error"] = self._error_view(error) self._deferred([r["client_event_id"] for r in records], error) return report records = [self.local(r["client_event_id"]) for r in records] records = [r for r in records if r["status"] == "pending"] report["quarantined"] += self.status()["quarantined"] - quarantined_before if not records: report["pending"] = self.status()["pending"] return report ids = [r["client_event_id"] for r in records] with self._transaction(): for client_id in ids: self._db.execute("UPDATE records SET attempts=attempts+1,uncertain=1,updated_at=? WHERE client_event_id=?", (_now(), client_id)) self._audit("delivery_attempt") report["attempted"] = len(records) body = {"mode": "partial", "events": [r["body"] for r in records]} # Each explicit delivery attempt gets a new batch key. A complete # batch acknowledgement can permanently include rejected items; # reusing its key after a pause/budget change would replay rejection. # Stable event keys still prevent duplicate ledger records. body["idempotency_key"] = "outbox:" + _digest({"body": body, "attempts": [r["attempts"] + 1 for r in records]}) route = self._stream_prefix + "/events/batch" if self._key.startswith("rso_"): route = "/v1/operations/ingest" body.update(workspace_id=self.scope["workspace_id"], stream_id=self.scope["stream_id"]) try: response = self._request(route, body) results = response.get("results") if not isinstance(results, list) or len(results) != len(records): raise OperationsError("invalid_response") updates = {} for result in results: if not isinstance(result, dict) or isinstance(result.get("index"), bool) or not isinstance(result.get("index"), int) or not 0 <= result["index"] < len(records): raise OperationsError("invalid_response") target = records[result["index"]]["client_event_id"] if result.get("client_event_id") != target or target in updates: raise OperationsError("invalid_response") if result.get("status") in ("accepted", "duplicate"): updates[target] = {"receipt": self._receipt(result.get("event"), target)} elif result.get("status") == "rejected" and isinstance(result.get("error"), dict): status = result["error"].get("status") status = status if isinstance(status, int) and not isinstance(status, bool) and 400 <= status <= 599 else None code = result["error"].get("code") code = code if isinstance(code, str) and code in _HTTP_CODES and not (self._key and self._key in code) else "http_error" updates[target] = {"error": OperationsError(code, status)} else: raise OperationsError("invalid_response") with self._transaction(): for client_id, result in updates.items(): if "receipt" in result: self._db.execute("UPDATE records SET status='accepted',receipt=?,uncertain=0,last_error=NULL,updated_at=? WHERE client_event_id=?", (_json(result["receipt"]), _now(), client_id)) report["accepted"] += 1 else: error = result["error"] terminal = error.status in (400, 409, 422) current = self.local(client_id) delay = min(self.max_backoff_ms, self.backoff_ms * 2 ** min(20, current["attempts"] - 1)) self._db.execute("UPDATE records SET status=?,uncertain=?,last_error=?,next_attempt_at=?,updated_at=? WHERE client_event_id=?", ("quarantined" if terminal else "pending", 0 if terminal else 1, _json(self._error_view(error)), int(time.time() * 1000) + delay, _now(), client_id)) report["quarantined"] += 1 if terminal else 0 self._audit("delivery_result") except OperationsError as error: report["error"] = self._error_view(error) self._deferred(ids, error) report["pending"] = self.status()["pending"] return report finally: self._busy = False def _explicit(self, route: str, body: Optional[dict] = None): with self._mutex: if not self._guard(): return None return self._request(route, body) def event(self, receipt_id: str): if not self._guard(): return None receipt_id = _identifier(receipt_id, r"act_[a-f0-9]{20}") route = "/v1/operations/records/" + self.scope["stream_id"] + "/" + receipt_id if self._key.startswith("rso_") else self._stream_prefix + "/events/" + receipt_id return self._explicit(route) def export_event(self, receipt_id: str): if not self._guard(): return None receipt_id = _identifier(receipt_id, r"act_[a-f0-9]{20}") route = "/v1/operations/records/" + self.scope["stream_id"] + "/" + receipt_id if self._key.startswith("rso_") else self._stream_prefix + "/events/" + receipt_id return self._explicit(route + "/export") def summary(self): return self._explicit(self._stream_prefix + "/summary") def cases(self): return self._explicit(self._prefix + "/cases") def create_case(self, body: dict): return self._explicit(self._prefix + "/cases", body) def case_action(self, case_id: str, action: str, body: dict): if action not in ("update", "assign", "events", "events/remove", "notes", "decision", "close", "reopen", "export"): raise OperationsError("invalid_options") return self._explicit(self._prefix + "/cases/" + _identifier(case_id, r"opcase_[a-f0-9]{24}") + "/" + action, body) def policies(self): return self._explicit(self._prefix + "/policies") def create_policy(self, body: dict): return self._explicit(self._prefix + "/policies", body) def evaluate_policy(self, policy_id: str, body: Optional[dict] = None): return self._explicit(self._prefix + "/policies/" + _identifier(policy_id, r"oppol_[a-f0-9]{24}") + "/evaluate", body or {}) def exceptions(self): return self._explicit(self._prefix + "/exceptions") def transition_exception(self, exception_id: str, body: dict): return self._explicit(self._prefix + "/exceptions/" + _identifier(exception_id, r"opex_[a-f0-9]{24}") + "/transition", body) def health(self): return self._explicit(self._prefix + "/health") class AsyncRightsholdOperations: """Async adapter; each operation runs in a thread under the same writer lock. Cancelling a waiting coroutine does not prove the local write or network call stopped. Reconcile the stable event id after cancellation or lost response. """ def __init__(self, *args, **kwargs): self.sync = RightsholdOperations(*args, **kwargs) async def __aenter__(self): return self async def __aexit__(self, *args): await self.close() async def _run(self, method: str, *args, **kwargs): import functools return await asyncio.to_thread(functools.partial(getattr(self.sync, method), *args, **kwargs)) async def enqueue(self, *args, **kwargs): return await self._run("enqueue", *args, **kwargs) async def flush(self, **kwargs): return await self._run("flush", **kwargs) async def reconcile(self, **kwargs): return await self._run("reconcile", **kwargs) async def retry(self, *args): return await self._run("retry", *args) async def drop(self, *args): return await self._run("drop", *args) async def local(self, *args): return await self._run("local", *args) async def status(self): return await self._run("status") async def local_export(self): return await self._run("local_export") async def event(self, *args): return await self._run("event", *args) async def export_event(self, *args): return await self._run("export_event", *args) async def summary(self): return await self._run("summary") async def cases(self): return await self._run("cases") async def create_case(self, *args): return await self._run("create_case", *args) async def case_action(self, *args): return await self._run("case_action", *args) async def policies(self): return await self._run("policies") async def create_policy(self, *args): return await self._run("create_policy", *args) async def evaluate_policy(self, *args): return await self._run("evaluate_policy", *args) async def exceptions(self): return await self._run("exceptions") async def transition_exception(self, *args): return await self._run("transition_exception", *args) async def health(self): return await self._run("health") async def close(self): return await self._run("close") def inspect_operations_outbox(directory: str, max_disk_bytes: int = 64 * 1024 * 1024) -> dict: """Read-only inspection: no lock, initialization, writes, repair, or network.""" db = None try: if not isinstance(directory, str) or not os.path.isabs(directory): raise OperationsError("invalid_options") folder = Path(directory) if not folder.exists() and not folder.is_symlink(): return {"exists": False, "secure": True, "records": 0} _private_path(folder, True) file = folder / "outbox.sqlite3" if not file.exists(): return {"exists": False, "secure": True, "records": 0} _private_path(file, False) if file.stat().st_size > max_disk_bytes: raise OperationsError("backpressure") # Read-only URI plus query_only prevents a status command from creating a # journal, writing a checkpoint, or initializing a missing database. db = sqlite3.connect(file.as_uri() + "?mode=ro", uri=True) db.execute("PRAGMA query_only=ON") if db.execute("PRAGMA quick_check").fetchone()[0] != "ok": raise OperationsError("storage_corrupt") scope, revision = db.execute("SELECT scope,revision FROM metadata WHERE id=1").fetchone() for body, digest in db.execute("SELECT body,request_sha256 FROM records"): if _digest(json.loads(body)) != digest: raise OperationsError("storage_corrupt") counts = dict(pending=0, accepted=0, quarantined=0, dropped=0) for status, count in db.execute("SELECT status,COUNT(*) FROM records GROUP BY status"): if status not in counts: raise OperationsError("storage_corrupt") counts[status] = count return dict(exists=True, secure=True, spec="rs-operations-outbox/1", scope=json.loads(scope), revision=revision, **counts, bytes=file.stat().st_size, audit_entries=db.execute("SELECT COUNT(*) FROM audit").fetchone()[0]) except Exception as error: safe = error if isinstance(error, OperationsError) else OperationsError("storage_corrupt" if isinstance(error, sqlite3.DatabaseError) else "storage_error") return {"exists": Path(directory).exists() if isinstance(directory, str) else False, "secure": False, "error": safe.code} finally: if db is not None: db.close() def operations_cli(argv: Optional[list] = None) -> int: parser = argparse.ArgumentParser(description="Rightshold operations: explicit local storage and delivery.") parser.add_argument("command", choices=("init", "status", "enqueue", "flush", "reconcile", "export")) parser.add_argument("config", help="Absolute path to a private JSON configuration; excludes credentials.") parser.add_argument("input", nargs="?", help="Absolute path to private JSON event fields for enqueue.") args = parser.parse_args(argv) try: if not os.path.isabs(args.config): raise OperationsError("invalid_config") config = json.loads(_safe_read(Path(args.config), 65536)) allowed = {"directory", "workspace_id", "stream_id", "base_url", "timeout_seconds", "batch_size", "max_records", "max_record_bytes", "max_disk_bytes", "max_audit_entries", "backoff_ms", "max_backoff_ms"} if not isinstance(config, dict) or set(config) - allowed: raise OperationsError("invalid_config") if args.command == "status": result = inspect_operations_outbox(config["directory"]) print(_json(result)) return 1 if result.get("error") else 0 with RightsholdOperations(enabled=True, **config) as client: if args.command == "init": result = client.status() elif args.command == "enqueue": if args.input is None or not os.path.isabs(args.input): raise OperationsError("invalid_command") event = json.loads(_safe_read(Path(args.input), 1024 * 1024)) record = client.enqueue(**event) result = {k: record[k] for k in ("client_event_id", "status", "request_sha256")} elif args.command == "flush": result = client.flush() elif args.command == "reconcile": result = client.reconcile() else: result = client.local_export() print(_json(result)) return 0 except Exception as error: safe = error if isinstance(error, OperationsError) else OperationsError("invalid_config") print(_json({"error": safe.code, "status": safe.status}), file=sys.stderr) return 1 if __name__ == "__main__": sys.exit(operations_cli())