"""Local crash/replay lab. No model API, payment API, or production service.""" import asyncio import json import os from pathlib import Path import sqlite3 import subprocess import sys from datetime import timedelta from typing import TypedDict from langgraph.checkpoint.sqlite import SqliteSaver from langgraph.func import entrypoint, task from langgraph.graph import END, START, StateGraph from langgraph.types import Command, interrupt from temporalio import activity, workflow from temporalio.client import Client from temporalio.common import RetryPolicy from temporalio.testing import WorkflowEnvironment from temporalio.worker import Replayer, Worker ROOT = Path(os.environ.get("T24_LAB_DIR", "/tmp/t24-lab")) ROOT.mkdir(parents=True, exist_ok=True) def db(case): c = sqlite3.connect(ROOT / f"{case}-sink.db") c.execute("CREATE TABLE IF NOT EXISTS calls(kind TEXT, payload TEXT)") c.execute("CREATE TABLE IF NOT EXISTS effects(id INTEGER PRIMARY KEY, key TEXT, payload TEXT)") c.execute("CREATE TABLE IF NOT EXISTS receipts(key TEXT PRIMARY KEY, result TEXT, payload TEXT)") c.commit() return c def crash(case, point): if os.environ.get("T24_FAIL") == point: marker = ROOT / f"{case}-{point}.once" if not marker.exists(): marker.write_text("injected") os._exit(91) def predict(arg): case = arg["case"] crash(case, "before_llm") with db(case) as c: c.execute("INSERT INTO calls VALUES ('llm','')") n = c.execute("SELECT count(*) FROM calls WHERE kind='llm'").fetchone()[0] crash(case, "llm_return") return f"draft-{n}" def effect(arg): case = arg["case"] crash(case, "before_write") payload, key = arg["draft"], "tenant:order-42:publish:v1" with db(case) as c: # One receiver-side transaction: dedupe, mutation, and saved response. c.execute("BEGIN IMMEDIATE") c.execute("INSERT INTO calls VALUES ('write',?)", (payload,)) old = c.execute("SELECT result,payload FROM receipts WHERE key=?", (key,)).fetchone() if arg["dedup"] and old: if old[1] != payload: raise ValueError("same key with different payload") result = old[0] else: c.execute("INSERT INTO effects(key,payload) VALUES (?,?)", (key, payload)) result = f"receipt-{c.execute('SELECT count(*) FROM effects').fetchone()[0]}" if arg["dedup"]: c.execute("INSERT INTO receipts VALUES (?,?,?)", (key, result, payload)) crash(case, "effect_commit") return result class State(TypedDict, total=False): case: str dedup: bool draft: str receipt: str def lg(case, mode, resume=False, historical=False): config = {"configurable": {"thread_id": case}} arg = {"case": case, "dedup": "unsafe" not in case} with SqliteSaver.from_conn_string(str(ROOT / f"{case}-graph.db")) as saver: if mode == "interrupt": g = StateGraph(State) def approval(s): receipt = effect({**s, "draft": "fixed"}) interrupt("Approve?") return {"receipt": receipt} g.add_node("approval", approval) g.add_edge(START, "approval") g.add_edge("approval", END) graph = g.compile(checkpointer=saver) elif mode == "functional": ptask, etask = task(predict), task(effect) @entrypoint(checkpointer=saver) def graph(data): draft = ptask(data).result() return etask({**data, "draft": draft}).result() else: g = StateGraph(State) if mode == "combined": def combined(s): draft = predict(s) return {"draft": draft, "receipt": effect({**s, "draft": draft})} g.add_node("combined", combined) g.add_edge(START, "combined") g.add_edge("combined", END) else: g.add_node("llm", lambda s: {"draft": predict(s)}) g.add_node("write", lambda s: {"receipt": effect(s)}) g.add_edge(START, "llm") g.add_edge("llm", "write") g.add_edge("write", END) graph = g.compile(checkpointer=saver) if historical: config = next(s.config for s in graph.get_state_history(config) if s.next == ("write",)) data = Command(resume=True) if mode == "interrupt" and resume else None if resume or historical else arg value = graph.invoke(data, config, durability="sync") (ROOT / f"{case}-result.json").write_text(json.dumps(value, default=str)) @activity.defn async def model_activity(arg: dict) -> str: return predict(arg) @activity.defn async def write_activity(arg: dict) -> str: return effect(arg) @workflow.defn(sandboxed=False) class Publish: @workflow.run async def run(self, arg: dict) -> str: options = dict(start_to_close_timeout=timedelta(seconds=3), retry_policy=RetryPolicy(initial_interval=timedelta(seconds=1), maximum_attempts=3)) draft = await workflow.execute_activity(model_activity, arg, **options) return await workflow.execute_activity(write_activity, {**arg, "draft": draft}, **options) async def worker(address, queue): client = await Client.connect(address) async with Worker(client, task_queue=queue, workflows=[Publish], activities=[model_activity, write_activity]): await asyncio.Event().wait() def counts(case): with db(case) as c: return {"llm": c.execute("SELECT count(*) FROM calls WHERE kind='llm'").fetchone()[0], "write": c.execute("SELECT count(*) FROM calls WHERE kind='write'").fetchone()[0], "effects": c.execute("SELECT count(*) FROM effects").fetchone()[0]} def spawn(args, fail=""): return subprocess.Popen([sys.executable, str(Path(__file__).resolve()), *args], env={**os.environ, "T24_FAIL": fail}) async def suite(): rows = [] cases = [("split-before", "split", "before_llm", (1, 1, 1)), ("split-llm-return", "split", "llm_return", (2, 1, 1)), ("split-before-write", "split", "before_write", (1, 1, 1)), ("split-effect-safe", "split", "effect_commit", (1, 2, 1)), ("split-effect-unsafe", "split", "effect_commit", (1, 2, 2)), ("combined-effect-unsafe", "combined", "effect_commit", (2, 2, 2)), ("functional-before-write", "functional", "before_write", (1, 1, 1)), ("functional-effect-safe", "functional", "effect_commit", (1, 2, 1))] for case, mode, fail, expected in cases: p = spawn(["lg", case, mode], fail) rc = await asyncio.to_thread(p.wait) assert rc == 91, (case, rc) p = spawn(["lg", case, mode, "resume"]) assert await asyncio.to_thread(p.wait) == 0 actual = counts(case) assert tuple(actual.values()) == expected, (case, actual, expected) rows.append({"engine": "langgraph", "case": case, **actual}) # Historical checkpoint intentionally repeats the write node. for case in ["historical-safe", "historical-unsafe"]: assert await asyncio.to_thread(spawn(["lg", case, "split"]).wait) == 0 assert await asyncio.to_thread(spawn(["lg", case, "split", "history"]).wait) == 0 actual = counts(case) assert tuple(actual.values()) == (1, 2, 2 if "unsafe" in case else 1) rows.append({"engine": "langgraph", "case": case, **actual}) for case in ["interrupt-safe", "interrupt-unsafe"]: assert await asyncio.to_thread(spawn(["lg", case, "interrupt"]).wait) == 0 assert await asyncio.to_thread(spawn(["lg", case, "interrupt", "resume"]).wait) == 0 actual = counts(case) assert tuple(actual.values()) == (0, 2, 2 if "unsafe" in case else 1) rows.append({"engine": "langgraph", "case": case, **actual}) case = "completed-resume" assert await asyncio.to_thread(spawn(["lg", case, "split"]).wait) == 0 assert await asyncio.to_thread(spawn(["lg", case, "split", "resume"]).wait) == 0 actual = counts(case) assert tuple(actual.values()) == (1, 1, 1) rows.append({"engine": "langgraph", "case": case, **actual}) async with await WorkflowEnvironment.start_local(dev_server_log_level="error") as env: address = env.client.service_client.config.target_host for case, fail, expected in [ ("temporal-llm-return", "llm_return", (2, 1, 1)), ("temporal-before-write", "before_write", (1, 1, 1)), ("temporal-effect-safe", "effect_commit", (1, 2, 1)), ("temporal-effect-unsafe", "effect_commit", (1, 2, 2))]: p = spawn(["worker", address, case], fail) handle = await env.client.start_workflow(Publish.run, {"case": case, "dedup": "unsafe" not in case}, id=case, task_queue=case) assert await asyncio.wait_for(asyncio.to_thread(p.wait), 30) == 91 p = spawn(["worker", address, case]) try: result = await asyncio.wait_for(handle.result(), 30) history = await handle.fetch_history() (ROOT / f"{case}-history.json").write_text(history.to_json()) before = counts(case) await Replayer(workflows=[Publish]).replay_workflow(history) after = counts(case) assert before == after, ("Replay called Activity", case) assert tuple(after.values()) == expected, (case, after, expected) rows.append({"engine": "temporal", "case": case, **after, "offline_replay": "no_new_calls", "result": result}) finally: p.terminate() await asyncio.to_thread(p.wait) print(json.dumps(rows, indent=2)) (ROOT / "results.json").write_text(json.dumps(rows, indent=2)) if __name__ == "__main__": if len(sys.argv) == 1: asyncio.run(suite()) elif sys.argv[1] == "lg": lg(sys.argv[2], sys.argv[3], len(sys.argv) > 4 and sys.argv[4] == "resume", len(sys.argv) > 4 and sys.argv[4] == "history") elif sys.argv[1] == "worker": asyncio.run(worker(sys.argv[2], sys.argv[3]))