From 19b1c1f1a21ad00746c53d9219dee51da22de071 Mon Sep 17 00:00:00 2001 From: Minglong Pan <56749246+minglong51@users.noreply.github.com> Date: Sat, 22 Aug 2026 12:06:28 -0700 Subject: [PATCH 1/2] docs: score durability against the published seven criteria A conformance table (met/partial/out-of-scope-by-design, each verdict citing the module or doc section behind it) against the Diagrid critique's definition of durable execution. Formalizes the honest positioning already spread across README, production.md, and dsl-comparison.md; linked from the README support boundary. --- docs/durability-conformance.md | 23 +++++++++++++++++++++++ 1 file changed, 23 insertions(+) create mode 100644 docs/durability-conformance.md diff --git a/docs/durability-conformance.md b/docs/durability-conformance.md new file mode 100644 index 0000000..4c4b95d --- /dev/null +++ b/docs/durability-conformance.md @@ -0,0 +1,23 @@ +# ThreadLang durability conformance + +A widely-read 2026 critique — [the Diagrid durability critique](https://www.diagrid.io/blog/checkpoints-are-not-durable-execution-why-langgraph-crewai-google-adk-and-others-fall-short-for-production-agent-workflows) — argues that checkpointing is not durable execution, scores LangGraph, CrewAI, and Google ADK as falling short, and defines the term with seven published criteria. ThreadLang's own docs already decline the full label: the README bounds support to one POSIX process, one local SQLite store, and journaled LLM calls that re-execute at most the single in-flight call of the interrupted step across a hard crash, and [`benchmarks/dsl-comparison.md`](benchmarks/dsl-comparison.md) concludes v0.13 is "step-checkpoint durability, not durable replay". This table scores ThreadLang v0.13.3 against the same seven criteria, so the claim and its limits sit in one place. **Met**, **partial**, and **out of scope by design** describe what the code does today; every verdict cites the module or doc section behind it. + +| # | Criterion | Verdict | What the code does | +|---|---|---|---| +| 1 | Automatic persistence at every step boundary | **met** | `run_durable` wires persistence into execution: a write-through trace lands every `TraceEvent` in the `events` table as it is appended, and the `on_step_complete` hook checkpoints each finished step to `step_outputs`, on an autocommit connection (`src/threadlang/store.py`). No save calls, decorators, or persistence configuration in the program. A checkpoint is written only after output validation and edge resolution succeed, so a bad route label cannot poison it (`src/threadlang/runtime.py`). | +| 2 | Deterministic replay from an event log | **partial** | The event log is real: persisted, ordered by `seq`, append-only across attempts (`src/threadlang/store.py`; `docs/design/phase-2-durability.md`, decision 3). Replaying a *completed* run returns the stored output with zero model calls (`store.py`, `run_durable`). But resume is checkpoint-resume, not event-history replay: completed steps are reused, and everything after the last checkpoint re-executes — its completed model calls replay from the per-run journal (`src/threadlang/journal.py`), so at most the interrupted step's single in-flight call is made again. This is the "step-checkpoint durability, not durable replay" conclusion of [`benchmarks/dsl-comparison.md`](benchmarks/dsl-comparison.md). | +| 3 | Implicit activity boundaries | **met** | The step graph the author writes is the activity decomposition — there is no second layer of activity declarations, decorators, or per-step opt-in; every declared step is a persistence and resume boundary (`src/threadlang/runtime.py`, `_run_steps`). The boundary is the whole step: model/tool turns inside an `agent` loop are not individually checkpointed, so a mid-loop crash re-runs the entire step (`docs/design/phase-2-durability.md`, decision 2), and `emit llm` is not a step checkpoint (`docs/production.md`, durability contract). | +| 4 | Exactly-once side effects | **out of scope by design** | Model calls, including `emit llm`, are journaled per run — across a hard crash at most the interrupted step's single in-flight call repeats — and tool calls in the interrupted step remain at-least-once; exactly-once external effects are out of scope (`docs/production.md`, durability contract). The durable path enforces the mitigation: tools declared `side_effects=True, idempotent=False` are rejected (`src/threadlang/tools.py`, `ToolRegistry.validate_durable`, called from `run_durable`). The store prevents duplicate *execution* of a run — atomic pending claim, CAS resume (`src/threadlang/store.py`) — but suppressing duplicate *effects* is the tool author's declared idempotency contract, not a runtime guarantee. | +| 5 | Built-in recovery via durable reminders | **partial** | Recovery exists on the queue path: startup acquires the exclusive worker lock and requeues crash-stranded `running` runs to `pending`, and re-dispatching the same id is safe because resume is idempotent (`src/threadlang/control.py`, `WorkerPool`; `src/threadlang/store.py`, `requeue_orphans`; `docs/design/phase-2-durability.md`, decision 4). But nothing fires on its own: there are no durable reminders or timers, a crashed CLI run waits for an operator's `--resume`, and a dead server process recovers only when the server is restarted. | +| 6 | Linear code with no manual skip logic | **met** | Programs contain no recovery code: the language has no conditionals, and resume is runtime-managed — checkpointed steps are skipped with the skip itself traced, and a resumed `route` step re-derives its edge from the stored label with no model call (`src/threadlang/runtime.py`, `_run_steps`). The author writes a forward-only graph; the only identifier anyone manages is the run id printed by a failed run. | +| 7 | Distributed rebalancing | **out of scope by design** | The supported boundary is one POSIX process and one local SQLite store (README; `docs/production.md`, supported boundary). Distributed workers, leases across hosts, and network filesystems are explicit non-goals (`docs/production.md`, explicit non-goals). The worker pool is threads in one process; a second process fails startup against `.worker.lock` rather than rebalancing work. | + +## What we deliberately do not claim + +- **Durable replay.** No Temporal/Dapr-style event-history reconstruction, patch markers, or in-flight version migration (`docs/production.md`, explicit non-goals). Resume reuses step checkpoints and re-executes the remainder; journaled model calls replay, and at most the single in-flight call is made again. +- **Exactly-once external effects.** Tool calls are at-least-once across a hard crash, a model call repeats at most the single in-flight invocation, and a crash inside a step re-runs that whole step. The runtime enforces declaration (`ToolSpec.side_effects` / `idempotent`) and rejects what it cannot safely replay; it does not deduplicate effects. +- **Distributed execution.** One process, one store, no cross-host leases or rebalancing. The failure model is process death and restart on one node, not node loss in a fleet. +- **Self-firing recovery.** No durable reminder or timer fires while the server is down. Recovery is startup orphan-requeue on the queue path and operator `--resume` on the CLI path; detection of a dead process belongs to whatever supervises the process. +- **Sub-step granularity.** An `agent` step's tool-use loop is one checkpoint unit — per-turn *response journaling* ships (a journaled turn replays rather than re-calling the provider, `src/threadlang/journal.py`), but the checkpoint/resume unit is still the whole step (`docs/design/phase-2-durability.md`, decision 2). + +*Scored 2026-08-22 against ThreadLang v0.13.3. Where this doc disagrees with the code, the code wins and this doc is stale.* From 6c57001258bdc9d342a9153a83c6757fd12ebf57 Mon Sep 17 00:00:00 2001 From: Minglong Pan <56749246+minglong51@users.noreply.github.com> Date: Sat, 22 Aug 2026 12:06:28 -0700 Subject: [PATCH 2/2] feat: journal LLM responses per run, narrowing the at-least-once window run_durable now wraps the LLM client in a per-run journaling layer (journal_llm=True by default): each call's request fingerprint and response land in a new llm_journal table, and a resumed run replays fingerprint-matched responses instead of re-calling the provider. Across a hard crash, at most the single in-flight call of the interrupted step re-executes. Exactly-once remains out of scope. This is the per-agent-turn checkpointing phase-2-durability.md deferred. 7 new tests, 250 total pass; ruff/mypy clean; LLD+HLD updated in lockstep. --- README.md | 19 ++- docs/design/HLD.md | 8 +- docs/design/LLD.md | 46 ++++++- docs/production.md | 8 +- src/threadlang/journal.py | 167 ++++++++++++++++++++++++ src/threadlang/store.py | 92 +++++++++++-- tests/test_llm_journal.py | 263 ++++++++++++++++++++++++++++++++++++++ 7 files changed, 574 insertions(+), 29 deletions(-) create mode 100644 src/threadlang/journal.py create mode 100644 tests/test_llm_journal.py diff --git a/README.md b/README.md index ad35599..e566e0b 100644 --- a/README.md +++ b/README.md @@ -20,8 +20,11 @@ The current source version is **v0.13.3 (alpha)**. It includes: established AST runtime. The support boundary is deliberately narrow: one POSIX process, one local -SQLite store, and at-least-once LLM calls across a hard crash. Design records -and deliberate cuts live in [`docs/design/`](docs/design/). +SQLite store, and journaled LLM calls that re-execute at most the single +in-flight call of the interrupted step across a hard crash. Design records +and deliberate cuts live in [`docs/design/`](docs/design/); the boundary is +scored against the published seven durable-execution criteria in +[`docs/durability-conformance.md`](docs/durability-conformance.md). ```thread thread TwoStep { @@ -196,9 +199,10 @@ finally: The runtime stays storage-agnostic — `run_durable` hands it a write-through trace and a checkpoint callback, so the same executor runs durable or -ephemeral. Checkpoints are step-level; a crash mid-step re-runs that step, -finished steps are reused. Replaying a *completed* run returns the stored -result and makes no model calls. Details: +ephemeral. Checkpoints are step-level; a crash mid-step re-runs that step (its +completed model calls replay from the run's journal — only the in-flight call +re-executes), finished steps are reused. Replaying a *completed* run returns +the stored result and makes no model calls. Details: [`docs/design/phase-2-durability.md`](docs/design/phase-2-durability.md). ## Control plane (v0.5) @@ -552,8 +556,9 @@ platform. Each shipped layer keeps the determinism/trace bet: parser hardening, authenticated single-node control plane, source/input integrity binding, exclusive worker ownership, CAS resume, packaging, security gates, and a non-root container. The support boundary is one POSIX - process and one local SQLite store; LLM calls remain at-least-once across a - hard crash. See [`docs/production.md`](docs/production.md), + process and one local SQLite store; journaled LLM calls re-execute at most + the single in-flight call across a hard crash. See + [`docs/production.md`](docs/production.md), [`SECURITY.md`](SECURITY.md), and the [semantic comparison](docs/benchmarks/dsl-comparison.md). 11. **Canonical Workflow IR** *(v0.13, shipped)* — deterministic JSON and diff --git a/docs/design/HLD.md b/docs/design/HLD.md index d6071ce..9037039 100644 --- a/docs/design/HLD.md +++ b/docs/design/HLD.md @@ -4,8 +4,9 @@ would make the design document brittle.* The supported production boundary is one POSIX process and one local SQLite -store. "Durable" below means step-boundary checkpoints and crash recovery, not -deterministic event-history replay. See [`../production.md`](../production.md). +store. "Durable" below means step-boundary checkpoints, per-call model-call +journaling, and crash recovery, not deterministic event-history replay. See +[`../production.md`](../production.md). ## Purpose @@ -84,7 +85,8 @@ JS build (server-rendered HTML with inline CSS, `dashboard.py`). | `src/threadlang/llm.py` | Client backends behind a baseline protocol plus optional capabilities: `LLMClient.complete`, `AgentLLMClient.agent_step`, and `RouteLLMClient.route`. `DryRunClient` (deterministic echo + two-phase agent stub), `OpenAICompatClient` (stdlib HTTP), `AnthropicClient` (SDK). | | `src/threadlang/tools.py` | The agent execution boundary: `ToolSpec`/`Tool`/`FunctionTool`, `ToolRegistry` allow-list, deterministic built-ins `echo` + `calculator` (AST-walked arithmetic, no `eval`, no `**`, `tools.py`). | | `src/threadlang/trace.py` | `TraceEvent(phase, message, data)`, `Trace` alias, `DenialCode` enum. The durable record's unit. | -| `src/threadlang/store.py` | Durability (L3): `RunStore` (sqlite tables `runs`/`events`/`step_outputs`, WAL/autocommit), canonical definition/input binding with legacy source fencing, bounded queue/retention, CAS resume, write-through traces, step checkpoints, replay, and metrics queries. | +| `src/threadlang/store.py` | Durability (L3): `RunStore` (sqlite tables `runs`/`events`/`step_outputs`/`llm_journal`, WAL/autocommit), canonical definition/input binding with legacy source fencing, bounded queue/retention, CAS resume, write-through traces, step checkpoints, replay, and metrics queries. | +| `src/threadlang/journal.py` | Per-call LLM response journal for durable runs: `run_durable` wraps the run's client in `JournaledLLMClient`, which records request fingerprints + responses in `llm_journal` and replays fingerprint-matched responses on resume — a crash re-executes at most the interrupted step's single in-flight model call. | | `src/threadlang/control.py` | Control plane workers (L4): exclusive per-store process lock, orphan requeue, atomic claim, source-or-IR execution, per-thread stores, exception-contained worker loops, and readiness state. | | `src/threadlang/server.py` | Authenticated stdlib JSON API + dashboard host: source-or-IR `POST /runs`, paginated run queries, metrics, liveness/readiness, Host/origin/body/input admission checks, and HTML views. `serve()` starts the exclusive worker pool and server together. | | `src/threadlang/dashboard.py` | Observability (L5): pure `(record, events, metrics) -> HTML` renderers for the run list (with aggregate panel) and per-run trace timeline; everything `html.escape`d; meta-refresh while a run is in flight. | diff --git a/docs/design/LLD.md b/docs/design/LLD.md index 92bd7c1..04b5355 100644 --- a/docs/design/LLD.md +++ b/docs/design/LLD.md @@ -4,9 +4,10 @@ Refreshed for v0.13.3. The supported boundary is one POSIX process and one local SQLite store; see [`../production.md`](../production.md). Historical line references elsewhere in this document are explanatory and not API contracts. -> **Refreshed 2026-08-19.** Covers the shipped canonical-IR execution and -> durable-binding path plus the v0.12 admission, recovery, and ownership -> hardening. Where this disagrees with the code, the code wins. +> **Refreshed 2026-08-22.** Covers the shipped canonical-IR execution and +> durable-binding path, the v0.12 admission, recovery, and ownership +> hardening, and per-call LLM response journaling on the durable path +> (`journal.py`). Where this disagrees with the code, the code wins. ## Module Breakdown @@ -194,15 +195,41 @@ the shared-client `WorkerPool` relies on (`control.py`). under `BEGIN IMMEDIATE`. `claim_next_pending()` atomically claims the oldest row. `requeue_orphans()` moves restart-stranded sourced/IR runs back to `pending`. -- Events are sequenced and timestamped; step outputs are upserted checkpoints. - Per-run and aggregate metrics are folds over those persisted events. +- Events are sequenced and timestamped; step outputs are upserted checkpoints; + journaled model calls are appended with a per-run `call_seq` and looked up + by `(run_id, request_fingerprint, occurrence)`. Per-run and aggregate + metrics are folds over those persisted events. - `run_durable(...)` compiles the current program to canonical IR and binds its digest with canonical inputs. The source digest remains metadata and the identity fence for legacy rows lacking canonical definition identity. Resume verifies stored IR integrity, definition/input identity, IR version, and eligible status before it loads checkpoints. A completed run replays without model calls; a fresh run moves `created→running`; any execution exception - marks `failed`; success marks `completed`. + marks `failed`; success marks `completed`. Unless `journal_llm=False`, the + run's LLM client is wrapped in `JournaledLLMClient` (`journal.py`) before + `run_program` sees it, so a resumed run replays the interrupted step's + completed model calls from `llm_journal` and re-executes at most the single + in-flight call. + +### `journal.py` — per-call LLM response journal + +- `JournaledLLMClient(client, store, run_id)` (`journal.py`) — the per-run + wrapper `run_durable` installs. It exposes `complete` unconditionally and + `route`/`agent_step` only when the wrapped client has them (class-level + annotations, conditionally assigned in `__init__`), so the runtime's + `getattr` capability probes see exactly the wrapped client's surface. +- Every call is keyed by `(run_id, request_fingerprint, occurrence)`: the + fingerprint is SHA-256 over the canonical JSON of the full request (`kind` + + `model` + prompt / options / messages+tools; `ToolCall`-carrying messages + and `ToolSpec`s serialize via `dataclasses.asdict`), and `occurrence` is the + per-attempt ordinal of that fingerprint, so two identical requests in one + run keep distinct rows (`journal.py`). +- A journal hit replays the recorded response with no provider call — + `agent_step` payloads reconstruct an `AgentTurn` (`_agent_turn_from_json`, + `journal.py`); a miss calls through and persists request + response JSON. A + fresh run_id starts with an empty journal, so first attempts always call + live; only resumed runs replay. Tool calls are not journaled and + re-execute; exactly-once remains out of scope (`docs/production.md`). ### `control.py` — worker pool @@ -391,6 +418,11 @@ events (run_id TEXT, seq INTEGER, phase TEXT, message TEXT, PRIMARY KEY (run_id, seq)) step_outputs (run_id TEXT, step_name TEXT, output TEXT, PRIMARY KEY (run_id, step_name)) +llm_journal (run_id TEXT, call_seq INTEGER, -- append order within the run + request_fingerprint TEXT, -- sha256 of canonical request JSON + occurrence INTEGER, -- per-attempt ordinal of the fingerprint + request_json TEXT, response_json TEXT, created_at TEXT, + PRIMARY KEY (run_id, call_seq)) ``` ### HTTP JSON contracts (`server.py`) @@ -547,7 +579,7 @@ Programmatic knobs: `AnthropicClient(api_key, max_tokens=1024)`; `OpenAICompatClient(base_url, api_key, max_tokens=1024, timeout=120.0)`; `WorkerPool(n_workers=2, poll_interval=0.05)`; `serve(host="127.0.0.1", port=8765, n_workers=2, llm_client, tools)`; `run_program(tools=...)` / -`run_durable(run_id=...)`. +`run_durable(run_id=..., journal_llm=...)`. Packaging: zero runtime deps (`pyproject.toml`); optional extra `anthropic>=0.40,<1.0` (`pyproject.toml`); `requires-python >= 3.11` diff --git a/docs/production.md b/docs/production.md index 59ed955..e8d672a 100644 --- a/docs/production.md +++ b/docs/production.md @@ -7,8 +7,8 @@ ThreadLang v0.13 retains the deliberately narrow **single-node, POSIX, local-fil - One `threadlang-serve` process per SQLite store. - Linux or macOS/POSIX filesystem with working advisory file locks. - SQLite WAL on a local disk; network filesystems are unsupported. -- Step-boundary checkpoints. A process death can repeat model/tool calls in the current incomplete step, or an incomplete `emit llm`. -- Model and tool calls are therefore at-least-once. Durable runs reject custom tools declared as both side-effecting and non-idempotent. +- Step-boundary checkpoints, with every model call journaled per run. A process death re-executes the interrupted step, but that step's completed model calls — including a completed `emit llm` — replay from the journal; at most the single in-flight model call repeats. Tool calls in the interrupted step re-execute. +- Model calls are therefore journaled-replayable and tool calls remain at-least-once. Durable runs reject custom tools declared as both side-effecting and non-idempotent. - Forward-only graphs only; `max_iters` is capped by runtime policy. ## Start safely @@ -56,11 +56,11 @@ The worker pool owns `.worker.lock`. A second process fails startup rathe - Regex output contracts execute in a killable isolated interpreter with size and time limits. - Non-idempotent side-effecting tools are rejected on the durable path. -Model and tool calls, including `emit llm`, remain at-least-once across a hard crash. Tool authors must truthfully declare `ToolSpec.side_effects` and `ToolSpec.idempotent`. Exactly-once external effects are out of scope. +Model calls, including `emit llm`, are journaled per run: across a hard crash the interrupted step's completed calls replay from the journal and at most the single in-flight call re-executes; tool calls in the interrupted step remain at-least-once. Tool authors must truthfully declare `ToolSpec.side_effects` and `ToolSpec.idempotent`. Exactly-once external effects are out of scope. ## Data and secrets -SQLite stores inputs, outputs, traces, and tool observations in plaintext. Protect the database and lock file using OS permissions and encrypted storage where required. For the OpenAI-compatible client, HTTP response bodies and endpoint details are not copied into durable errors, redirects are refused, endpoint URLs cannot embed credentials, queries, or fragments, keyed non-loopback endpoints require HTTPS, loopback HTTP bypasses environment proxies, and responses are capped at 8 MiB. Malformed OpenAI-compatible text and tool-call payloads fail before persistence or tool execution. Retention is count-based; legal/time-based erasure remains an operator responsibility. +SQLite stores inputs, outputs, traces, tool observations, and journaled model request/response payloads in plaintext. Protect the database and lock file using OS permissions and encrypted storage where required. For the OpenAI-compatible client, HTTP response bodies and endpoint details are not copied into durable errors, redirects are refused, endpoint URLs cannot embed credentials, queries, or fragments, keyed non-loopback endpoints require HTTPS, loopback HTTP bypasses environment proxies, and responses are capped at 8 MiB. Malformed OpenAI-compatible text and tool-call payloads fail before persistence or tool execution. Retention is count-based; legal/time-based erasure remains an operator responsibility. ## Upgrade to v0.13 diff --git a/src/threadlang/journal.py b/src/threadlang/journal.py new file mode 100644 index 0000000..2a35dc6 --- /dev/null +++ b/src/threadlang/journal.py @@ -0,0 +1,167 @@ +"""Per-call LLM response journaling for durable runs. + +Step checkpoints (store.py) bound crash recovery at step granularity: a crash +*inside* a step re-runs that whole step on resume, re-calling the provider for +every call the step had already completed. This module narrows that window. +`run_durable` wraps the run's LLM client in a `JournaledLLMClient`, which +records each model call — canonical request fingerprint plus response — in the +store's `llm_journal` table. On resume, a call whose fingerprint matches a +journaled row replays the recorded response instead of hitting the provider, +so a hard crash re-executes at most the single in-flight call of the +interrupted step. Tool calls are not journaled and re-execute; exactly-once +external effects remain impossible client-side and stay out of scope +(docs/production.md). + +Crash semantics: a fresh run gets a fresh run_id, so its journal is empty and +every call goes live — the journal is write-only on a first attempt. Only a +resumed run (same run_id, re-executing an incomplete step) replays +fingerprint-matched responses. + +The replay key is `(run_id, request_fingerprint, occurrence)`. The fingerprint +is SHA-256 over the canonical JSON of the full request — `kind` plus `model` +and the verb's arguments (prompt / options / messages+tools) — so distinct +requests never collide. `occurrence` is the per-attempt ordinal of that +fingerprint, so a run issuing the identical request twice keeps two rows, each +replaying its own response. + +The wrapper is transparent to the runtime's duck-typing: it exposes `complete` +unconditionally and `route`/`agent_step` only when the wrapped client has them +(the runtime probes with `getattr`). It is created per run inside +`run_durable`, so its mutable per-attempt ordinal map is never shared across +worker threads — the shared-stateless-client contract the `WorkerPool` relies +on is untouched. +""" + +from __future__ import annotations + +import hashlib +import json +from dataclasses import asdict +from typing import TYPE_CHECKING, Callable, Dict, List, Sequence, TypeVar, cast + +from .llm import AgentTurn, LLMClient, Message, ToolCall +from .tools import ToolSpec + +if TYPE_CHECKING: + from .store import RunStore + +_T = TypeVar("_T") + + +def _sha256_json(value: object) -> str: + encoded = json.dumps(value, sort_keys=True, separators=(",", ":")).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + +def _message_json(message: Message) -> Dict[str, object]: + """JSON-safe rendering of a runtime-owned message: `ToolCall` dataclasses + become plain dicts; every other key passes through. Deterministic under + `sort_keys`, so a resumed attempt hashes an identical request identically.""" + rendered: Dict[str, object] = {} + for key, value in message.items(): + if key == "tool_calls": + rendered[key] = [asdict(call) for call in cast(Sequence[ToolCall], value)] + else: + rendered[key] = value + return rendered + + +def _agent_turn_json(turn: AgentTurn) -> Dict[str, object]: + return {"text": turn.text, "tool_calls": [asdict(call) for call in turn.tool_calls]} + + +def _agent_turn_from_json(payload: object) -> AgentTurn: + """Rebuild an `AgentTurn` from its journaled payload. The journal is written + only by `_agent_turn_json`; the isinstance checks keep a hand-edited store + from crashing a resume rather than from being wrong.""" + data = cast(Dict[str, object], payload) if isinstance(payload, dict) else {} + calls: List[ToolCall] = [] + for raw in cast(List[object], data.get("tool_calls", [])): + if isinstance(raw, dict): + calls.append( + ToolCall( + id=str(raw.get("id", "")), + name=str(raw.get("name", "")), + arguments=cast(Dict[str, object], raw.get("arguments", {})), + ) + ) + return AgentTurn(text=str(data.get("text", "")), tool_calls=tuple(calls)) + + +class JournaledLLMClient: + """A per-run journaling wrapper around an LLM client. Each call is looked + up in the store's `llm_journal` table by (run_id, request fingerprint, + occurrence); a hit replays the recorded response without a provider call, + a miss calls through and persists the response.""" + + # Assigned in __init__ only when the wrapped client exposes the verb, so + # the runtime's `getattr(client, "route"/"agent_step", None)` probes see + # exactly the wrapped client's capability surface. + route: Callable[..., str] + agent_step: Callable[..., AgentTurn] + + def __init__(self, client: LLMClient, store: RunStore, run_id: str) -> None: + self._client = client + self._store = store + self._run_id = run_id + self._occurrences: Dict[str, int] = {} + if getattr(client, "route", None) is not None: + self.route = self._route + if getattr(client, "agent_step", None) is not None: + self.agent_step = self._agent_step + + def complete(self, model: str, prompt: str) -> str: + return self._call_journaled( + {"kind": "complete", "model": model, "prompt": prompt}, + lambda: self._client.complete(model=model, prompt=prompt), + lambda response: response, + str, + ) + + def _route(self, model: str, prompt: str, options: Sequence[str]) -> str: + route_fn = getattr(self._client, "route") # present: see __init__ + return self._call_journaled( + {"kind": "route", "model": model, "prompt": prompt, "options": list(options)}, + lambda: route_fn(model=model, prompt=prompt, options=list(options)), + lambda response: response, + str, + ) + + def _agent_step( + self, model: str, messages: Sequence[Message], tools: Sequence[ToolSpec] + ) -> AgentTurn: + agent_fn = getattr(self._client, "agent_step") # present: see __init__ + return self._call_journaled( + { + "kind": "agent_step", + "model": model, + "messages": [_message_json(message) for message in messages], + "tools": [asdict(spec) for spec in tools], + }, + lambda: agent_fn(model=model, messages=messages, tools=tools), + _agent_turn_json, + _agent_turn_from_json, + ) + + def _call_journaled( + self, + request: Dict[str, object], + call: Callable[[], _T], + encode: Callable[[_T], object], + decode: Callable[[object], _T], + ) -> _T: + fingerprint = _sha256_json(request) + occurrence = self._occurrences.get(fingerprint, 0) + self._occurrences[fingerprint] = occurrence + 1 + stored = self._store.load_llm_journal_entry(self._run_id, fingerprint, occurrence) + if stored is not None: + return decode(json.loads(stored)) + response = call() + self._store.save_llm_journal_entry( + self._run_id, + fingerprint, + occurrence, + json.dumps(request, sort_keys=True, separators=(",", ":")), + json.dumps(encode(response), sort_keys=True, separators=(",", ":")), + ) + return response diff --git a/src/threadlang/store.py b/src/threadlang/store.py index b9337bb..1f54485 100644 --- a/src/threadlang/store.py +++ b/src/threadlang/store.py @@ -9,16 +9,20 @@ The design keeps the runtime storage-agnostic. This module supplies: - `RunStore` — a thin sqlite wrapper (stdlib `sqlite3`, no dependency) holding - three tables: `runs`, `events`, `step_outputs`. + four tables: `runs`, `events`, `step_outputs`, and `llm_journal` (the + per-call model-call journal; see journal.py). - `run_durable()` — orchestrates a persisted run: it hands the runtime a write-through trace (every appended event lands in `events`) and a - step-complete hook (every step output lands in `step_outputs`), then marks - the run completed or failed. On resume it pre-loads the completed step - outputs and tells the runtime to skip them. + step-complete hook (every step output lands in `step_outputs`), wraps the + LLM client in a per-call journaling wrapper, then marks the run completed + or failed. On resume it pre-loads the completed step outputs and tells the + runtime to skip them. Checkpoint granularity is one step. A crash *inside* a step (mid agent loop) -re-runs that whole step on resume; steps that already finished do not. That is -the right boundary for v0.4 — coarse enough to be simple and correct, fine +re-runs that whole step on resume; steps that already finished do not. Within +the re-run step, model calls completed before the crash replay from +`llm_journal` — at most the single in-flight call re-executes. That is the +right boundary — coarse enough to be simple and correct, fine enough that a long pipeline doesn't redo completed work. """ @@ -40,7 +44,8 @@ load_ir_bytes, workflow_fingerprint, ) -from .llm import LLMClient +from .journal import JournaledLLMClient +from .llm import DryRunClient, LLMClient from .metrics import AggregateMetrics, RunMetrics, aggregate, compute_metrics, trace_span_ms from .policy import DEFAULT_MAX_PENDING_RUNS, DEFAULT_MAX_RETAINED_RUNS from .runtime import RuntimeResult, run_program @@ -81,9 +86,22 @@ PRIMARY KEY (run_id, step_name), FOREIGN KEY (run_id) REFERENCES runs(id) ON DELETE CASCADE ); +CREATE TABLE IF NOT EXISTS llm_journal ( + run_id TEXT NOT NULL, + call_seq INTEGER NOT NULL, -- append order within the run + request_fingerprint TEXT NOT NULL, -- sha256 of the canonical request JSON + occurrence INTEGER NOT NULL, -- per-attempt ordinal of this fingerprint + request_json TEXT NOT NULL, + response_json TEXT NOT NULL, + created_at TEXT NOT NULL, + PRIMARY KEY (run_id, call_seq), + FOREIGN KEY (run_id) REFERENCES runs(id) ON DELETE CASCADE +); CREATE INDEX IF NOT EXISTS idx_runs_status_created ON runs(status, created_at, id); CREATE INDEX IF NOT EXISTS idx_runs_created ON runs(created_at DESC, id DESC); CREATE INDEX IF NOT EXISTS idx_events_run ON events(run_id, seq); +CREATE INDEX IF NOT EXISTS idx_llm_journal_lookup + ON llm_journal(run_id, request_fingerprint, occurrence); """ @@ -357,6 +375,7 @@ def _prune_terminal_locked(self, max_retained: int) -> int: # foreign-key declarations existed. self._conn.execute("DELETE FROM events WHERE run_id = ?", (run_id,)) self._conn.execute("DELETE FROM step_outputs WHERE run_id = ?", (run_id,)) + self._conn.execute("DELETE FROM llm_journal WHERE run_id = ?", (run_id,)) self._conn.execute("DELETE FROM runs WHERE id = ?", (run_id,)) return len(ids) @@ -512,6 +531,47 @@ def load_step_outputs(self, run_id: str) -> Dict[str, str]: ).fetchall() return {r["step_name"]: r["output"] for r in rows} + # ----- LLM call journal (the replay half of journal.py) ----- + + def load_llm_journal_entry( + self, run_id: str, request_fingerprint: str, occurrence: int + ) -> Optional[str]: + """The recorded response JSON for the `occurrence`-th call with this + request fingerprint in `run_id`, or None — in which case the caller + must go live and then persist via `save_llm_journal_entry`.""" + row = self._conn.execute( + "SELECT response_json FROM llm_journal " + "WHERE run_id = ? AND request_fingerprint = ? AND occurrence = ?", + (run_id, request_fingerprint, occurrence), + ).fetchone() + return row["response_json"] if row is not None else None + + def save_llm_journal_entry( + self, + run_id: str, + request_fingerprint: str, + occurrence: int, + request_json: str, + response_json: str, + ) -> None: + row = self._conn.execute( + "SELECT COALESCE(MAX(call_seq), -1) + 1 AS next FROM llm_journal WHERE run_id = ?", + (run_id,), + ).fetchone() + self._conn.execute( + "INSERT INTO llm_journal (run_id, call_seq, request_fingerprint, occurrence, " + "request_json, response_json, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)", + ( + run_id, + row["next"], + request_fingerprint, + occurrence, + request_json, + response_json, + _now(), + ), + ) + # ----- metrics (a derived view of the persisted trace) ----- def _event_timestamps(self, run_id: str) -> List[Optional[str]]: @@ -580,6 +640,7 @@ def run_durable( run_id: Optional[str] = None, source: Optional[str] = None, claimed: bool = False, + journal_llm: bool = True, ) -> DurableRun: """Execute a program with its trace and step checkpoints persisted to `store`. @@ -589,6 +650,13 @@ def run_durable( one. Passing the id of an active run is rejected; passing a `completed` run returns its stored result without re-executing. Omit `run_id` to start a fresh run. Queue workers set `claimed=True` only after an atomic claim. + + Model calls are journaled per call (`journal_llm=False` opts out): on + resume, a call whose request fingerprint matches a journaled row replays + the recorded response instead of re-calling the provider, so a crash + re-executes at most the interrupted step's single in-flight call. A fresh + run always calls live. Exactly-once stays out of scope; see journal.py and + docs/production.md. """ if tools is not None: for step in program.steps.steps: @@ -690,6 +758,14 @@ def run_durable( trace = _WriteThroughTrace(store, run_id) + client: LLMClient = llm_client if llm_client is not None else DryRunClient() + if journal_llm: + # Per-call response journal: on resume, the interrupted step's + # completed calls replay from `llm_journal`; at most the call that was + # in flight at crash time re-executes. A fresh run_id's journal is + # empty, so a first attempt always calls live. + client = JournaledLLMClient(client, store, run_id) + def _checkpoint(step_name: str, output: str) -> None: store.save_step_output(run_id, step_name, output) @@ -697,7 +773,7 @@ def _checkpoint(step_name: str, output: str) -> None: result = run_program( program, inputs, - llm_client=llm_client, + llm_client=client, tools=tools, trace=trace, resume_outputs=resume_outputs, diff --git a/tests/test_llm_journal.py b/tests/test_llm_journal.py new file mode 100644 index 0000000..af60944 --- /dev/null +++ b/tests/test_llm_journal.py @@ -0,0 +1,263 @@ +"""LLM response journal tests — per-call checkpointing inside a step. + +What these guard — the narrowed crash window the journal claims: + 1. A run that crashes mid-step replays the step's already-journaled model + calls from the store on resume instead of re-calling the provider; at + most the single in-flight call re-executes (asserted by counting calls). + 2. Replay is keyed by request fingerprint + per-attempt occurrence: a + fingerprint the journal has never seen falls through to a live call, a + fresh run always calls live, and identical requests keep distinct rows. + 3. The wrapper preserves the client's duck-typed capability surface and + round-trips `AgentTurn` through serialization. + 4. `journal_llm=False` restores the plain step-level replay window. + +All offline: scripted clients crash one specific call, so resume is exercised +deterministically without a network or an API key. +""" + +from __future__ import annotations + +from pathlib import Path +import sys +from typing import Dict, List, Sequence + +REPO_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(REPO_ROOT / "src")) + +import pytest # type: ignore # noqa: E402 + +from threadlang.journal import JournaledLLMClient # noqa: E402 +from threadlang.llm import AgentTurn, DryRunClient, ToolCall # noqa: E402 +from threadlang.parser import parse_program # noqa: E402 +from threadlang.store import RunStore, run_durable # noqa: E402 +from threadlang.tools import ToolSpec # noqa: E402 + +_ONE_STEP = """ +thread Pipe { + context {} + steps { step a { llm "m1" { "A:" + inputs.x } } } + emit text { steps.a.output } +} +""" + +_AGENT_PIPE = """ +thread Pipe { + context {} + steps { + step a { llm "m1" { "A:" + inputs.x } } + step b { agent "m2" { tools [ echo ] max_iters 4 "B:" + steps.a.output } } + } + emit text { steps.b.output } +} +""" + + +class _CountingCompleteClient: + """A `complete`-only client that counts provider invocations and returns a + distinct response per call, so a replay is detectable by value.""" + + def __init__(self) -> None: + self.calls: List[str] = [] + + def complete(self, model: str, prompt: str) -> str: + self.calls.append(prompt) + return f"response-{len(self.calls)}" + + +class _CrashyAgentClient: + """Answers `complete` for step a; for the agent step, requests the echo + tool on turns 1–2, raises on turn 3 while armed (the simulated crash), and + answers with a final text once disarmed. Counts every provider call.""" + + def __init__(self) -> None: + self.complete_calls = 0 + self.agent_calls = 0 + self._armed = True + + def complete(self, model: str, prompt: str) -> str: + self.complete_calls += 1 + return f"[{model}] {prompt}" + + def agent_step( + self, model: str, messages: Sequence[Dict[str, object]], tools: Sequence[ToolSpec] + ) -> AgentTurn: + self.agent_calls += 1 + if len(messages) < 5: + return AgentTurn( + text="", + tool_calls=( + ToolCall(id=f"call_{len(messages)}", name="echo", arguments={"text": "hi"}), + ), + ) + if self._armed: + self._armed = False + raise RuntimeError("simulated crash in agent step b") + return AgentTurn(text="final answer", tool_calls=()) + + +def _journal_rows(store: RunStore, run_id: str) -> int: + row = store._conn.execute( + "SELECT COUNT(*) AS c FROM llm_journal WHERE run_id = ?", (run_id,) + ).fetchone() + return row["c"] + + +def test_resume_replays_journaled_calls_within_the_interrupted_step(tmp_path: Path) -> None: + store = RunStore(str(tmp_path / "runs.db")) + program = parse_program(_AGENT_PIPE) + client = _CrashyAgentClient() + + # First attempt: step a completes and is checkpointed; the agent step's + # turns 1–2 complete (and are journaled); turn 3 raises mid-step. + with pytest.raises(Exception, match="simulated crash"): + run_durable(program, {"x": "hi"}, store, llm_client=client) + run_id = store.list_runs()[0].id + assert set(store.load_step_outputs(run_id)) == {"a"} + assert (client.complete_calls, client.agent_calls) == (1, 3) + + # Resume: step a is skipped from its checkpoint, turns 1–2 replay from the + # journal (their requests hash identically — turn 2's request carries the + # assistant/tool message history), and only turn 3 calls live. + durable = run_durable(program, {"x": "hi"}, store, llm_client=client, run_id=run_id) + assert durable.result.output == "final answer" + assert store.get_run(run_id).status == "completed" # type: ignore[union-attr] + assert client.complete_calls == 1, "checkpointed step a must not re-call the provider" + assert client.agent_calls == 4, "resume must re-execute only the in-flight call" + # Journal: step a + turns 1–2 written pre-crash; turn 3 written on resume. + assert _journal_rows(store, run_id) == 4 + store.close() + + +def test_replay_is_keyed_by_fingerprint_and_occurrence(tmp_path: Path) -> None: + store = RunStore(str(tmp_path / "runs.db")) + run_id = store.create_run("Pipe", {"x": "hi"}) + client = _CountingCompleteClient() + + first = JournaledLLMClient(client, store, run_id) + assert first.complete(model="m", prompt="p") == "response-1" + assert first.complete(model="m", prompt="p") == "response-2" # occurrence 1, live + assert len(client.calls) == 2 + + # A new wrapper over the same run_id is the resume shape: both occurrences + # replay their own recorded responses without touching the provider. + resumed = JournaledLLMClient(client, store, run_id) + assert resumed.complete(model="m", prompt="p") == "response-1" + assert resumed.complete(model="m", prompt="p") == "response-2" + assert len(client.calls) == 2, "fingerprint-matched calls must replay from the journal" + + # A request the journal has never seen falls through to a live call. + assert resumed.complete(model="m", prompt="changed") == "response-3" + assert len(client.calls) == 3 + + # A fresh run_id has an empty journal, so it always calls live. + fresh = JournaledLLMClient(client, store, store.create_run("Pipe", {"x": "hi"})) + assert fresh.complete(model="m", prompt="p") == "response-4" + assert len(client.calls) == 4 + store.close() + + +def test_wrapper_mirrors_the_wrapped_client_capability_surface(tmp_path: Path) -> None: + store = RunStore(str(tmp_path / "runs.db")) + run_id = store.create_run("Pipe", {}) + + bare = JournaledLLMClient(_CountingCompleteClient(), store, run_id) + assert getattr(bare, "route", None) is None + assert getattr(bare, "agent_step", None) is None + + full = JournaledLLMClient(DryRunClient(), store, run_id) + assert getattr(full, "route", None) is not None + assert getattr(full, "agent_step", None) is not None + store.close() + + +def test_route_replays_from_the_journal(tmp_path: Path) -> None: + class _CountingRouteClient: + def __init__(self) -> None: + self.calls = 0 + + def complete(self, model: str, prompt: str) -> str: + raise AssertionError("route steps with a route-capable client never complete") + + def route(self, model: str, prompt: str, options: Sequence[str]) -> str: + self.calls += 1 + return options[-1] + + store = RunStore(str(tmp_path / "runs.db")) + run_id = store.create_run("Pipe", {}) + client = _CountingRouteClient() + + first = JournaledLLMClient(client, store, run_id) + assert first.route(model="m", prompt="p", options=["a", "b"]) == "b" + resumed = JournaledLLMClient(client, store, run_id) + assert resumed.route(model="m", prompt="p", options=["a", "b"]) == "b" + assert client.calls == 1 + # Options are part of the fingerprint: a different closed set calls live. + assert resumed.route(model="m", prompt="p", options=["a", "c"]) == "c" + assert client.calls == 2 + store.close() + + +def test_agent_step_round_trips_through_the_journal(tmp_path: Path) -> None: + class _ScriptedAgentClient: + def __init__(self) -> None: + self.turns = 0 + + def agent_step( + self, model: str, messages: Sequence[Dict[str, object]], tools: Sequence[ToolSpec] + ) -> AgentTurn: + self.turns += 1 + return AgentTurn( + text="done", + tool_calls=(ToolCall(id="c1", name="echo", arguments={"text": "hi"}),), + ) + + store = RunStore(str(tmp_path / "runs.db")) + run_id = store.create_run("Pipe", {}) + client = _ScriptedAgentClient() + tools = [ToolSpec(name="echo", description="echo", parameters={"type": "object"})] + messages: List[Dict[str, object]] = [{"role": "user", "content": "go"}] + + turn = JournaledLLMClient(client, store, run_id).agent_step( + model="m", messages=messages, tools=tools + ) + assert client.turns == 1 + + replayed = JournaledLLMClient(client, store, run_id).agent_step( + model="m", messages=messages, tools=tools + ) + assert client.turns == 1, "the replayed turn must not call the provider" + assert replayed == turn + assert replayed.tool_calls[0].arguments == {"text": "hi"} + store.close() + + +def test_journal_opt_out_restores_step_level_replay(tmp_path: Path) -> None: + store = RunStore(str(tmp_path / "runs.db")) + program = parse_program(_AGENT_PIPE) + client = _CrashyAgentClient() + + with pytest.raises(Exception, match="simulated crash"): + run_durable(program, {"x": "hi"}, store, llm_client=client, journal_llm=False) + run_id = store.list_runs()[0].id + + durable = run_durable( + program, {"x": "hi"}, store, llm_client=client, run_id=run_id, journal_llm=False + ) + assert durable.result.output == "final answer" + assert client.complete_calls == 1 + assert client.agent_calls == 6, "opt-out re-executes every call of the interrupted step" + assert _journal_rows(store, run_id) == 0 + store.close() + + +def test_terminal_prune_deletes_journal_rows(tmp_path: Path) -> None: + store = RunStore(str(tmp_path / "runs.db")) + durable = run_durable(parse_program(_ONE_STEP), {"x": "hi"}, store, llm_client=DryRunClient()) + assert _journal_rows(store, durable.run_id) == 1 + + # Retention pruning removes the terminal run's journal along with its + # events and checkpoints (explicit deletes, for pre-foreign-key stores). + store.enqueue_run("Pipe", _ONE_STEP, {"x": "2"}, max_retained=0) + assert store.get_run(durable.run_id) is None + assert _journal_rows(store, durable.run_id) == 0 + store.close()