#!/usr/bin/env python3 """Private local journal for the Hermes opportunity-sequence skill. This module intentionally has no provider client, network transport, Lightfield reader, Slack client, business transition chooser, or copy generator. """ from __future__ import annotations import argparse import contextlib import datetime as dt import hashlib import json import os from pathlib import Path import re import sqlite3 import subprocess import sys from typing import Any, Iterator SCHEMA_VERSION = "opportunity-sequence-journal/v1" SNAPSHOT_VERSION = "opportunity-sequence-journal-snapshot/v1" DEFAULT_ROOT = Path("/opt/data/.sellable-agent/admin-runtime/opportunity-sequences") DB_NAME = "journal.sqlite3" MAX_SNAPSHOT_BYTES = 1_000_000 MAX_SNAPSHOT_ROWS = 500 SAFE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$") SHA256 = re.compile(r"^[a-f0-9]{64}$") PLAYBOOKS = ( ( "outbound-multichannel@2#540a5213fee2ba62d8c54d9f92a5f32d5e27fc232f6b3b81a7e12ccb13b56034", "540a5213fee2ba62d8c54d9f92a5f32d5e27fc232f6b3b81a7e12ccb13b56034", ), ( "positive-reply-to-agreed-meeting@1#6083f2db0e859a81e78456ae6677658abb937636c420bee16407c942a25ca5ff", "6083f2db0e859a81e78456ae6677658abb937636c420bee16407c942a25ca5ff", ), ) PUBLIC_VERBS = ( "begin-run", "admit", "advance-cursor", "notification", "proposal-evidence", "status", "replay", "rebuild", ) class JournalError(Exception): pass def now_iso() -> str: return dt.datetime.now(dt.timezone.utc).isoformat().replace("+00:00", "Z") def canonical_json(value: Any) -> str: return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True) def digest(value: Any) -> str: return hashlib.sha256(canonical_json(value).encode("utf-8")).hexdigest() def require_id(value: str, label: str) -> str: if not SAFE_ID.fullmatch(value or ""): raise JournalError(f"{label}_invalid") return value def require_hash(value: str, label: str) -> str: if not SHA256.fullmatch(value or ""): raise JournalError(f"{label}_invalid") return value def parse_time(value: str | None, label: str) -> str | None: if value is None: return None try: parsed = dt.datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError as error: raise JournalError(f"{label}_invalid") from error if parsed.tzinfo is None: raise JournalError(f"{label}_invalid") return parsed.astimezone(dt.timezone.utc).isoformat().replace("+00:00", "Z") def later(current: str | None, offered: str | None) -> str | None: if offered is None: return current if current is None: return offered return max(current, offered, key=lambda value: dt.datetime.fromisoformat(value.replace("Z", "+00:00"))) def exact_keys(value: dict[str, Any], keys: set[str], label: str) -> None: if set(value) != keys: raise JournalError(f"{label}_shape_invalid") def private_root(root: Path) -> Path: path = root.expanduser().resolve() if path.exists() and path.is_symlink(): raise JournalError("journal_root_symlink_rejected") path.mkdir(parents=True, exist_ok=True, mode=0o700) os.chmod(path, 0o700) if path.stat().st_mode & 0o077: raise JournalError("journal_root_mode_rejected") return path def init_schema(connection: sqlite3.Connection) -> None: connection.executescript( """ CREATE TABLE IF NOT EXISTS meta ( key TEXT PRIMARY KEY, value TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS runs ( run_id TEXT PRIMARY KEY, mode TEXT NOT NULL CHECK(mode IN ('cron','operator')), started_at TEXT NOT NULL, status TEXT NOT NULL CHECK(status IN ('running','complete','blocked')) ); CREATE TABLE IF NOT EXISTS admissions ( semantic_key TEXT PRIMARY KEY, run_id TEXT NOT NULL, item_kind TEXT NOT NULL, source_fingerprint TEXT NOT NULL, reason_codes_json TEXT NOT NULL, admitted_at TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'admitted' ); CREATE INDEX IF NOT EXISTS admissions_run_idx ON admissions(run_id, admitted_at); CREATE TABLE IF NOT EXISTS progress ( scope TEXT PRIMARY KEY, continuation TEXT, checkpoint TEXT, inbound_watermark TEXT, opportunity_source_high TEXT, inbound_source_high TEXT, updated_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS notifications ( attempt_id TEXT PRIMARY KEY, call_request_id TEXT NOT NULL, parent_attempt_id TEXT, status TEXT NOT NULL CHECK(status IN ('prepared','emission_started','observed','uncertain')), created_at TEXT NOT NULL, updated_at TEXT NOT NULL, FOREIGN KEY(parent_attempt_id) REFERENCES notifications(attempt_id) ); CREATE UNIQUE INDEX IF NOT EXISTS notification_initial_request_idx ON notifications(call_request_id) WHERE parent_attempt_id IS NULL; CREATE TABLE IF NOT EXISTS proposals ( proposal_key TEXT PRIMARY KEY, source_hash TEXT NOT NULL, evidence_start TEXT NOT NULL, evidence_end TEXT NOT NULL, high_water TEXT NOT NULL, sample_count INTEGER NOT NULL CHECK(sample_count >= 0), status TEXT NOT NULL CHECK(status IN ('observation','proposed')), markdown_hash TEXT, created_at TEXT NOT NULL ); """ ) connection.execute( "INSERT INTO meta(key,value) VALUES('schema_version',?) ON CONFLICT(key) DO NOTHING", (SCHEMA_VERSION,), ) observed = connection.execute("SELECT value FROM meta WHERE key='schema_version'").fetchone() if observed is None or observed[0] != SCHEMA_VERSION: raise JournalError("journal_schema_rejected") @contextlib.contextmanager def database(root: Path) -> Iterator[sqlite3.Connection]: directory = private_root(root) path = directory / DB_NAME if path.exists() and path.is_symlink(): raise JournalError("journal_database_symlink_rejected") connection = sqlite3.connect(path, timeout=5.0, isolation_level=None) connection.row_factory = sqlite3.Row try: connection.execute("PRAGMA busy_timeout=5000") if connection.execute("PRAGMA journal_mode=WAL").fetchone()[0].lower() != "wal": raise JournalError("journal_wal_rejected") connection.execute("PRAGMA synchronous=FULL") connection.execute("PRAGMA foreign_keys=ON") init_schema(connection) os.chmod(path, 0o600) yield connection finally: connection.close() for candidate in (path, Path(f"{path}-wal"), Path(f"{path}-shm")): if candidate.exists() and not candidate.is_symlink(): os.chmod(candidate, 0o600) @contextlib.contextmanager def transaction(connection: sqlite3.Connection) -> Iterator[None]: connection.execute("BEGIN IMMEDIATE") try: yield except Exception: connection.execute("ROLLBACK") raise else: connection.execute("COMMIT") def begin_run(connection: sqlite3.Connection, run_id: str, mode: str) -> dict[str, Any]: require_id(run_id, "run_id") if mode not in {"cron", "operator"}: raise JournalError("run_mode_invalid") with transaction(connection): connection.execute( "INSERT INTO runs(run_id,mode,started_at,status) VALUES(?,?,?,'running') " "ON CONFLICT(run_id) DO NOTHING", (run_id, mode, now_iso()), ) row = connection.execute("SELECT * FROM runs WHERE run_id=?", (run_id,)).fetchone() if row["mode"] != mode: raise JournalError("run_identity_drift") return dict(row) def admit( connection: sqlite3.Connection, run_id: str, semantic_key: str, item_kind: str, source_fingerprint: str, reason_codes: list[str], ) -> dict[str, Any]: require_id(run_id, "run_id") require_id(semantic_key, "semantic_key") require_id(item_kind, "item_kind") require_hash(source_fingerprint, "source_fingerprint") normalized_reasons = sorted(set(require_id(code, "reason_code") for code in reason_codes)) if len(normalized_reasons) > 32: raise JournalError("reason_codes_bound_exceeded") reasons_json = canonical_json(normalized_reasons) with transaction(connection): if connection.execute("SELECT 1 FROM runs WHERE run_id=?", (run_id,)).fetchone() is None: raise JournalError("run_not_found") connection.execute( "INSERT INTO admissions(semantic_key,run_id,item_kind,source_fingerprint,reason_codes_json,admitted_at) " "VALUES(?,?,?,?,?,?) ON CONFLICT(semantic_key) DO NOTHING", (semantic_key, run_id, item_kind, source_fingerprint, reasons_json, now_iso()), ) row = connection.execute("SELECT * FROM admissions WHERE semantic_key=?", (semantic_key,)).fetchone() if ( row["item_kind"] != item_kind or row["source_fingerprint"] != source_fingerprint or row["reason_codes_json"] != reasons_json ): raise JournalError("admission_identity_drift") return {**dict(row), "reason_codes": json.loads(row["reason_codes_json"])} def advance_cursor(connection: sqlite3.Connection, values: dict[str, Any]) -> dict[str, Any]: scope = require_id(values["scope"], "scope") normalized = { "inbound_watermark": parse_time(values.get("inbound_watermark"), "inbound_watermark"), "opportunity_source_high": parse_time(values.get("opportunity_source_high"), "opportunity_source_high"), "inbound_source_high": parse_time(values.get("inbound_source_high"), "inbound_source_high"), } for opaque in (values.get("continuation"), values.get("checkpoint")): if opaque is not None and (not isinstance(opaque, str) or not 1 <= len(opaque) <= 8192): raise JournalError("opaque_progress_invalid") with transaction(connection): current = connection.execute("SELECT * FROM progress WHERE scope=?", (scope,)).fetchone() row = dict(current) if current else { "scope": scope, "continuation": None, "checkpoint": None, "inbound_watermark": None, "opportunity_source_high": None, "inbound_source_high": None, } row.update({ "continuation": values.get("continuation") or row["continuation"], "checkpoint": values.get("checkpoint") or row["checkpoint"], "inbound_watermark": later(row["inbound_watermark"], normalized["inbound_watermark"]), "opportunity_source_high": later(row["opportunity_source_high"], normalized["opportunity_source_high"]), "inbound_source_high": later(row["inbound_source_high"], normalized["inbound_source_high"]), "updated_at": now_iso(), }) connection.execute( "INSERT INTO progress(scope,continuation,checkpoint,inbound_watermark,opportunity_source_high,inbound_source_high,updated_at) " "VALUES(:scope,:continuation,:checkpoint,:inbound_watermark,:opportunity_source_high,:inbound_source_high,:updated_at) " "ON CONFLICT(scope) DO UPDATE SET continuation=excluded.continuation,checkpoint=excluded.checkpoint," "inbound_watermark=excluded.inbound_watermark,opportunity_source_high=excluded.opportunity_source_high," "inbound_source_high=excluded.inbound_source_high,updated_at=excluded.updated_at", row, ) return row NOTIFICATION_TRANSITIONS = { "prepared": {"emission_started"}, "emission_started": {"observed", "uncertain"}, "observed": set(), "uncertain": set(), } def notification( connection: sqlite3.Connection, attempt_id: str, call_request_id: str, transition: str, ) -> dict[str, Any]: require_id(attempt_id, "attempt_id") require_id(call_request_id, "call_request_id") if transition not in NOTIFICATION_TRANSITIONS: raise JournalError("notification_transition_invalid") with transaction(connection): row = connection.execute("SELECT * FROM notifications WHERE attempt_id=?", (attempt_id,)).fetchone() if row is None: if transition != "prepared": raise JournalError("notification_attempt_not_found") created = now_iso() connection.execute( "INSERT INTO notifications(attempt_id,call_request_id,parent_attempt_id,status,created_at,updated_at) " "VALUES(?,?,NULL,'prepared',?,?)", (attempt_id, call_request_id, created, created), ) else: if row["call_request_id"] != call_request_id: raise JournalError("notification_identity_drift") if row["status"] != transition: if transition not in NOTIFICATION_TRANSITIONS[row["status"]]: raise JournalError("notification_transition_refused") connection.execute( "UPDATE notifications SET status=?,updated_at=? WHERE attempt_id=?", (transition, now_iso(), attempt_id), ) row = connection.execute("SELECT * FROM notifications WHERE attempt_id=?", (attempt_id,)).fetchone() return dict(row) def replay(connection: sqlite3.Connection, attempt_id: str, new_attempt_id: str) -> dict[str, Any]: require_id(attempt_id, "attempt_id") require_id(new_attempt_id, "new_attempt_id") with transaction(connection): parent = connection.execute("SELECT * FROM notifications WHERE attempt_id=?", (attempt_id,)).fetchone() if parent is None or parent["status"] != "uncertain": raise JournalError("notification_replay_not_allowed") created = now_iso() connection.execute( "INSERT INTO notifications(attempt_id,call_request_id,parent_attempt_id,status,created_at,updated_at) " "VALUES(?,?,?,'prepared',?,?) ON CONFLICT(attempt_id) DO NOTHING", (new_attempt_id, parent["call_request_id"], attempt_id, created, created), ) row = connection.execute("SELECT * FROM notifications WHERE attempt_id=?", (new_attempt_id,)).fetchone() if row["call_request_id"] != parent["call_request_id"] or row["parent_attempt_id"] != attempt_id: raise JournalError("notification_replay_identity_drift") return dict(row) def proposal_evidence(connection: sqlite3.Connection, values: dict[str, Any]) -> dict[str, Any]: proposal_key = require_id(values["proposal_key"], "proposal_key") source_hash = require_hash(values["source_hash"], "source_hash") markdown_hash = values.get("markdown_hash") if markdown_hash is not None: require_hash(markdown_hash, "markdown_hash") status = values["status"] if status not in {"observation", "proposed"} or (status == "proposed") != (markdown_hash is not None): raise JournalError("proposal_status_invalid") start = parse_time(values["evidence_start"], "evidence_start") end = parse_time(values["evidence_end"], "evidence_end") high = parse_time(values["high_water"], "high_water") if start > end or end > high or not 0 <= values["sample_count"] <= 100_000: raise JournalError("proposal_evidence_invalid") record = { "proposal_key": proposal_key, "source_hash": source_hash, "evidence_start": start, "evidence_end": end, "high_water": high, "sample_count": values["sample_count"], "status": status, "markdown_hash": markdown_hash, "created_at": now_iso(), } with transaction(connection): connection.execute( "INSERT INTO proposals VALUES(:proposal_key,:source_hash,:evidence_start,:evidence_end,:high_water," ":sample_count,:status,:markdown_hash,:created_at) ON CONFLICT(proposal_key) DO NOTHING", record, ) row = connection.execute("SELECT * FROM proposals WHERE proposal_key=?", (proposal_key,)).fetchone() comparable = {key: record[key] for key in record if key != "created_at"} if any(row[key] != value for key, value in comparable.items()): raise JournalError("proposal_identity_drift") return dict(row) def status(connection: sqlite3.Connection) -> dict[str, Any]: count = lambda table: connection.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0] progress = [dict(row) for row in connection.execute("SELECT * FROM progress ORDER BY scope")] unsafe = [ dict(row) for row in connection.execute( "SELECT * FROM notifications WHERE status IN ('emission_started','uncertain') ORDER BY created_at,attempt_id" ) ] return { "schema_version": SCHEMA_VERSION, "runs": count("runs"), "admissions": count("admissions"), "notifications": count("notifications"), "proposals": count("proposals"), "progress": progress, "unsafe_notifications": unsafe, "ordinary_notification_retry_allowed": len(unsafe) == 0, } def validate_snapshot(value: Any) -> dict[str, Any]: if not isinstance(value, dict): raise JournalError("snapshot_shape_invalid") exact_keys(value, {"schema_version", "playbooks", "progress", "admissions", "notifications", "proposals", "snapshot_hash"}, "snapshot") if value["schema_version"] != SNAPSHOT_VERSION: raise JournalError("snapshot_schema_invalid") expected_playbooks = [{"ref": ref, "hash": hash_value} for ref, hash_value in PLAYBOOKS] if value["playbooks"] != expected_playbooks: raise JournalError("snapshot_playbooks_invalid") unsigned = {key: item for key, item in value.items() if key != "snapshot_hash"} if value["snapshot_hash"] != digest(unsigned): raise JournalError("snapshot_hash_invalid") for name in ("progress", "admissions", "notifications", "proposals"): if not isinstance(value[name], list) or len(value[name]) > MAX_SNAPSHOT_ROWS: raise JournalError("snapshot_bound_exceeded") for row in value["progress"]: exact_keys(row, {"scope", "continuation", "checkpoint", "inbound_watermark", "opportunity_source_high", "inbound_source_high", "updated_at"}, "snapshot_progress") require_id(row["scope"], "scope") for key in ("inbound_watermark", "opportunity_source_high", "inbound_source_high", "updated_at"): parse_time(row[key], key) for row in value["admissions"]: exact_keys(row, {"semantic_key", "run_id", "item_kind", "source_fingerprint", "reason_codes_json", "admitted_at", "status"}, "snapshot_admission") require_id(row["semantic_key"], "semantic_key") require_id(row["run_id"], "run_id") require_id(row["item_kind"], "item_kind") require_hash(row["source_fingerprint"], "source_fingerprint") if not isinstance(json.loads(row["reason_codes_json"]), list): raise JournalError("snapshot_reason_codes_invalid") parse_time(row["admitted_at"], "admitted_at") for row in value["notifications"]: exact_keys(row, {"attempt_id", "call_request_id", "parent_attempt_id", "status", "created_at", "updated_at"}, "snapshot_notification") require_id(row["attempt_id"], "attempt_id") require_id(row["call_request_id"], "call_request_id") if row["parent_attempt_id"] is not None: require_id(row["parent_attempt_id"], "parent_attempt_id") if row["status"] not in NOTIFICATION_TRANSITIONS: raise JournalError("snapshot_notification_status_invalid") parse_time(row["created_at"], "created_at") parse_time(row["updated_at"], "updated_at") for row in value["proposals"]: exact_keys(row, {"proposal_key", "source_hash", "evidence_start", "evidence_end", "high_water", "sample_count", "status", "markdown_hash", "created_at"}, "snapshot_proposal") require_id(row["proposal_key"], "proposal_key") require_hash(row["source_hash"], "source_hash") if row["markdown_hash"] is not None: require_hash(row["markdown_hash"], "markdown_hash") return value def rebuild(connection: sqlite3.Connection, snapshot: dict[str, Any]) -> dict[str, Any]: value = validate_snapshot(snapshot) with transaction(connection): for table in ("admissions", "notifications", "proposals", "progress", "runs"): connection.execute(f"DELETE FROM {table}") connection.execute("PRAGMA defer_foreign_keys=ON") for row in value["progress"]: connection.execute("INSERT INTO progress VALUES(:scope,:continuation,:checkpoint,:inbound_watermark,:opportunity_source_high,:inbound_source_high,:updated_at)", row) run_ids = sorted({row["run_id"] for row in value["admissions"]}) for run_id in run_ids: connection.execute("INSERT INTO runs VALUES(?, 'operator', ?, 'complete')", (run_id, now_iso())) for row in value["admissions"]: connection.execute("INSERT INTO admissions VALUES(:semantic_key,:run_id,:item_kind,:source_fingerprint,:reason_codes_json,:admitted_at,:status)", row) pending = list(value["notifications"]) while pending: inserted = 0 for row in pending[:]: if row["parent_attempt_id"] is None or connection.execute("SELECT 1 FROM notifications WHERE attempt_id=?", (row["parent_attempt_id"],)).fetchone(): connection.execute("INSERT INTO notifications VALUES(:attempt_id,:call_request_id,:parent_attempt_id,:status,:created_at,:updated_at)", row) pending.remove(row) inserted += 1 if inserted == 0: raise JournalError("snapshot_notification_parent_invalid") for row in value["proposals"]: connection.execute("INSERT INTO proposals VALUES(:proposal_key,:source_hash,:evidence_start,:evidence_end,:high_water,:sample_count,:status,:markdown_hash,:created_at)", row) connection.execute("INSERT INTO meta(key,value) VALUES('last_rebuild_hash',?) ON CONFLICT(key) DO UPDATE SET value=excluded.value", (value["snapshot_hash"],)) return {"rebuilt": True, "snapshot_hash": value["snapshot_hash"], **status(connection)} def snapshot_from_database(connection: sqlite3.Connection) -> dict[str, Any]: value = { "schema_version": SNAPSHOT_VERSION, "playbooks": [{"ref": ref, "hash": hash_value} for ref, hash_value in PLAYBOOKS], "progress": [dict(row) for row in connection.execute("SELECT * FROM progress ORDER BY scope")], "admissions": [dict(row) for row in connection.execute("SELECT * FROM admissions ORDER BY semantic_key")], "notifications": [dict(row) for row in connection.execute("SELECT * FROM notifications ORDER BY created_at,attempt_id")], "proposals": [dict(row) for row in connection.execute("SELECT * FROM proposals ORDER BY proposal_key")], } return {**value, "snapshot_hash": digest(value)} def self_test(root: Path) -> dict[str, Any]: with database(root) as connection: begin_run(connection, "self-test-run", "operator") commands = [] for key in ("item-a", "item-a", "item-b", "item-c"): commands.append([ sys.executable, str(Path(__file__).resolve()), "admit", "--root", str(root), "--run-id", "self-test-run", "--semantic-key", key, "--item-kind", "due_work", "--source-fingerprint", "a" * 64, "--reason-code", "step_due", ]) children = [subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) for command in commands] for child in children: stdout, stderr = child.communicate(timeout=10) if child.returncode != 0: raise JournalError(f"self_test_child_failed:{digest([stdout, stderr])}") with database(root) as connection: if status(connection)["admissions"] != 3: raise JournalError("self_test_admission_convergence_failed") advance_cursor(connection, { "scope": "default", "continuation": "opaque-next", "checkpoint": "opaque-checkpoint", "inbound_watermark": "2026-08-11T10:00:00Z", "opportunity_source_high": "2026-08-11T10:00:00Z", "inbound_source_high": "2026-08-11T10:00:00Z", }) progress = advance_cursor(connection, { "scope": "default", "inbound_watermark": "2026-08-11T09:00:00Z", "opportunity_source_high": "2026-08-11T09:00:00Z", "inbound_source_high": "2026-08-11T09:00:00Z", }) if progress["inbound_watermark"] != "2026-08-11T10:00:00Z" or progress["continuation"] != "opaque-next": raise JournalError("self_test_progress_regressed") notification(connection, "notice-1", "lf-call-request-1", "prepared") notification(connection, "notice-1", "lf-call-request-1", "emission_started") if status(connection)["ordinary_notification_retry_allowed"]: raise JournalError("self_test_blind_retry_allowed") notification(connection, "notice-1", "lf-call-request-1", "uncertain") replayed = replay(connection, "notice-1", "notice-2") if replayed["call_request_id"] != "lf-call-request-1" or replayed["parent_attempt_id"] != "notice-1": raise JournalError("self_test_replay_identity_failed") proposal_evidence(connection, { "proposal_key": "proposal-1", "source_hash": "b" * 64, "evidence_start": "2026-08-01T00:00:00Z", "evidence_end": "2026-08-10T00:00:00Z", "high_water": "2026-08-11T00:00:00Z", "sample_count": 4, "status": "observation", "markdown_hash": None, }) snapshot = snapshot_from_database(connection) for suffix in ("", "-wal", "-shm"): path = private_root(root) / f"{DB_NAME}{suffix}" if path.exists(): path.unlink() with database(root) as connection: rebuilt = rebuild(connection, snapshot) if rebuilt["admissions"] != 3 or rebuilt["notifications"] != 2 or rebuilt["progress"][0]["continuation"] != "opaque-next": raise JournalError("self_test_rebuild_failed") rebuild(connection, snapshot) malformed = {**snapshot, "endpoint": "forbidden"} try: rebuild(connection, malformed) except JournalError: pass else: raise JournalError("self_test_snapshot_shape_failed") directory = private_root(root) db_path = directory / DB_NAME if directory.stat().st_mode & 0o077 or db_path.stat().st_mode & 0o077: raise JournalError("self_test_private_mode_failed") return { "ok": True, "schema_version": SCHEMA_VERSION, "public_verbs": list(PUBLIC_VERBS), "admissions": 3, "notification_attempts": 2, "same_lightfield_call_request": True, "rebuild_idempotent": True, "network_transports": 0, } def extract_root(argv: list[str]) -> tuple[Path, list[str]]: args = list(argv) root = DEFAULT_ROOT if "--root" in args: index = args.index("--root") if index + 1 >= len(args): raise JournalError("root_missing") root = Path(args[index + 1]) del args[index : index + 2] return root, args def parser_for(command: str) -> argparse.ArgumentParser: parser = argparse.ArgumentParser(prog=f"journal.py {command}") if command == "begin-run": parser.add_argument("--run-id", required=True) parser.add_argument("--mode", choices=("cron", "operator"), default="cron") elif command == "admit": parser.add_argument("--run-id", required=True) parser.add_argument("--semantic-key", required=True) parser.add_argument("--item-kind", required=True) parser.add_argument("--source-fingerprint", required=True) parser.add_argument("--reason-code", action="append", default=[]) elif command == "advance-cursor": parser.add_argument("--scope", default="default") parser.add_argument("--continuation") parser.add_argument("--checkpoint") parser.add_argument("--inbound-watermark") parser.add_argument("--opportunity-source-high") parser.add_argument("--inbound-source-high") elif command == "notification": parser.add_argument("--attempt-id", required=True) parser.add_argument("--call-request-id", required=True) parser.add_argument("--transition", choices=tuple(NOTIFICATION_TRANSITIONS), required=True) elif command == "proposal-evidence": parser.add_argument("--proposal-key", required=True) parser.add_argument("--source-hash", required=True) parser.add_argument("--evidence-start", required=True) parser.add_argument("--evidence-end", required=True) parser.add_argument("--high-water", required=True) parser.add_argument("--sample-count", type=int, required=True) parser.add_argument("--status", choices=("observation", "proposed"), required=True) parser.add_argument("--markdown-hash") elif command == "replay": parser.add_argument("--attempt-id", required=True) parser.add_argument("--new-attempt-id", required=True) elif command == "rebuild": parser.add_argument("--snapshot", required=True) elif command not in {"status", "self-test"}: raise JournalError("public_verb_invalid") return parser def main(argv: list[str]) -> int: try: root, remaining = extract_root(argv) if not remaining: raise JournalError("public_verb_required") command = remaining[0] if command not in {*PUBLIC_VERBS, "self-test"}: raise JournalError("public_verb_invalid") options = vars(parser_for(command).parse_args(remaining[1:])) if command == "self-test": result = self_test(root) else: with database(root) as connection: if command == "begin-run": result = begin_run(connection, options["run_id"], options["mode"]) elif command == "admit": result = admit(connection, options["run_id"], options["semantic_key"], options["item_kind"], options["source_fingerprint"], options["reason_code"]) elif command == "advance-cursor": result = advance_cursor(connection, options) elif command == "notification": result = notification(connection, options["attempt_id"], options["call_request_id"], options["transition"]) elif command == "proposal-evidence": result = proposal_evidence(connection, options) elif command == "status": result = status(connection) elif command == "replay": result = replay(connection, options["attempt_id"], options["new_attempt_id"]) else: snapshot_path = Path(options["snapshot"]) if snapshot_path.is_symlink() or not snapshot_path.is_file() or snapshot_path.stat().st_size > MAX_SNAPSHOT_BYTES: raise JournalError("snapshot_file_rejected") result = rebuild(connection, json.loads(snapshot_path.read_text("utf-8"))) print(canonical_json({"ok": True, "result": result})) return 0 except (JournalError, json.JSONDecodeError, sqlite3.Error, OSError) as error: code = str(error) if isinstance(error, JournalError) else "journal_failed" print(canonical_json({"ok": False, "error": code}), file=sys.stderr) return 2 if __name__ == "__main__": raise SystemExit(main(sys.argv[1:]))