#!/usr/bin/env python3
"""LockedIn FDE: real LangGraph SQLite checkpoint/restart lab, edition 3.
Python 3.11+, langgraph==1.2.16, langgraph-checkpoint==4.3.0,
langgraph-checkpoint-sqlite==3.1.2. Use a fresh disposable local SQLite file.

No model, network, sender, credentials or real customer data. Tenant/reviewer
flags are synthetic host inputs, NOT authentication. Thread naming is not access
control. A local checkpoint is not evidence for distributed execution, hard-kill
recovery, schema migration or an external exactly-once effect.
"""
import argparse
import hashlib
import json
import os
from pathlib import Path
import socket
import subprocess
import sys
import tempfile
from typing import TypedDict
import unittest
from unittest.mock import patch

# Override inherited remote tracing; restrict checkpoint deserialization before import.
os.environ["LANGSMITH_TRACING"] = "false"
os.environ["LANGCHAIN_TRACING_V2"] = "false"
os.environ["LANGCHAIN_TRACING"] = "false"
os.environ["LANGGRAPH_STRICT_MSGPACK"] = "true"

from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, interrupt


class DraftState(TypedDict):
    tenant: str
    request: str
    source_version: int
    draft: str
    digest: str
    status: str


def proposal_digest(state):
    content = {key: state[key] for key in ("tenant", "request", "source_version", "draft")}
    return hashlib.sha256(json.dumps(content, sort_keys=True, separators=(",", ":")).encode()).hexdigest()


def draft(state: DraftState):
    proposed = {**state, "draft": "Operator draft: inspect the disconnected training unit."}
    return {"draft": proposed["draft"], "digest": proposal_digest(proposed), "status": "awaiting_review"}


def review(state: DraftState):
    # On resume this node re-enters. No side effect belongs before interrupt.
    packet = interrupt({"request": state["request"], "digest": state["digest"], "draft": state["draft"], "source_version": state["source_version"]})
    if not isinstance(packet, dict) or set(packet) != {"decision", "digest"} or packet["decision"] not in ("accepted", "rejected") or packet["digest"] != state["digest"]:
        raise ValueError("invalid_review_binding")
    return {"status": packet["decision"]}


class ReviewDependencyUnavailable(RuntimeError):
    """A deliberate local node failure, not an observed external outage."""


def build(checkpointer, calls=None, fail_review=False):
    calls = calls if calls is not None else []

    def traced_draft(state):
        calls.append("draft")
        return draft(state)

    def traced_review(state):
        calls.append("review")
        if fail_review:
            raise ReviewDependencyUnavailable("scripted_review_dependency_unavailable")
        return review(state)

    graph = StateGraph(DraftState)
    graph.add_node("draft", traced_draft)
    graph.add_node("review", traced_review)
    graph.add_edge(START, "draft")
    graph.add_edge("draft", "review")
    graph.add_edge("review", END)
    return graph.compile(checkpointer=checkpointer)


def config_for(tenant, request):
    # Host-derived scope in a deployed app. These CLI values only simulate it.
    thread = hashlib.sha256(json.dumps([tenant, request], separators=(",", ":")).encode()).hexdigest()
    return {"configurable": {"thread_id": thread}}


def observation(graph, config, calls, code="checkpoint-observed"):
    observed = graph.get_state(config)
    interrupted = any(task.interrupts for task in observed.tasks)
    error_nodes = [task.name for task in observed.tasks if task.error]
    return {"code": code, "status": observed.values["status"],
            "digest": observed.values["digest"], "source_version": observed.values["source_version"],
            "next": list(observed.next), "interrupted": interrupted,
            "task_error_nodes": error_nodes,
            "pending_failure": bool(observed.next and error_nodes and not interrupted),
            "node_calls": calls}


def execute(args):
    if args.command in ("resume", "retry") and args.deny_review:
        return {"code": "review-denied"}, 2
    config = config_for(args.tenant, args.request)
    with SqliteSaver.from_conn_string(args.db) as checkpointer:
        calls = []
        graph = build(checkpointer, calls, args.fail_before_review)
        snapshot = graph.get_state(config)
        values = snapshot.values
        if args.command == "start":
            if values:
                return {"code": "request-exists", "status": values["status"]}, 2
            try:
                graph.invoke({"tenant": args.tenant, "request": args.request, "source_version": args.source_version, "draft": "", "digest": "", "status": "new"}, config)
            except ReviewDependencyUnavailable:
                return observation(graph, config, calls, "node-failed"), 2
        elif not values:
            return {"code": "missing-thread"}, 2
        elif args.command in ("resume", "retry"):
            # Sequential local host checks, not a distributed compare-and-set.
            # Reauthorize and bind before passing a Command to the graph.
            if args.digest != values["digest"]:
                return {"code": "proposal-mismatch"}, 2
            if args.source_version != values["source_version"]:
                return {"code": "source-changed"}, 2
            interrupted = any(task.interrupts for task in snapshot.tasks)
            failed = any(task.error for task in snapshot.tasks) and not interrupted
            if args.command == "retry":
                if not failed or snapshot.next != ("review",):
                    return {"code": "not-failed"}, 2
                # No new input and no human decision: continue the failed node.
                graph.invoke(None, config)
                return observation(graph, config, calls, "node-retried"), 0
            if failed:
                return {"code": "retry-required"}, 2
            if values["status"] in ("accepted", "rejected"):
                if values["status"] != args.decision:
                    return {"code": "decision-conflict"}, 2
                return observation(graph, config, calls, "decision-replayed"), 0
            if values["status"] != "awaiting_review" or snapshot.next != ("review",):
                return {"code": "not-awaiting-review"}, 2
            graph.invoke(Command(resume={"decision": args.decision, "digest": args.digest}), config)
        return observation(graph, config, calls), 0


def run_child(db, command, *extra):
    result = subprocess.run([sys.executable, str(Path(__file__).resolve()), command, "--db", str(db), *extra], capture_output=True, text=True, check=False, timeout=15)
    try:
        payload = json.loads(result.stdout)
    except json.JSONDecodeError as exc:
        raise AssertionError("child_returned_no_observation") from exc
    return payload, result.returncode


def demo(db):
    if Path(db).exists():
        return {"code": "demo-needs-new-file"}, 2
    first, code = run_child(db, "start")
    if code:
        return first, code
    second, code = run_child(db, "inspect")
    if code:
        return second, code
    resumed, code = run_child(db, "resume", "--decision", "accepted", "--digest", second["digest"])
    if code:
        return resumed, code
    replayed, code = run_child(db, "resume", "--decision", "accepted", "--digest", second["digest"])
    return {"mode": "offline-scripted-review", "separate_processes": 4, "observations": [first, second, resumed, replayed], "sent": False}, code


def failure_demo(db):
    if Path(db).exists():
        return {"code": "demo-needs-new-file"}, 2
    failed, code = run_child(db, "start", "--fail-before-review")
    if code != 2 or failed.get("code") != "node-failed":
        return {"code": "expected-node-failure-not-observed"}, 2
    stored, code = run_child(db, "inspect")
    if code:
        return stored, code
    retried, code = run_child(db, "retry", "--digest", stored["digest"])
    if code:
        return retried, code
    accepted, code = run_child(db, "resume", "--decision", "accepted", "--digest", retried["digest"])
    if code:
        return accepted, code
    replayed, code = run_child(db, "resume", "--decision", "accepted", "--digest", retried["digest"])
    return {"mode": "actual-graph-scripted-node-failure", "separate_processes": 5,
            "observations": [failed, stored, retried, accepted, replayed], "sent": False}, code


class ParallelState(DraftState):
    policy_status: str


class PolicyDependencyUnavailable(RuntimeError):
    """Scripted branch failure; no external policy service is contacted."""


def build_parallel(checkpointer, calls, fail_policy=False):
    def draft_branch(state):
        calls.append("draft")
        return draft(state)

    def policy_branch(state):
        calls.append("policy")
        if fail_policy:
            raise PolicyDependencyUnavailable("scripted_policy_dependency_unavailable")
        return {"policy_status": "checked"}

    def joined_review(state):
        calls.append("review")
        if state["policy_status"] != "checked":
            raise ValueError("policy_not_checked")
        return review(state)

    graph = StateGraph(ParallelState)
    graph.add_node("draft", draft_branch)
    graph.add_node("policy", policy_branch)
    graph.add_node("review", joined_review)
    graph.add_edge(START, "draft")
    graph.add_edge(START, "policy")
    graph.add_edge(["draft", "policy"], "review")
    graph.add_edge("review", END)
    return graph.compile(checkpointer=checkpointer)


def parallel_observation(graph, config, calls, code="parallel-checkpoint-observed"):
    state = graph.get_state(config)
    return {"code": code, "status": state.values["status"], "digest": state.values["digest"],
            "policy_status": state.values["policy_status"], "next": list(state.next),
            "task_results": {task.name: task.result is not None for task in state.tasks},
            "task_error_nodes": [task.name for task in state.tasks if task.error],
            "interrupted": any(task.interrupts for task in state.tasks), "node_calls": calls}


def execute_parallel(args):
    # Different synthetic thread identity, not tenant authentication or a migration.
    config = config_for(args.tenant, "parallel/" + args.request)
    calls = []
    with SqliteSaver.from_conn_string(args.db) as checkpointer:
        graph = build_parallel(checkpointer, calls, args.command == "parallel-fail")
        state = graph.get_state(config)
        if args.command == "parallel-fail":
            if state.values:
                return {"code": "request-exists"}, 2
            try:
                graph.invoke({"tenant": args.tenant, "request": args.request,
                              "source_version": args.source_version, "draft": "", "digest": "",
                              "status": "new", "policy_status": "unchecked"}, config)
            except PolicyDependencyUnavailable:
                return parallel_observation(graph, config, calls, "parallel-node-failed"), 2
            return {"code": "expected-branch-failure-not-observed"}, 2
        if not state.values:
            return {"code": "missing-thread"}, 2
        if args.command == "parallel-inspect":
            return parallel_observation(graph, config, calls), 0
        if args.deny_review:
            return {"code": "review-denied"}, 2
        observed = parallel_observation(graph, config, calls)
        if not observed["digest"] or args.digest != observed["digest"]:
            return {"code": "proposal-mismatch"}, 2
        if args.source_version != state.values["source_version"]:
            return {"code": "source-changed"}, 2
        if args.command == "parallel-retry":
            if observed["interrupted"] or observed["task_error_nodes"] != ["policy"] or not observed["task_results"].get("draft"):
                return {"code": "not-failed"}, 2
            graph.invoke(None, config)
            return parallel_observation(graph, config, calls, "parallel-node-retried"), 0
        if state.values["status"] in ("accepted", "rejected"):
            if args.decision != state.values["status"]:
                return {"code": "decision-conflict"}, 2
            return parallel_observation(graph, config, calls, "decision-replayed"), 0
        if not observed["interrupted"] or state.next != ("review",):
            return {"code": "not-awaiting-review"}, 2
        graph.invoke(Command(resume={"decision": args.decision, "digest": args.digest}), config)
        return parallel_observation(graph, config, calls), 0


def parallel_demo(db):
    if Path(db).exists():
        return {"code": "demo-needs-new-file"}, 2
    failed, code = run_child(db, "parallel-fail")
    if code != 2 or failed.get("code") != "parallel-node-failed":
        return {"code": "expected-branch-failure-not-observed"}, 2
    stored, code = run_child(db, "parallel-inspect")
    if code:
        return stored, code
    retried, code = run_child(db, "parallel-retry", "--digest", stored["digest"])
    if code:
        return retried, code
    accepted, code = run_child(db, "parallel-resume", "--decision", "accepted", "--digest", retried["digest"])
    if code:
        return accepted, code
    return {"mode": "actual-graph-scripted-parallel-failure", "separate_processes": 4,
            "observations": [failed, stored, retried, accepted], "sent": False}, 0


class CheckpointTests(unittest.TestCase):
    def setUp(self):
        self.directory = tempfile.TemporaryDirectory()
        self.addCleanup(self.directory.cleanup)
        self.db = Path(self.directory.name) / "checkpoints.sqlite"

    def start(self):
        value, code = run_child(self.db, "start")
        self.assertEqual(code, 0)
        self.assertEqual(value["status"], "awaiting_review")
        self.assertTrue(value["interrupted"])
        return value

    def resume(self, digest, *extra):
        return run_child(self.db, "resume", "--decision", "accepted", "--digest", digest, *extra)

    def test_actual_checkpoint_survives_process_exit_and_resumes(self):
        first = self.start()
        after, code = run_child(self.db, "inspect")
        self.assertEqual(code, 0)
        self.assertEqual(after["digest"], first["digest"])
        self.assertEqual(after["next"], ["review"])
        accepted, code = self.resume(first["digest"])
        self.assertEqual(code, 0)
        self.assertEqual(accepted["status"], "accepted")
        self.assertFalse(accepted["interrupted"])
        after, code = run_child(self.db, "inspect")
        self.assertEqual(after["status"], "accepted")
        self.assertEqual(after["next"], [])

    def test_scope_naming_does_not_find_another_tenants_thread(self):
        first = self.start()
        other, code = run_child(self.db, "inspect", "--tenant", "south")
        self.assertEqual((other["code"], code), ("missing-thread", 2))
        unchanged, code = run_child(self.db, "inspect")
        self.assertEqual(unchanged["digest"], first["digest"])

    def test_refused_resume_preserves_the_pending_checkpoint(self):
        first = self.start()
        cases = [("0" * 64, (), "proposal-mismatch"), (first["digest"], ("--source-version", "4"), "source-changed"), (first["digest"], ("--deny-review",), "review-denied")]
        for digest, extra, expected in cases:
            refused, code = self.resume(digest, *extra)
            self.assertEqual((refused["code"], code), (expected, 2))
            waiting, code = run_child(self.db, "inspect")
            self.assertEqual(waiting["status"], "awaiting_review")
            self.assertTrue(waiting["interrupted"])
        accepted, code = self.resume(first["digest"])
        self.assertEqual(accepted["status"], "accepted")

    def test_terminal_replay_and_conflict_do_not_reopen_review(self):
        first = self.start()
        self.resume(first["digest"])
        replay, code = self.resume(first["digest"])
        self.assertEqual((replay["code"], code), ("decision-replayed", 0))
        conflict, code = run_child(self.db, "resume", "--decision", "rejected", "--digest", first["digest"])
        self.assertEqual((conflict["code"], code), ("decision-conflict", 2))
        duplicate, code = run_child(self.db, "start", "--source-version", "4")
        self.assertEqual((duplicate["code"], code), ("request-exists", 2))
        final, code = run_child(self.db, "inspect")
        self.assertEqual(final["status"], "accepted")

    def test_rejected_decision_is_a_distinct_terminal_outcome(self):
        first = self.start()
        rejected, code = run_child(self.db, "resume", "--decision", "rejected", "--digest", first["digest"])
        self.assertEqual(code, 0)
        self.assertEqual(rejected["status"], "rejected")
        self.assertEqual(rejected["next"], [])

    def test_another_request_cannot_reuse_the_first_proposal_binding(self):
        first = self.start()
        second, code = run_child(self.db, "start", "--request", "R2")
        self.assertEqual(code, 0)
        self.assertNotEqual(second["digest"], first["digest"])
        refused, code = run_child(self.db, "resume", "--request", "R2", "--decision", "accepted", "--digest", first["digest"])
        self.assertEqual((refused["code"], code), ("proposal-mismatch", 2))

    def test_failed_node_retries_in_a_fresh_process_without_redrafting(self):
        failed, code = run_child(self.db, "start", "--fail-before-review")
        self.assertEqual((failed["code"], code), ("node-failed", 2))
        self.assertEqual(failed["node_calls"], ["draft", "review"])
        self.assertEqual(failed["task_error_nodes"], ["review"])
        self.assertFalse(failed["interrupted"])
        self.assertTrue(failed["pending_failure"])
        stored, code = run_child(self.db, "inspect")
        self.assertEqual(stored["node_calls"], [])
        self.assertEqual(stored["task_error_nodes"], ["review"])
        self.assertEqual(stored["digest"], failed["digest"])
        for extra, expected in ((["--source-version", "4"], "source-changed"), (["--deny-review"], "review-denied")):
            denied, code = run_child(self.db, "retry", "--digest", stored["digest"], *extra)
            self.assertEqual((denied["code"], code), (expected, 2))
        wrong, code = run_child(self.db, "retry", "--digest", "0" * 64)
        self.assertEqual((wrong["code"], code), ("proposal-mismatch", 2))
        not_review, code = self.resume(stored["digest"])
        self.assertEqual((not_review["code"], code), ("retry-required", 2))
        retried, code = run_child(self.db, "retry", "--digest", stored["digest"])
        self.assertEqual((retried["code"], code), ("node-retried", 0))
        self.assertEqual(retried["node_calls"], ["review"])
        # The pinned SDK retains the earlier task error alongside the new interrupt.
        self.assertEqual(retried["task_error_nodes"], ["review"])
        self.assertFalse(retried["pending_failure"])
        self.assertTrue(retried["interrupted"])
        self.assertEqual(retried["digest"], stored["digest"])
        accepted, code = self.resume(stored["digest"])
        self.assertEqual((accepted["status"], code), ("accepted", 0))
        self.assertEqual(accepted["node_calls"], ["review"])
        self.assertEqual(accepted["next"], [])

    def test_retry_cannot_release_a_human_interrupt_or_terminal_result(self):
        pending = self.start()
        denied, code = run_child(self.db, "retry", "--digest", pending["digest"])
        self.assertEqual((denied["code"], code), ("not-failed", 2))
        stored, code = run_child(self.db, "inspect")
        self.assertTrue(stored["interrupted"])
        self.resume(pending["digest"])
        denied, code = run_child(self.db, "retry", "--digest", pending["digest"])
        self.assertEqual((denied["code"], code), ("not-failed", 2))
        stored, code = run_child(self.db, "inspect")
        self.assertEqual(stored["status"], "accepted")

    def test_failure_demo_records_five_processes_and_preserves_existing_files(self):
        result, code = failure_demo(self.db)
        self.assertEqual(code, 0)
        self.assertEqual(result["separate_processes"], 5)
        self.assertEqual([r["node_calls"] for r in result["observations"]], [["draft", "review"], [], ["review"], ["review"], []])
        original = self.db.read_bytes()
        refused, code = failure_demo(self.db)
        self.assertEqual((refused["code"], code), ("demo-needs-new-file", 2))
        self.assertEqual(self.db.read_bytes(), original)

    def test_parallel_pending_write_survives_fresh_process_and_is_not_recomputed(self):
        failed, code = run_child(self.db, "parallel-fail")
        self.assertEqual((failed["code"], code), ("parallel-node-failed", 2))
        self.assertCountEqual(failed["node_calls"], ["draft", "policy"])
        self.assertTrue(failed["task_results"]["draft"])
        self.assertFalse(failed["task_results"]["policy"])
        self.assertEqual(failed["task_error_nodes"], ["policy"])
        self.assertFalse(failed["interrupted"])
        self.assertEqual(failed["policy_status"], "unchecked")
        stored, code = run_child(self.db, "parallel-inspect")
        self.assertEqual(code, 0)
        self.assertEqual(stored["node_calls"], [])
        self.assertTrue(stored["task_results"]["draft"])
        self.assertEqual(stored["digest"], failed["digest"])
        retried, code = run_child(self.db, "parallel-retry", "--digest", stored["digest"])
        self.assertEqual((retried["code"], code), ("parallel-node-retried", 0))
        self.assertEqual(retried["node_calls"], ["policy", "review"])
        self.assertEqual(retried["digest"], failed["digest"])
        self.assertEqual(retried["policy_status"], "checked")
        self.assertEqual(retried["next"], ["review"])
        self.assertTrue(retried["interrupted"])
        accepted, code = run_child(self.db, "parallel-resume", "--decision", "accepted", "--digest", retried["digest"])
        self.assertEqual((accepted["status"], code), ("accepted", 0))
        self.assertEqual(accepted["node_calls"], ["review"])
        self.assertEqual(accepted["next"], [])

    def test_parallel_refusals_preserve_pending_work_and_do_not_release_review(self):
        failed, code = run_child(self.db, "parallel-fail")
        digest = failed["digest"]
        for extra, expected in ((["--digest", "0" * 64], "proposal-mismatch"),
                                (["--digest", digest, "--deny-review"], "review-denied"),
                                (["--digest", digest, "--source-version", "4"], "source-changed")):
            denied, code = run_child(self.db, "parallel-retry", *extra)
            self.assertEqual((denied["code"], code), (expected, 2))
        denied, code = run_child(self.db, "parallel-resume", "--decision", "accepted", "--digest", digest)
        self.assertEqual((denied["code"], code), ("not-awaiting-review", 2))
        still, code = run_child(self.db, "parallel-inspect")
        self.assertTrue(still["task_results"]["draft"])
        self.assertEqual(still["task_error_nodes"], ["policy"])
        run_child(self.db, "parallel-retry", "--digest", digest)
        denied, code = run_child(self.db, "parallel-retry", "--digest", digest)
        self.assertEqual((denied["code"], code), ("not-failed", 2))
        run_child(self.db, "parallel-resume", "--decision", "rejected", "--digest", digest)
        replayed, code = run_child(self.db, "parallel-resume", "--decision", "rejected", "--digest", digest)
        self.assertEqual((replayed["code"], code), ("decision-replayed", 0))
        self.assertEqual(replayed["node_calls"], [])
        conflict, code = run_child(self.db, "parallel-resume", "--decision", "accepted", "--digest", digest)
        self.assertEqual((conflict["code"], code), ("decision-conflict", 2))

    def test_parallel_thread_naming_cannot_select_sequential_or_other_tenant_work(self):
        self.start()
        missing, code = run_child(self.db, "parallel-inspect")
        self.assertEqual((missing["code"], code), ("missing-thread", 2))
        run_child(self.db, "parallel-fail")
        missing, code = run_child(self.db, "parallel-inspect", "--tenant", "south")
        self.assertEqual((missing["code"], code), ("missing-thread", 2))
        normal, code = run_child(self.db, "inspect")
        self.assertTrue(normal["interrupted"])
        self.assertEqual(normal["task_error_nodes"], [])

    def test_parallel_demo_observes_four_processes_without_overwriting_a_file(self):
        report, code = parallel_demo(self.db)
        self.assertEqual(code, 0)
        self.assertEqual(report["separate_processes"], 4)
        self.assertEqual([o["node_calls"] for o in report["observations"]][1:], [[], ["policy", "review"], ["review"]])
        original = self.db.read_bytes()
        refused, code = parallel_demo(self.db)
        self.assertEqual((refused["code"], code), ("demo-needs-new-file", 2))
        self.assertEqual(self.db.read_bytes(), original)

    def test_demo_does_not_overwrite_an_existing_file(self):
        self.db.write_text("keep this unrelated file", encoding="utf8")
        result, code = demo(self.db)
        self.assertEqual((result["code"], code), ("demo-needs-new-file", 2))
        self.assertEqual(self.db.read_text(), "keep this unrelated file")


def main():
    if sys.argv[1:] == ["test"]:
        unittest.main(argv=[sys.argv[0]], verbosity=2)
        return
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("command", choices=["start", "inspect", "resume", "retry", "demo", "failure-demo", "parallel-demo", "parallel-fail", "parallel-inspect", "parallel-retry", "parallel-resume"])
    parser.add_argument("--db", required=True, help="fresh disposable local SQLite file")
    parser.add_argument("--tenant", choices=["north", "south"], default="north")
    parser.add_argument("--request", choices=["R1", "R2"], default="R1")
    parser.add_argument("--source-version", type=int, choices=[3, 4], default=3)
    parser.add_argument("--decision", choices=["accepted", "rejected"])
    parser.add_argument("--digest")
    parser.add_argument("--deny-review", action="store_true", help="simulate a host revoking reviewer permission")
    parser.add_argument("--fail-before-review", action="store_true", help="script a node exception before interrupt; start only")
    args = parser.parse_args()
    if args.command not in ("resume", "retry", "parallel-resume", "parallel-retry") and (args.decision or args.digest or args.deny_review):
        parser.error("decision/digest/deny-review apply only to resume or retry")
    if args.command in ("retry", "parallel-retry") and args.decision:
        parser.error("retry takes no human decision")
    if args.fail_before_review and args.command != "start":
        parser.error("fail-before-review applies only to start")
    if args.command in ("demo", "failure-demo", "parallel-demo") and (args.tenant != "north" or args.request != "R1" or args.source_version != 3):
        parser.error("demo uses fixed north/R1 inputs; use start/inspect/resume to change scope")
    if args.command in ("resume", "parallel-resume") and (not args.decision or not args.digest):
        parser.error("resume requires --decision and the digest observed during inspect")
    if args.command in ("retry", "parallel-retry") and not args.digest:
        parser.error("retry requires the digest observed during inspect")
    try:
        with patch.object(socket.socket, "connect", side_effect=RuntimeError("network_disabled")):
            if args.command == "parallel-demo":
                result, code = parallel_demo(args.db)
            elif args.command.startswith("parallel-"):
                result, code = execute_parallel(args)
            elif args.command == "failure-demo":
                result, code = failure_demo(args.db)
            else:
                result, code = demo(args.db) if args.command == "demo" else execute(args)
        print(json.dumps(result))
        return code
    except Exception as exc:
        print(json.dumps({"code": "local-lab-error", "error_type": type(exc).__name__}))
        return 2


if __name__ == "__main__":
    sys.exit(main() or 0)