TwinEdgeAI

↑ Course home

← Previous: Week 3, Tool Use and Function Calling  ·  Next: Week 4a, Agent Design Patterns and Scaffolds →

Engineering AI Agents · Week 3b

Frameworks and Orchestration

How to build an agent system that you can change, test and trust: the three pillars of data, system and evaluation, DSPy for programs that tune their own prompts, LangGraph for state and control flow, open coding-agent harnesses, Temporal for durable runs, and the tools that let you see what happened.

Author: Grok Bot Version: 1.0, October 28, 2026 Study time: about 12 hours

CreditThis coursework is adapted from, and inspired by, Stanford CS329Z: Engineering AI Agents (Diyi Yang, Michael Ryan, John Yang), Week 3, "Frameworks & Agent Design: Frameworks & Orchestration" (Wednesday, October 7). The topic list and the three required readings come from the course's public schedule [40]. I checked the topic order against the public lecture slides [41], and I looked at the course's public code repository for this lecture [42] to see what it covers. The notes, lab, quiz and assignment are original writing. The lab uses a different example (support tickets) and its own code. This pack does not reproduce the slides or the course code. It is not an official Stanford document. I added extra sources of my own. Each is marked "added" in the references.

1. Week overview and learning objectives

Last time the model got hands. This time we ask how to build the whole body around it. A real agent is not one model call. It is prompts, tools, state, retries, a log of what happened, and a way to tell whether a change made things better. Code that wires these together has a name: a framework, or, for coding agents, a harness.

The lecture sorts the work into three pillars. Data is what the system learns from. System is what runs. Evaluation is how you know it works. Frameworks mostly help with the system pillar. That is also where it is easy to add layers that hide what the model sees. So this week has a second theme: use a framework when it saves you work you would otherwise do badly, and read what it sends to the model.

We walk through six tools. DSPy lets you write each model call as a typed signature and then tune the prompts with an optimizer. LangGraph gives you state, nodes and edges, with checkpoints and a pause for a human. Three open coding agents (mini-SWE-agent, OpenHands and Pi) show how far a harness can be stripped down. Temporal makes a long run survive a crash. And tracing tools, built on a shared naming scheme from OpenTelemetry, show what each step cost.

The lab builds a small version of each of the five ideas with the standard library only: a mini-DSPy, a mini-LangGraph with interrupt and resume, a bash-only agent loop, a durable runtime with crash and replay, and a tracer with token and cost accounting. A mock model stands in for the LLM, so you need no API key.

Learning objectives

After this week you should be able to:

1. Name the three pillars of agent engineering and say which one a given problem belongs to. [2.1]
2. Place a task on the agency spectrum, from a single call to a workflow to an agent, and say what you give up at each step. [2.2]
3. Explain signatures, modules and optimizers in DSPy, and say what a bootstrapping optimizer does and does not fix. [2.3]
4. Describe a LangGraph program in terms of state, reducers, nodes, edges, checkpoints and interrupts, and say why a node restarts after an interrupt. [2.4]
5. Compare mini-SWE-agent, OpenHands and Pi on tools, history and safety. [2.6]
6. Read the HarnessTax results and say what they do and do not show about harness choice. [2.7]
7. Explain why Temporal workflow code must be deterministic, how replay works, and why activities should be idempotent. [2.8]
8. Describe a trace as a tree of spans, name the OpenTelemetry GenAI span kinds, and compute tokens and cost from spans. [2.9]
9. Apply a short list of best practices and say when not to use a framework. [2.10, 2.11]
10. Build and test a signature parser, an optimizer, a state graph, a sandboxed agent loop, a replaying runtime and a tracer. [Lab]

How to use this pack

Time	What to do
2.5 hours	Read the notes (section 2). Skim the figures first.
3.5 hours	Read the three required readings with the reading guide questions. For the LangGraph overview and the HarnessTax post, read the whole page. For the DSPy paper, read the sections listed in the guide.
3 hours	Do the lab and at least five exercises.
1 hour	Discussion and the quiz.
2 hours	The graded assignment.

2. Lecture notes

2.1 Why frameworks, and the three pillars

A demo agent is a prompt, a loop and a few tools. It works on the example you tried. A product has to work on the next thousand, survive failures, cost a sane amount, and be changed by a team without breaking. The code that turns the first into the second is where frameworks live.

The lecture frames agent engineering as three pillars. I will use the same three words, and fill them in with my own examples. Figure 1 shows how they connect.

Pillar	What it holds	Questions it answers
Data	Training and fine-tuning data. Demonstrations. Rollouts you sample from your own system. Human feedback. The handful of examples that tune a prompt.	What does the system learn from? Where do good examples come from? Who labels them?
System	One model call, a compound pipeline of calls and code, or an agent loop with tools. The harness around it: prompts, state, retries, limits.	What runs at inference time? Who decides the next step: your code or the model?
Evaluation	A metric (exact match, pass rate, a model as judge). A set of cases. Cost and latency.	Did the change help? By how much? At what price? How sure are we?

The pillars are not separate. An optimizer such as the one in DSPy turns data (a few examples) and evaluation (a metric) into a better system (new prompts) [1]. A trace from a production run is data too: it can become a test case or a demonstration. This is why the pillars come as a set. If you work on only the system, you cannot tell whether your change helped. If you work on only evals, you have nothing to improve.

2.2 The agency spectrum and the landscape

Anthropic draws a line that is useful here. A workflow is a system in which code decides the path through a set of model calls and tools. An agent is a system in which the model decides its own steps and tool use [30]. A single call sits at one end. A free-running agent sits at the other. Figure 2 puts this line at the top, and puts the tools of this week under the part of the line where they help most.

Moving right buys flexibility. You pay in four ways. Cost goes up, because there are more calls. Latency goes up. Errors compound, since each step can go wrong. And testing gets harder, because the path is not fixed. Anthropic's guidance is to add complexity only when it shows a gain, and to look first at the simplest pattern that could work [30].

The 12-factor agents project, in the HumanLayer repository, makes a related point from experience. Its README says the author saw few frameworks in customer-facing agents, that many products billed as agents are "mostly deterministic code, with LLM steps sprinkled in", and it lists twelve principles, such as "Own your prompts", "Own your context window" and "Own your control flow" [31]. I read it as an opinion piece, not a standard. It is useful as a checklist of things a framework may take away from you.

2.3 DSPy: signatures, modules, optimizers

Required reading 1 is the DSPy paper by Khattab and co-authors [1]. Its starting complaint is that many LM pipelines are built from "hard-coded prompt templates", long strings found by trial and error. DSPy treats the pipeline as a program instead. The paper's abstract describes it as a programming model that abstracts LM pipelines as text transformation graphs, in which LMs are called through declarative modules [1]. There are three ideas. Figure 3 shows how they fit.

Signatures: say what, not how

A signature is a typed declaration of a step's inputs and outputs. The paper's shortest example is "question -> answer" [1]. The docs add typed fields and instructions, and describe a signature as a declarative spec of what a module needs to do, rather than how to prompt for it [4]. For the lab, I made up a different one: "ticket -> category". A string like that is enough to build a prompt, and to parse the model's reply back into fields.

Modules: say how

A module is a reusable strategy that takes a signature. The paper names Predict (one call), ChainOfThought (adds a reasoning field before the answer), ProgramOfThought, MultiChainComparison and ReAct [1]. The current docs list these and more, such as BestOfN and RLM [5]. A program is ordinary Python that calls modules, so you can use loops, branches and your own functions. The lecture slides use the same three-level picture: signatures are the what, modules are the how, and optimizers tune the result.

Optimizers: tune the prompts for a metric

An optimizer (the paper calls them "teleprompters"; the docs say they were renamed) takes a program, a metric and a few training inputs, and returns a better program [1] [6]. The first one to know is BootstrapFewShot. As the paper describes it, it runs a teacher program (by default the zero-shot program itself) on training inputs, records the traces of every module, and keeps a trace only if the final output passes the metric. Those kept traces become demonstrations for each module [1]. Other optimizers do more: BootstrapFewShotWithRandomSearch tries several candidates and keeps the best on a dev set. MIPROv2 comes from the MIPRO line of work, which optimizes the free-form instructions and the few-shot demonstrations of every module, and reports gains of up to 13% accuracy over baseline optimizers on five of seven programs with Llama-3-8B [7]. GEPA samples trajectories, reflects on them in natural language and proposes prompt updates [9]. The docs suggest starting with BootstrapFewShot at about 10 examples, and moving to random search at 50 or more [6].

What can an optimizer do? The paper's GSM8K table is the clearest evidence. With GPT-3.5, a plain Predict program scored 24.0 on the dev set with no compilation, and 64.7 after two rounds of bootstrapping. A ChainOfThought program went from 50.0 to 80.3 with one round of bootstrapping, and 72.9 on the test set [1]. The abstract summarizes the case studies as outperforming standard few-shot prompting, generally by over 25% for GPT-3.5 and 65% for Llama2-13b-chat [1]. These are results on math word problems with a few hundred training questions. Do not expect the same gain on your task.

When to use it, and when not

• Good fit: a pipeline with a clear metric and a few dozen examples, where you are tempted to spend days editing prompt strings. Swapping the model also means re-tuning, and an optimizer makes that a command.
• Poor fit: you have no metric, or only a handful of vague examples. The optimizer then has nothing to climb. Write the eval first (section 2.11).
• Watch for: overfitting to the dev set. The paper reports its extensive comparisons on a dev set to avoid overfitting the test set [1]. In my lab, train and dev tickets share templates, so the dev scores are optimistic.
• Constraints: the DSPy Assertions paper adds checks inside a program, so a failed constraint can trigger a retry with feedback [8]. This is a way to put a rule into the program instead of into a longer prompt.

2.4 LangGraph: state, nodes, edges

Required reading 2 is the LangGraph overview [2]. The page calls LangGraph "a low-level orchestration framework and runtime for building, managing, and deploying long-running, stateful agents". It says the core strength is mixing hand-coded, deterministic steps with LLM-driven steps in one graph. It also says LangGraph does not abstract prompts or architecture [2]. That last point sets it apart from DSPy: DSPy writes prompts for you, and LangGraph leaves them to you and manages control flow and state.

Three parts of a graph

• State. The graph docs say the state is a schema plus reducer functions that say how an update is applied. Each key has its own reducer. If no reducer is given, an update replaces the old value [10]. A list of messages usually has an "append" reducer, so that two nodes can both add to it.
• Nodes. A node is a function. It reads the state and returns a dictionary of updates. A node can call a model, call a tool, or run plain code.
• Edges. A normal edge says "after A, run B". A conditional edge runs a routing function on the state and returns the name of the next node. A node with several outgoing edges starts all of its targets together in the next step [10].

How it runs: super-steps

The docs describe execution as message passing in rounds, which they call super-steps. Nodes that run in parallel are in the same super-step, and nodes that run one after another are in separate ones [10]. Two things follow. First, you can set a limit on super-steps, and the run raises GraphRecursionError when it is hit. The docs state a default of 1000 starting in version 1.0.6 [10]. A loop with no exit is the usual cause. Second, state is saved at super-step boundaries. That is where checkpoints come from.

Checkpoints, threads and time travel

When you compile a graph with a checkpointer, LangGraph saves the state after each super-step, under a thread_id that names one conversation or run [11]. The persistence page lists four uses: continuing a conversation, human-in-the-loop, replay or "time travel", and fault tolerance [11]. The in-memory saver loses everything when the process restarts. The docs point to SQLite and Postgres savers for real use [11]. So "my agent resumed" is only true if the checkpoint is stored somewhere that outlives the process.

Interrupts: a person in the loop

A node can call interrupt(value). The run stops, and the caller sees the value. Later the caller resumes with Command(resume=answer) on the same thread, and the call to interrupt returns that answer. This needs a checkpointer and a thread id [12]. There is one rule you must learn early. On resume the node does not continue from the line of the interrupt. The docs say it restarts from the beginning, so any code before the interrupt runs again, and side effects before it must be idempotent [12]. My lab's graph does the same, and its demo shows the "approve" node running twice for one approval.

What the graph is good for

• Flows with branches, loops and checks that you want to see as a diagram: route, draft, review, send.
• Runs that pause for a person, or for a day, and need to resume with the same state.
• Teams that want one place to read the control flow, instead of control flow hidden inside prompts.

It is not magic about the model. LangGraph does not make the model's calls better. It makes the program around the calls easier to test and to resume. The overview also names a higher-level option: the LangChain agent abstractions are built on top of LangGraph, and the docs recommend them if you just want a prebuilt loop [2]. The lecture's notebook shows both levels, a hand-built graph and the prebuilt create_agent.

2.5 Other frameworks in one table

Many more frameworks exist. I list four that have public docs, and describe each in the terms its own docs use. I did not build anything with them for this pack. Treat the table as a map, and read the docs before you choose.

Framework	Central ideas, from its docs	Closest idea in this pack
OpenAI Agents SDK [33]	Agents, handoffs between agents, guardrails, and built-in tracing.	A harness with a few primitives, and tracing (2.9).
CrewAI [34]	Agents grouped into crews, plus "flows": start, listen and router steps with managed state that can persist and resume.	Role-based multi-agent, with a state-graph-like layer (2.4).
LlamaIndex Workflows [35]	Event-driven steps: a step receives an event and emits another.	Like a graph, but driven by events (2.4).
Pydantic AI [36]	Typed agents built on Pydantic. Its docs list durable-execution integrations with Temporal, DBOS, Prefect and Restate.	Typed signatures (2.3) and durability (2.8).

Two questions help you compare any of them. Who decides the next step? (your code, an event, or the model). Where does state live, and can it survive a crash? If the answer to the second is "in memory", you will need something like section 2.8.

2.6 A tour of agent harnesses

Meta-Harness, a 2026 paper, defines a harness as the code that determines what information an LLM system stores, retrieves and presents to the model [18]. The HarnessTax post gives a close definition for coding agents: a software system that manages a model's tools, context and task execution [3]. In week 3 you built a small harness for tool calls. Coding agents are the best place to study harness design, because they are used a lot, their tasks are scored, and several are open source.

Where the idea came from: the agent-computer interface

SWE-agent made a claim that shaped later work. Language-model agents are a new kind of user, and they need interfaces built for them, as people use IDEs. Its custom agent-computer interface (ACI) improved how an agent creates and edits files, navigates a repository and runs tests. The abstract reports a pass@1 of 12.5% on SWE-bench and 87.7% on HumanEvalFix [14]. The lesson is that the shape of the tools matters, not only the model. Those numbers are from 2024 and are far below what newer agents reach.

mini-SWE-agent: bash and nothing else

The README calls it "the 100 line AI agent". It makes four design choices that the lecture also stresses. It needs no tool-calling interface: the model writes a bash command, and the harness runs it. It keeps a completely linear history, where each step just appends to the messages. It runs each action with subprocess.run, so every action is independent, and there is no stateful shell session. And it makes sandboxing easy: the README says you can switch subprocess.run for docker exec [13]. It also reports a score above 74% on SWE-bench Verified, which is a claim by the project, as of the day I read the page. The cost of this simplicity is that nothing carries over between commands. A cd in one action is gone in the next. The model must repeat the directory, or chain commands with &&. My lab builds this loop in part 3 and shows that effect.

OpenHands: events, a sandbox and a persistent terminal

OpenHands is a larger platform. The paper describes a sandboxed runtime (a Docker-based operating system with a bash shell, a web browser and an IPython server), an interface for agents to write code, run commands and browse, and the delegation of work between agents [15]. The lecture adds three details I did not check in the code. Each step is stored as an event, one JSON file per event, and the messages sent to the model are rebuilt from those events at each step. The terminal is persistent, a tmux session, so cd and environment variables carry over. And a stuck detector ends a run after four repeats of the same action and result. The event log is the big idea. If the history is a log of events, you can replay it, trim it, show it in a UI, or rebuild a different prompt from it. The lab's rebuild_messages does this on a small scale.

Pi: four tools, a tiny prompt, no safety net

Pi's README describes a "minimal, extensible agent harness". Its default tools are read, write, edit and bash, with a few optional extras. By design it leaves out sub-agents and a plan mode [16]. The author's blog post gives the reasons. The system prompt and tool definitions together come in under 1,000 tokens. It runs in "YOLO" mode by default, with no permission prompts, and the post argues that this is how people end up running agents anyway [17]. The post lists session management with continue, resume and branching, and the lecture describes the history as a tree, so you can branch from an earlier message. The repository README I read lives at earendil-works/pi. The lecture and older links may point to the earlier address, so search by name if a link fails [16]. My lab's SessionTree and FileTools are small versions of those two ideas.

"No safety net" is a design position, not advice. The lab's sandbox is the opposite choice: an allowlist. Neither is a real security boundary. If a model can run shell commands against your files, use a container or a VM.

A comparison

	mini-SWE-agent [13]	OpenHands [15]	Pi [16] [17]
Tools	bash only	bash, IPython, browser, file edits	read, write, edit, bash
History	one linear list of messages	an event stream, messages rebuilt per step (lecture)	a tree of messages, with branching
Shell state	none: a new subprocess per action	persistent terminal (lecture)	bash as a tool; tmux suggested for long jobs
Safety	your choice of environment	Docker sandbox	none by default
Extras	almost none	delegation between agents; a stuck detector (lecture)	extensible by design; no plan mode and no sub-agent tool

The lecture also describes OpenClaw, an assistant with a scheduled "heartbeat". I did not read its source or docs, so I do not describe it here.

Harnesses can be searched, and traces can be edited

If the harness matters, you can tune it like a model. Meta-Harness uses a coding agent as a proposer that reads the source, scores and traces of earlier harness candidates from a filesystem, and writes new harness code. The paper reports a gain of 7.7 points on online text classification with 4 times fewer context tokens, and says its discovered harnesses surpass the best hand-engineered baselines on TerminalBench-2 [18]. Shepherd goes the other way: it records an agent's run as a reversible, Git-like trace, so that a "meta-agent" can inspect it, revert to a past state, and replay [19]. The lecture cites Shepherd under "make failures observable and recoverable". Both papers are recent preprints, and I read only their abstracts.

2.7 What the harness costs: HarnessTax

Required reading 3 asks a clean question: for the same model, how much does the choice of harness change success and cost [3]? The post is from the Arena team, with authors Pan, Yang, Arabzadeh, Chiang, Stoica and Zaharia in its citation block. It was published on September 16, 2026, and last updated on September 18, 2026. Figure 6 shows its headline numbers.

The setup

• 21 pairs: seven models, each with three harnesses (Claude Code, Codex CLI and Pi).
• Two benchmarks: SWE-bench Lite and Terminal-Bench 2.0.
• Sample: the same 30 randomly picked tasks from each benchmark, three runs per task, a cap of 100 turns per attempt, and each harness at its high effort setting.
• Cost: computed from tokens with one fixed price list dated September 1, 2026, so a harness cannot win by price alone.
• Uncertainty: 95% intervals from 10,000 bootstrap resamples of the 30 task averages.

The findings, in the post's own terms

1. Cost moves more than success. The same model can reach similar success at up to 5 times the cost. For Claude Fable 5, the post reports 97.8% in Claude Code and 96.7% in Codex and in Pi, with Claude Code at $1.33 against $0.67 for Pi. It reports Claude Code at about 2.0 times Pi and 1.6 times Codex on SWE-bench Lite, and about 1.5 times Pi on Terminal-Bench 2.0 (geometric means of cost ratios). The average effect of the harness on success stays within about 2 points on SWE-bench Lite and about 5 on Terminal-Bench 2.0 [3].
2. A simple harness can be competitive. Pi, with four tools, reaches the Pareto frontier on both benchmarks. Claude Code's mean first-call context is over 10 times Pi's, with longer instructions and larger tool schemas. The post notes that total cost also depends on caching, output tokens and later calls [3].
3. Models can do well outside their own provider's harness. Across the six Anthropic and OpenAI models and both benchmarks, an alternative harness had the highest observed success in nine of twelve comparisons [3].

Why it belongs in this week

It makes the case for the third pillar. Without cost on the axis, you would pick the harness with 97.8% and never know it costs twice as much. Kapoor and co-authors argued the same point in 2024: a narrow focus on accuracy leads to agents that are needlessly complex and costly, and one should optimize accuracy and cost jointly [32]. The lab's pareto_front and bootstrap_ci are the two tools you need to repeat this kind of comparison on your own tasks.

2.8 Durable orchestration with Temporal

An agent run can take minutes, hours or days. It calls models that time out, tools that fail, and people who answer tomorrow. If the process dies in the middle, a plain loop loses everything. Durable execution is the idea that the program's progress is saved, so that a new process can pick up where the old one stopped. Temporal is the best-known system for this. The lecture uses it, and its docs now name LLM calls as a typical case [20] [22].

Workflows and activities

Temporal splits your code in two. A workflow holds the control flow: do this, wait, then do that. An activity holds work that touches the outside world: an API call, a database query, an LLM call, file I/O [20]. A worker is a process that runs both, and it takes tasks from a task queue. The workflow function is ordinary code. What makes it durable is where its progress is stored.

The event history, and replay

Every step the workflow takes becomes an entry in the event history, a complete and durable log of what happened in that workflow run. The docs say that commands, such as "schedule this activity" or "start this timer", are turned into events that are persisted in case of failure [21]. If the worker dies, another worker starts the workflow again from the top, and the SDK replays it against the history. When the code reaches a step that is already in the history, the SDK returns the recorded result instead of running it. The workflow docs state that an activity's result is reused during replay and not recomputed, and that timers are recorded as events and do not wait again [20]. So an LLM call that already succeeded is not paid for twice.

The rule of determinism

Replay only works if the code takes the same steps in the same order each time. The docs say this directly: workflow code must be deterministic to support replay, and non-deterministic work such as API calls and LLM invocations belongs in activities [22]. Reading the clock, drawing a random number, or iterating a set in an unstable order inside a workflow can change the path. The workflow docs say the SDKs give you replacements: time is read from the workflow context so it matches the history, and random values can be captured once and reused [20]. If a replay finds that the code now asks for something different from the history, it fails with a non-determinism error rather than guess. My lab's runtime does the same, with a hash of the activity's name and arguments.

This has a practical cost. You cannot casually edit a workflow that has runs in flight. A change that adds, removes or reorders a step breaks the replay of old runs. The workflow definition page has a section on handling code changes and non-deterministic behavior [22]. I did not test versioning, and my lab does not model it.

Retries, and why activities must be idempotent

Activities are retried by default. The default policy uses exponential backoff with a coefficient of 2.0, a first interval of 1 second, a cap of 100 seconds, and unlimited attempts [24]. A workflow, by contrast, is not retried by default [24]. You can mark some errors as non-retryable. My lab's default of three attempts is my own choice.

Retries mean an activity can run more than once. There is also a gap: the activity can finish its work, and the worker can die before the result is recorded. The server then sees no result, and the activity runs again. Temporal recommends that activities be idempotent, so that retries cause no duplicate side effects [23]. The usual way is to send a key that stays the same across attempts, so the outside system can drop a repeat. Part 4 of the lab shows this with a crash placed in exactly that gap. Without a key, "notify" runs twice and applies twice. With a key, it runs twice and applies once.

What it gives an agent

Problem	How durable execution helps
The process crashes mid-run	A new worker replays the history and continues. Finished model calls are not repeated.
A tool or model API is flaky	The activity retry policy backs off and tries again, without code in your agent loop.
The agent must wait three days for a person	A timer or a signal costs nothing while it waits. No process needs to stay alive.
You need an audit trail	The event history is a record of every step and result.

Temporal is not the only option. The Pydantic AI docs list durable execution with Temporal, DBOS, Prefect and Restate, and say that with Temporal every model and tool call becomes a durable activity [36]. LangGraph's checkpointers cover part of the same ground, saving state at step boundaries [11]. The difference is that a checkpoint saves state, and Temporal saves the history of events that built the state. Both need you to think about repeated side effects.

2.9 Observability: traces, tokens, cost

When a normal program fails you read a stack trace. When an agent fails, the bug may be in a prompt, in a tool result three steps back, or in a retry that doubled the bill. You need a record of each step. This is observability: logging enough that you can answer questions about a run you did not watch.

A trace is a tree of spans

The standard idea, from OpenTelemetry, is a trace made of spans. A span is one timed operation, with a name, a start and end time, a parent, and key-value attributes. A model call is a span. A tool call is a span. An agent run is a span that holds them. Figure 8 shows a real trace from the lab.

OpenTelemetry's GenAI conventions

If every tool names things differently, you cannot move traces between tools. OpenTelemetry has a set of semantic conventions for generative AI to fix that. The conventions now live in their own repository, and their status is "Development", so names can still change [25]. What I used in the lab, from the docs at the time of writing:

• Span names such as chat {model} for a model call, invoke_agent {agent name}, execute_tool {tool name}, and invoke_workflow {workflow name}.
• Attributes such as gen_ai.operation.name, gen_ai.provider.name, gen_ai.request.model, gen_ai.usage.input_tokens, gen_ai.usage.output_tokens, gen_ai.response.finish_reasons, gen_ai.tool.name and error.type.
• Prompt and tool-argument text is opt-in. Attributes such as gen_ai.input.messages can hold private data, so the conventions do not turn them on by default. My tracer follows this: content is off unless you ask [25].

The conventions have no attribute for cost. I used app.cost_usd, which is my own name, and a made-up price table. A real system computes cost from token counts and the current price list, and it must handle cached tokens and other price tiers, which my lab ignores. HarnessTax did something similar by fixing one price list for all runs [3].

The tools

Four products appear in the lecture. I describe each from its own docs and have not run them here.

Tool	What its docs say
Langfuse [26]	Open-source LLM observability and application tracing, with prompt management and evaluation features.
Arize Phoenix [27]	Traces, evaluation, and prompt iteration. Built on OpenTelemetry, with OpenInference instrumentation.
LangSmith [28]	Observability for LLM apps, from single traces to production metrics. Works with many frameworks, not only LangChain's.
MLflow Tracing [29]	Described as fully OpenTelemetry-compatible. Captures inputs, outputs and metadata of each intermediate step.

The common thread is OpenTelemetry. Phoenix and MLflow say so directly. If your spans follow the conventions, you can change the viewer without changing your code. This is one reason to use the shared names even in a toy.

What to look for in a trace

• Steps. How many model calls did one task take? Loops show up as repeated spans.
• Where the tokens went. A large first call may be a long system prompt or large tool schemas, as in HarnessTax [3].
• Where the time went. One slow tool can dominate.
• The first error. The failing span is often not the first wrong step. Read what the model saw just before.
• The exact prompt. Hamel Husain's post on this theme argues that you should look at the prompt that actually goes to the model, since libraries may rewrite it [38]. A trace with content turned on does this, and a proxy does too.

2.10 Best practices

The lecture ends with a short list. I give it here with the reasons, and with pointers to where the lab shows each one. None of these is a law. They are habits that people who ship agents tend to report.

1. Keep it simple. Start with the simplest thing that could work, often one model call with good context. Add a loop, a graph or a second agent only when a test shows the need [30]. In the lab: part 3's agent is one short loop.
2. Look at what the model sees. A framework that builds prompts for you can hide a bad prompt. Print the final prompt. Anthropic warns that extra layers can obscure the underlying prompts and responses, making them harder to debug [30]. In the lab: render_prompt returns a plain string.
3. Make failures observable. Every model call, tool call and retry should leave a span with tokens, time and the error type (section 2.9). Without it you cannot answer "why did that run cost $4?".
4. Make failures recoverable. Save progress at step boundaries. Make steps that change the world idempotent. Decide what a retry means before you add one (section 2.8). The lecture points to Shepherd on reversible traces [19]. In the lab: parts 2 and 4.
5. Put a person where the risk is. An interrupt before an irreversible step is cheap compared with a wrong email. Remember that the node restarts on resume, so keep side effects after the interrupt (section 2.4). In the lab: part 2.
6. Run evals, and make them yours. A generic benchmark tells you about the benchmark. A set of cases from your own domain, with a metric you chose, tells you about your product (section 2.11) [37].
7. Measure cost and variance, not only accuracy. Report cost next to success. Run each case more than once. HarnessTax and Kapoor et al. both make this case [3] [32].
8. Own your control flow. Whatever you use, be able to say which code decides the next step. The 12-factor agents list puts this as one of its principles [31].
9. Pin and read your dependencies. These tools change fast. For example, the OpenTelemetry GenAI conventions I cite moved to a new repository and are still marked Development, and the Pi repository changed owner. Pin versions, and read the changelog when you upgrade. The lecture's notebook, for example, lists minimum versions for each library [42].

2.11 Evals: the pillar that is easy to skip

The notes and required readings spend most time on the system pillar, because that is where frameworks live. The third pillar decides whether any of it worked. Hamel Husain's post "Your AI Product Needs Evals" gives a three-level picture drawn from his work with a real-estate assistant. Level 1 is unit tests, which are assertions you can run on every change. Level 2 is human and model evaluation, which needs logged traces. Level 3 is A/B testing. He notes that cost rises with level, which sets how often you run each, and that unlike classic unit tests you do not need a 100% pass rate, because the pass rate is a product decision [37].

Kind of eval	Example	Watch out for
Exact or rule-based	The category equals the label. The output parses. The file diff matches.	Right answer in different words. Too strict a rule.
State-based	After the run, the database or repository is in the right state.	Needs a resettable test world.
Model as judge	A second model scores helpfulness against a rubric.	The judge has its own bias. Check it against human labels on a sample.
Human review	A person reads 20 traces and sorts the failures.	Slow, but nothing else tells you what you do not yet know to test.
Cost and latency	Dollars and seconds per task, with a percentile.	Price lists change. Fix one when you compare.

Three habits that make evals worth the effort

• Grow the set from failures. Each time you see a bad trace, add a case. Husain describes continuously updating unit tests based on new failures [37].
• Keep a held-out set. If an optimizer or you tune on the dev set, its score is optimistic. Keep some cases that nobody tunes on [32].
• Show the uncertainty. With 30 tasks, a one-task change is 3.3 points (my own arithmetic). The lab's bootstrap_ci turns a list of per-task scores into an interval. For 10 tasks with 8 successes, the interval runs from 0.50 to 1.00 in the lab's demo. That is not a typo. It is what a small set can tell you.

One more reference point. METR measures how long a task, in human expert time, an AI agent can complete with a given reliability, and its page defines the 50% time horizon as the task length at which an agent is predicted to succeed half the time [39]. The lecture cites that line of work. METR's page says it is no longer actively updated, so read it as a definition, not as a current ranking.

2.12 Limits, and from frameworks to a project

What these tools do not do

• They do not fix the model. A graph, a signature or a trace does not make a wrong answer right. They make the program around the model easier to change and to check.
• They move complexity, not remove it. A framework can replace twenty lines you understand with a hidden layer you do not [30].
• Optimizers need a good metric. If the metric is weak, the optimizer finds the weak spots in it. This is Goodhart's problem in a new place.
• Durability is not correctness. A replayed run repeats the same mistakes, faithfully.
• Most numbers here are narrow. The DSPy gains are on a few tasks. HarnessTax is two benchmarks. My lab numbers are from mocks.

From frameworks to a course project

This is a short, general guide, not the official project brief. For rules, deadlines and grading, use the course site [40].

1. Write the eval before the framework. Twenty cases and a metric. Run them against the simplest version you can write.
2. Add one piece at a time. A graph if you need branches and pauses. An optimizer if prompts are the bottleneck and the metric is solid. Durable execution if runs are long. Check each against the eval.
3. Turn on tracing on day one. Read 20 traces by hand each week.
4. Record cost. Put cost per task in the same table as success.
5. Keep a log of surprises. The honest part of a report is what the framework made harder.

2.13 Week 3b in one page

• Three pillars: data, system, evaluation. Frameworks mostly help the system. Evals tell you if a change helped [1] [37].
• Spectrum: model call, workflow (code picks the path), agent (model picks the path). More freedom costs more and fails in more ways [30].
• DSPy: signatures say what, modules say how, optimizers tune prompts for a metric. BootstrapFewShot keeps traces that pass the metric as demonstrations [1] [4] [5] [6].
• LangGraph: state with reducers, nodes, edges, conditional edges, super-steps, checkpoints by thread, interrupts. A node restarts on resume, so side effects before an interrupt must be idempotent [2] [10] [11] [12].
• Harnesses: mini-SWE-agent is bash only with a linear history. OpenHands has an event stream and a sandbox. Pi has four tools and no safety net by default [13] [15] [16] [17].
• HarnessTax: the same model, similar success, up to 5 times the cost across harnesses. Two benchmarks, 30 tasks each [3].
• Temporal: workflows are deterministic and replayed from an event history. Activities do the outside work and are retried, so make them idempotent [20] [21] [22] [23] [24].
• Observability: traces are trees of spans. OpenTelemetry GenAI names (in Development) let tools share traces. Prompt text is opt-in. Cost is your own attribute [25].
• Practice: keep it simple, read the prompt, trace everything, make steps recoverable, put a person at the risk, run your own evals, and report cost and variance [3] [30] [37].

3. Reading guide

Three readings are required. The rest are optional. For each required one you get: why it matters, the key ideas, and questions to think about as you read. Read the questions first. Take short notes as you go. The LangGraph overview and the HarnessTax post are web pages, not papers. Read them like documentation and like a lab report: note what is claimed, what was measured, and what is only a design choice.

Optional readings (added)

These are sources I added. Each is in the references.

More, if you want them

• Other frameworks: the OpenAI Agents SDK [33], CrewAI [34], LlamaIndex Workflows [35] and Pydantic AI [36].
• Time horizons of agents: METR [39].
• The course's own materials: the schedule [40], the lecture slides [41] and the code repository [42].

4. Hands-on lab: five small frameworks with a mock model

Time: about 3 hours. Goal: build small, testable versions of the five main ideas of this week, and see where each one breaks. You do not need an API key or any package. The "language model" is a mock: a scripted list of replies, or a few lines of Python that act like a model that copies the closest example it was shown.

What you will learn

• How a signature string becomes a prompt, and how a reply is parsed back into typed fields.
• What a bootstrapping optimizer keeps, and why it cannot teach what the zero-shot program never gets right.
• How state with reducers, conditional edges, super-steps and checkpoints fit together, and why a node restarts after an interrupt.
• What a bash-only loop with a linear history looks like, what an allowlist refuses, and how a stuck detector stops a loop.
• How a workflow is replayed from an event history, and where "at least once" turns into a duplicate unless you pass a key.
• How spans, token counts and costs are collected, exported as JSON Lines and read back as a tree.
• How a bootstrap interval and a Pareto front change how you read a comparison of two setups.

Setup

1. You need a recent Python 3. There is nothing to install. The code uses only the standard library. I ran it on Python 3.13 only.
2. Save the code below as frameworks_toolkit.py. Or download it: frameworks_toolkit.py.
3. Save the tests as test_frameworks_toolkit.py in the same folder. Or download: test_frameworks_toolkit.py. The test code is listed after the exercises.
4. Run python frameworks_toolkit.py for the six-section demo, and python test_frameworks_toolkit.py for the tests. To read a saved trace, run python frameworks_toolkit.py view spans.jsonl. The demo and the tests are deterministic. The tests include a real child process, a real timeout and a FIFO, so they take a second or two on a normal machine.

The code: frameworks_toolkit.py

The code has five parts, matching Figure 9, and a demo at the end. The comments name each choice that I made and the sources did not. The file is long. Read the part you are working on, with the matching section of the notes open.

"""frameworks_toolkit.py - a tiny toolkit for Week 3b (Frameworks & Orchestration).

Standard library only. No API key. Every "language model" here is a mock.
Everything is made up for teaching: the tickets, the tasks, the prices and the slip rates.

Parts, in order:
  1. A DSPy-like mini library: Signature, Predict, ChainOfThought, BestOfN, a metric,
     evaluate, and the optimizers LabeledFewShot, BootstrapFewShot and random search.
  2. A LangGraph-like StateGraph: typed state with reducers, nodes, conditional edges,
     checkpoints, and interrupt / resume for a human in the loop.
  3. A mini-SWE-agent-style harness: one bash action per step, a temp directory, an allowlist,
     a linear history, an event log, a stuck detector, and a tree of messages (Pi-style).
  4. A Temporal-style durable runtime: workflows, activities, an event history, retries,
     timers, signals, crash and deterministic replay.
  5. Tracing: spans with OpenTelemetry-GenAI-like attributes, token and cost accounting,
     JSONL export, a text trace viewer, a bootstrap confidence interval and a Pareto front.

Run `python frameworks_toolkit.py` for the demo. Tested on Python 3.13 only.
"""
import contextlib
import copy
import hashlib
import itertools
import json
import math
import operator
import os
import random
import re
import shlex
import subprocess
import tempfile
import threading
import time
from dataclasses import dataclass, field
from typing import (Annotated, Any, Callable, Dict, Iterable, List, Literal, Optional, Tuple,
                    get_args, get_origin, get_type_hints)

# =====================================================================================
# 1. A DSPy-like mini library
# =====================================================================================
_TYPES = {"str": str, "int": int, "float": float, "bool": bool}
_IDENT = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")


class SignatureError(ValueError):
    """The signature string is not valid."""


class ParseError(ValueError):
    """The model's reply could not be read as the output fields."""


@dataclass(frozen=True)
class Fld:
    name: str
    type: str = "str"


class Signature:
    """Inputs, outputs and instructions. 'question -> answer' is the smallest example."""

    def __init__(self, inputs: Tuple[Fld, ...], outputs: Tuple[Fld, ...], instructions: str = ""):
        self.inputs, self.outputs = tuple(inputs), tuple(outputs)
        names = [f.name for f in self.inputs + self.outputs]
        if not self.inputs or not self.outputs:
            raise SignatureError("a signature needs at least one input and one output")
        if len(set(names)) != len(names):
            raise SignatureError("field names must be unique: %r" % names)
        self.instructions = instructions or "Given the fields %s, produce the fields %s." % (
            ", ".join(f.name for f in self.inputs), ", ".join(f.name for f in self.outputs))

    @staticmethod
    def _side(text: str) -> Tuple[Fld, ...]:
        out = []
        for item in text.split(","):
            item = item.strip()
            name, _, typ = item.partition(":")
            name, typ = name.strip(), (typ.strip() or "str")
            if not _IDENT.match(name):
                raise SignatureError("bad field name %r" % name)
            if typ not in _TYPES:
                raise SignatureError("unknown type %r (use one of %s)" % (typ, sorted(_TYPES)))
            out.append(Fld(name, typ))
        return tuple(out)

    @classmethod
    def parse(cls, spec: str, instructions: str = "") -> "Signature":
        if spec.count("->") != 1:
            raise SignatureError("a signature needs exactly one '->': %r" % spec)
        left, right = spec.split("->")
        return cls(cls._side(left), cls._side(right), instructions)

    def prepend_output(self, name: str, type: str = "str") -> "Signature":
        """What ChainOfThought does: put a new output field first."""
        return Signature(self.inputs, (Fld(name, type),) + self.outputs, self.instructions)

    def __str__(self):
        side = lambda fs: ", ".join(f.name if f.type == "str" else "%s: %s" % (f.name, f.type) for f in fs)
        return "%s -> %s" % (side(self.inputs), side(self.outputs))


def _coerce(text: str, typ: str):
    text = text.strip()
    try:
        if typ == "int":
            return int(text)
        if typ == "float":
            return float(text)
        if typ == "bool":
            low = text.lower()
            if low in ("true", "yes"):
                return True
            if low in ("false", "no"):
                return False
            raise ValueError(text)
    except ValueError:
        raise ParseError("cannot read %r as %s" % (text, typ))
    return text


class Example:
    """One data point. with_inputs() says which keys are inputs; the rest are labels."""

    def __init__(self, **kw):
        self._d, self._inputs = dict(kw), ()

    def with_inputs(self, *keys) -> "Example":
        e = Example(**self._d)
        e._inputs = tuple(keys)
        return e

    def inputs(self) -> Dict[str, Any]:
        return {k: self._d[k] for k in self._inputs}

    def labels(self) -> Dict[str, Any]:
        return {k: v for k, v in self._d.items() if k not in self._inputs}

    def __getattr__(self, k):
        if k.startswith("_"):
            raise AttributeError(k)
        try:
            return self._d[k]
        except KeyError:
            raise AttributeError(k)

    def toDict(self):
        return dict(self._d)


class Prediction(Example):
    """The result of a module call. Fields are attributes."""


class LM:
    """The smallest LM interface: text in, text out. Subclasses count their calls."""

    def __init__(self):
        self.calls = 0
        self.history: List[str] = []

    def __call__(self, prompt: str) -> str:
        raise NotImplementedError


class _Settings:
    lm: Optional[LM] = None


settings = _Settings()


def configure(lm: Optional[LM] = None):
    settings.lm = lm


_trace_stack: List[list] = []   # single-threaded on purpose


def render_prompt(sig: Signature, demos: List[dict], inputs: Dict[str, Any]) -> str:
    """The text a Predict sends. The layout is my own; real DSPy adapters differ."""
    lines = ["INSTRUCTIONS: " + sig.instructions,
             "OUTPUT FIELDS: " + ", ".join(f.name for f in sig.outputs),
             'Reply with one line per output field, like "name: value".']
    for d in demos:
        lines.append("--- demo")
        lines += ["%s: %s" % (f.name, d[f.name]) for f in sig.inputs + sig.outputs if f.name in d]
    lines.append("--- input")
    lines += ["%s: %s" % (f.name, inputs[f.name]) for f in sig.inputs]
    return "\n".join(lines)


def parse_reply(sig: Signature, text: str) -> Dict[str, Any]:
    found = {}
    for line in text.splitlines():
        name, sep, value = line.partition(":")
        if sep and name.strip() in {f.name for f in sig.outputs} and name.strip() not in found:
            found[name.strip()] = value.strip()
    missing = [f.name for f in sig.outputs if f.name not in found]
    if missing:
        raise ParseError("reply is missing %s: %r" % (missing, text[:60]))
    return {f.name: _coerce(found[f.name], f.type) for f in sig.outputs}


class Module:
    """Base class. A program is a Module that holds Predict modules and plain Python."""

    def named_predictors(self, prefix: str = "") -> List[Tuple[str, "Predict"]]:
        out = []
        for k, v in sorted(vars(self).items()):
            if isinstance(v, Predict):
                out.append((prefix + k, v))
            elif isinstance(v, Module):
                out += v.named_predictors(prefix + k + ".")
        return out

    def set_lm(self, lm: Optional[LM]):
        for _, p in self.named_predictors():
            p.lm = lm

    def reset_demos(self):
        for _, p in self.named_predictors():
            p.demos = []

    def deepcopy(self):
        return copy.deepcopy(self)

    def __deepcopy__(self, memo):
        new = object.__new__(type(self))
        memo[id(self)] = new
        for k, v in vars(self).items():
            setattr(new, k, copy.deepcopy(v, memo))
        return new

    def __call__(self, **kw):
        return self.forward(**kw)


class Predict(Module):
    """One LM call that implements a signature. Holds demos (few-shot examples)."""

    def __init__(self, signature, lm: Optional[LM] = None):
        self.signature = Signature.parse(signature) if isinstance(signature, str) else signature
        self.lm, self.demos = lm, []

    def __deepcopy__(self, memo):       # share the LM, copy the demos
        new = object.__new__(Predict)
        memo[id(self)] = new
        new.signature, new.lm, new.demos = self.signature, self.lm, copy.deepcopy(self.demos, memo)
        return new

    def named_predictors(self, prefix=""):
        return [(prefix.rstrip(".") or "self", self)]

    def forward(self, **kw) -> Prediction:
        names = [f.name for f in self.signature.inputs]
        if sorted(kw) != sorted(names):
            raise TypeError("expected inputs %s, got %s" % (names, sorted(kw)))
        lm = self.lm or settings.lm
        if lm is None:
            raise RuntimeError("no LM: call configure(lm=...) or set_lm(...)")
        reply = lm(render_prompt(self.signature, self.demos, kw))
        out = parse_reply(self.signature, reply)
        if _trace_stack:
            _trace_stack[-1].append((self, dict(kw), dict(out)))
        return Prediction(**kw, **out)


class ChainOfThought(Module):
    """Predict with a 'reasoning' output field put first. The mock LM gets no benefit from it."""

    def __init__(self, signature, lm: Optional[LM] = None):
        sig = Signature.parse(signature) if isinstance(signature, str) else signature
        self.predict = Predict(sig.prepend_output("reasoning"), lm)

    def forward(self, **kw) -> Prediction:
        return self.predict(**kw)


class BestOfN(Module):
    """Run a module up to n times and keep the best reward. Stop early at the threshold."""

    def __init__(self, module: Module, n: int, reward_fn: Callable, threshold: float = 1.0):
        if n < 1:
            raise ValueError("n must be at least 1")
        self.module, self.n, self.reward_fn, self.threshold = module, n, reward_fn, threshold

    def forward(self, **kw) -> Prediction:
        best, best_r, last_err = None, -math.inf, None
        for _ in range(self.n):
            try:
                pred = self.module(**kw)
            except ParseError as e:
                last_err = e
                continue
            r = self.reward_fn(kw, pred)
            if r > best_r:
                best, best_r = pred, r
            if best_r >= self.threshold:
                break
        if best is None:
            raise last_err
        return best


# ---- made-up data: support tickets -----------------------------------------------
_TICKETS = {
    "billing": [("I was charged twice for my {plan} plan this month", "high"),
                ("please refund invoice {num}, the amount is wrong", "low"),
                ("my card was billed but I cancelled the {plan} plan", "high"),
                ("why is there an extra charge on invoice {num}", "low")],
    "shipping": [("my package {num} has not arrived and tracking is stuck", "high"),
                 ("the delivery of order {num} is late", "low"),
                 ("tracking says delivered but the package is missing", "high"),
                 ("can I change the delivery address for order {num}", "low")],
    "login": [("I cannot log in, the password reset email never came", "high"),
              ("my account is locked after too many login tries", "high"),
              ("the two factor code fails when I log in to the {plan} account", "high"),
              ("reset my password, I forgot it", "low")],
    "feedback": [("please add dark mode to the app", "low"),
                 ("the new {plan} dashboard is great, thanks", "low"),
                 ("feature idea: export reports as a PDF", "low"),
                 ("I suggest a shortcut key for search", "low")],
}
CATEGORIES = sorted(_TICKETS)


def make_tickets(n: int, seed: int = 0) -> List[Example]:
    """Made-up tickets. Train and dev share templates, so dev scores are optimistic."""
    rng = random.Random(seed)
    out = []
    for i in range(n):
        cat = CATEGORIES[i % len(CATEGORIES)]
        text, urgency = rng.choice(_TICKETS[cat])
        text = text.format(plan=rng.choice(["basic", "pro", "team"]), num=rng.choice(["4821", "1177", "9034", "3350"]))
        out.append(Example(ticket=text, category=cat, urgency=urgency).with_inputs("ticket"))
    rng.shuffle(out)
    return out


_STOP = set("a an the i my to of for is it in on at and or but not be was can this that with as me we you your are".split())


def _words(text: str) -> set:
    return {w for w in re.findall(r"[a-z0-9]+", text.lower()) if w not in _STOP}


def overlap(a: str, b: str) -> float:
    """Jaccard overlap of word sets, ignoring stop words."""
    wa, wb = _words(a), _words(b)
    return len(wa & wb) / len(wa | wb) if wa | wb else 0.0


class ImitatorLM(LM):
    """A mock LM that imitates the demos in its prompt.

    With no demos it guesses each field from `priors`, and with probability `p_format_slip`
    it forgets the "name: value" format. With demos it never slips on format. It copies the
    output of the demo whose inputs overlap most with the new input, if the overlap is at
    least `min_overlap`. These rules are my own simplification of "demos help a small model".
    """

    def __init__(self, priors: Dict[str, List[str]], seed: int = 0, p_format_slip: float = 0.3,
                 min_overlap: float = 0.2, model: str = "mock-imitator"):
        super().__init__()
        self.priors, self.rng, self.p_slip, self.min_overlap, self.model = priors, random.Random(seed), p_format_slip, min_overlap, model

    def __call__(self, prompt: str) -> str:
        self.calls += 1
        self.history.append(prompt)
        head, _, tail = prompt.partition("--- input\n")
        fields = re.search(r"OUTPUT FIELDS: (.*)", head).group(1).split(", ")
        demos = []
        for block in head.split("--- demo\n")[1:]:
            demos.append(dict(l.split(": ", 1) for l in block.split("\n") if ": " in l and not l.startswith(("OUTPUT", "Reply", "INSTR"))))
        new = dict(l.split(": ", 1) for l in tail.split("\n") if ": " in l)
        if not demos and self.rng.random() < self.p_slip:
            return "I think the answer is probably %s." % self.rng.choice(self.priors.get(fields[-1], ["unknown"]))
        lines = []
        for f in fields:
            if f == "reasoning":
                lines.append("reasoning: I looked at the key words and picked the closest known case.")
                continue
            best, best_o = None, -1.0
            for d in demos:
                if f not in d:
                    continue
                o = overlap(" ".join(new.values()), " ".join(v for k, v in d.items() if k in new))
                if o > best_o:
                    best, best_o = d, o
            if best is not None and best_o >= self.min_overlap:
                lines.append("%s: %s" % (f, best[f]))
            else:
                lines.append("%s: %s" % (f, self.rng.choice(self.priors.get(f, ["unknown"]))))
        return "\n".join(lines)


def make_lm(seed: int = 0, p_format_slip: float = 0.3) -> ImitatorLM:
    return ImitatorLM({"category": CATEGORIES, "urgency": ["low", "high"]}, seed=seed, p_format_slip=p_format_slip)


class Classify(Module):
    """One stage: ticket -> category."""

    def __init__(self):
        self.predict = Predict("ticket -> category", None)

    def forward(self, ticket):
        return self.predict(ticket=ticket)


class Triage(Module):
    """Two stages: ticket -> category, then (ticket, category) -> urgency. Plain Python glue between."""

    def __init__(self):
        self.category = Predict("ticket -> category")
        self.urgency = Predict("ticket, category -> urgency")

    def forward(self, ticket):
        c = self.category(ticket=ticket).category
        u = self.urgency(ticket=ticket, category=c).urgency
        return Prediction(ticket=ticket, category=c, urgency=u)


def category_match(example, pred, trace=None) -> bool:
    return getattr(pred, "category", None) == example.category


def triage_match(example, pred, trace=None) -> bool:
    return pred.category == example.category and pred.urgency == example.urgency


@dataclass
class EvalResult:
    score: float
    n: int
    failures: int          # examples where the program raised
    lm_calls: int
    rows: List[Tuple[bool, str]]


def evaluate(program: Module, devset: List[Example], metric: Callable, lm: Optional[LM] = None) -> EvalResult:
    """Mean metric over a dev set. An exception counts as a miss, like DSPy's Evaluate."""
    lm = lm or settings.lm
    before = lm.calls if lm else 0
    hits, fails, rows = 0, 0, []
    for ex in devset:
        try:
            ok = bool(metric(ex, program(**ex.inputs())))
            rows.append((ok, ""))
        except Exception as e:      # ParseError and friends
            ok = False
            fails += 1
            rows.append((False, type(e).__name__))
        hits += ok
    return EvalResult(hits / len(devset) if devset else 0.0, len(devset), fails, (lm.calls - before) if lm else 0, rows)


def _label_demo(pred: Predict, ex: Example) -> Optional[dict]:
    d = dict(ex.toDict())
    need = [f.name for f in pred.signature.inputs + pred.signature.outputs]
    return {k: d[k] for k in need} if all(k in d for k in need) else None


class LabeledFewShot:
    """Put k labeled training examples into every predictor whose fields the labels cover."""

    def __init__(self, k: int = 4, seed: int = 0):
        self.k, self.seed = k, seed

    def compile(self, student: Module, trainset: List[Example]) -> Module:
        student = student.deepcopy()
        rng = random.Random(self.seed)
        pool = list(trainset)
        rng.shuffle(pool)
        for _, p in student.named_predictors():
            demos = [d for d in (_label_demo(p, e) for e in pool) if d]
            p.demos = demos[: self.k]
        return student


class BootstrapFewShot:
    """Run a teacher on training inputs. Keep traces whose final output passes the metric.

    The kept (input, output) pairs of every predictor become demos for that predictor.
    `max_rounds` retries the examples that failed. `max_labeled_demos` adds labeled examples
    after the bootstrapped ones. The teacher defaults to the student with no demos.
    """

    def __init__(self, metric: Callable, max_bootstrapped_demos: int = 4, max_labeled_demos: int = 0,
                 max_rounds: int = 1, seed: int = 0):
        self.metric, self.max_b, self.max_l, self.rounds, self.seed = metric, max_bootstrapped_demos, max_labeled_demos, max_rounds, seed

    def compile(self, student: Module, trainset: List[Example], teacher: Optional[Module] = None) -> Module:
        student = student.deepcopy()
        student.reset_demos()
        teacher = teacher.deepcopy() if teacher is not None else student.deepcopy()
        pool = list(trainset)
        random.Random(self.seed).shuffle(pool)
        t_preds = dict(teacher.named_predictors())
        kept: Dict[str, List[dict]] = {name: [] for name in t_preds}
        used, done = set(), 0
        for _ in range(self.rounds):
            for i, ex in enumerate(pool):
                if done >= self.max_b:
                    break
                if i in used:
                    continue
                _trace_stack.append([])
                try:
                    ok = bool(self.metric(ex, teacher(**ex.inputs())))
                except Exception:
                    ok = False
                finally:
                    trace = _trace_stack.pop()
                if ok and trace:
                    used.add(i)
                    done += 1
                    names = {id(p): n for n, p in t_preds.items()}
                    for p, inp, out in trace:
                        kept[names[id(p)]].append({**inp, **out})
        for name, p in student.named_predictors():
            p.demos = kept.get(name, [])[: self.max_b]
            if self.max_l:
                extra = [d for d in (_label_demo(p, pool[i]) for i in range(len(pool)) if i not in used) if d]
                p.demos += extra[: self.max_l]
        student._bootstrapped = done            # how many training examples produced a demo
        return student


class RandomSearch:
    """Try candidate programs, score each on valset, keep the best (earliest wins ties).

    Candidates: the program with no demos, labeled few-shot, then `num_candidates` bootstraps
    with different shuffles. This is a simplified take on DSPy's BootstrapFewShotWithRandomSearch.
    """

    def __init__(self, metric: Callable, num_candidates: int = 4, max_bootstrapped_demos: int = 4,
                 max_labeled_demos: int = 0, max_rounds: int = 2, seed: int = 0):
        self.metric, self.n, self.max_b, self.max_l, self.rounds, self.seed = metric, num_candidates, max_bootstrapped_demos, max_labeled_demos, max_rounds, seed
        self.scores: List[float] = []

    def compile(self, student: Module, trainset: List[Example], valset: Optional[List[Example]] = None) -> Module:
        valset = valset or trainset
        zero = student.deepcopy()
        zero.reset_demos()
        cands = [zero, LabeledFewShot(self.max_b, self.seed).compile(student, trainset)]
        for i in range(self.n):
            cands.append(BootstrapFewShot(self.metric, self.max_b, self.max_l, self.rounds, self.seed + i + 1).compile(student, trainset))
        self.scores = [evaluate(c, valset, self.metric).score for c in cands]
        return cands[max(range(len(cands)), key=lambda i: (self.scores[i], -i))]


# =====================================================================================
# 2. A LangGraph-like StateGraph
# =====================================================================================
START, END = "__start__", "__end__"


class GraphRecursionError(RuntimeError):
    """The graph ran more super-steps than the limit."""


class InvalidUpdateError(ValueError):
    """A node returned a bad update (unknown key, wrong type, or two writes to a key with no reducer)."""


class GraphInterrupt(Exception):
    """Raised inside a node by interrupt(). The graph catches it and saves a checkpoint."""

    def __init__(self, interrupt):
        super().__init__(interrupt)
        self.interrupt = interrupt


@dataclass
class Interrupt:
    value: Any
    node: str


@dataclass
class Command:
    """Pass Command(resume=value) to invoke() to continue a paused run."""
    resume: Any = None


class _ICtx:
    def __init__(self, node, values):
        self.node, self.values, self.counter = node, list(values), 0


_ictx: List[_ICtx] = []


def interrupt(value: Any) -> Any:
    """Pause the graph and show `value` to the caller. On resume, return the resume value.

    The node restarts from its first line on resume, so code before interrupt() runs again.
    """
    if not _ictx:
        raise RuntimeError("interrupt() can only be called inside a graph node")
    ctx = _ictx[-1]
    i, ctx.counter = ctx.counter, ctx.counter + 1
    if i < len(ctx.values):
        return ctx.values[i]
    raise GraphInterrupt(Interrupt(value, ctx.node))


def _empty(tp):
    origin = get_origin(tp) or tp
    return origin() if origin in (list, dict, set, str, int, float) else None


class InMemorySaver:
    """Checkpoints in RAM, per thread. Lost when the process ends."""

    def __init__(self):
        self.data: Dict[str, List[dict]] = {}

    def put(self, thread: str, ckpt: dict):
        self.data.setdefault(thread, []).append(copy.deepcopy(ckpt))

    def list(self, thread: str) -> List[dict]:
        return copy.deepcopy(self.data.get(thread, []))

    def latest(self, thread: str) -> Optional[dict]:
        items = self.data.get(thread)
        return copy.deepcopy(items[-1]) if items else None


class FileSaver(InMemorySaver):
    """Checkpoints appended to a JSON Lines file, so a new process can resume (state must be JSON)."""

    def __init__(self, path: str):
        super().__init__()
        self.path = path

    def put(self, thread: str, ckpt: dict):
        with open(self.path, "a", encoding="utf-8") as f:
            f.write(json.dumps({"thread": thread, "ckpt": ckpt}) + "\n")

    def list(self, thread: str) -> List[dict]:
        if not os.path.exists(self.path):
            return []
        with open(self.path, encoding="utf-8") as f:
            rows = [json.loads(l) for l in f if l.strip()]
        return [r["ckpt"] for r in rows if r["thread"] == thread]

    def latest(self, thread: str) -> Optional[dict]:
        items = self.list(thread)
        return items[-1] if items else None


@dataclass
class Snapshot:
    values: dict
    next: List[str]
    step: int
    status: str            # "running", "interrupted" or "done"


class StateGraph:
    """Build a graph of nodes that read and write one shared state.

    The schema is a TypedDict. A key typed Annotated[list, operator.add] has a reducer that
    merges writes. Keys with no reducer are replaced.
    """

    def __init__(self, schema):
        hints = get_type_hints(schema, include_extras=True)
        self.keys, self.reducers, self.types = list(hints), {}, {}
        for k, tp in hints.items():
            if get_origin(tp) is Annotated:
                base, *meta = get_args(tp)
                self.types[k] = base
                fns = [m for m in meta if callable(m)]
                if fns:
                    self.reducers[k] = fns[0]
            else:
                self.types[k] = tp
        self.nodes: Dict[str, Callable] = {}
        self.edges: List[Tuple[str, str]] = []
        self.cond: Dict[str, Tuple[Callable, Optional[dict]]] = {}

    def add_node(self, name: str, fn: Callable) -> "StateGraph":
        if name in (START, END) or name in self.nodes:
            raise ValueError("bad or duplicate node name %r" % name)
        self.nodes[name] = fn
        return self

    def _check(self, a, b=None):
        for n, allowed in ((a, {START}), (b, {END})):
            if n is not None and n not in self.nodes and n not in allowed:
                raise ValueError("unknown node %r" % n)

    def add_edge(self, src: str, dst: str) -> "StateGraph":
        self._check(src, dst)
        self.edges.append((src, dst))
        return self

    def add_conditional_edges(self, src: str, router: Callable, path_map: Optional[dict] = None) -> "StateGraph":
        self._check(src)
        if src in self.cond:
            raise ValueError("%r already has conditional edges" % src)
        self.cond[src] = (router, path_map)
        return self

    def compile(self, checkpointer=None, recursion_limit: int = 25) -> "CompiledGraph":
        if not any(a == START for a, _ in self.edges) and START not in self.cond:
            raise ValueError("the graph has no edge from START")
        for n in self.nodes:
            if not any(a == n for a, _ in self.edges) and n not in self.cond:
                raise ValueError("node %r has no outgoing edge" % n)
        return CompiledGraph(self, checkpointer, recursion_limit)


class CompiledGraph:
    def __init__(self, g: StateGraph, checkpointer, recursion_limit):
        self.g, self.saver, self.limit = g, checkpointer, recursion_limit
        self.events: List[Tuple[str, str]] = []     # ("run", node) / ("skip", node): for tests and demos

    # -- state updates ----------------------------------------------------------------
    def _apply(self, state: dict, writes: List[Tuple[str, dict]]) -> dict:
        state = dict(state)
        per_key: Dict[str, List[Any]] = {}
        for node, upd in writes:
            for k, v in upd.items():
                per_key.setdefault(k, []).append(v)
        for k, vals in per_key.items():
            red = self.g.reducers.get(k)
            if red is None and len(vals) > 1:
                raise InvalidUpdateError("key %r got %d writes in one step and has no reducer" % (k, len(vals)))
            for v in vals:
                if red is None:
                    state[k] = v
                else:
                    base = state[k] if k in state else _empty(self.g.types[k])
                    state[k] = v if base is None else red(base, v)
        return state

    def _check_update(self, node: str, upd) -> dict:
        if upd is None:
            return {}
        if not isinstance(upd, dict):
            raise InvalidUpdateError("node %r must return a dict or None, got %s" % (node, type(upd).__name__))
        bad = [k for k in upd if k not in self.g.keys]
        if bad:
            raise InvalidUpdateError("node %r wrote unknown keys %s" % (node, bad))
        return upd

    def _targets(self, ran: Iterable[str], state: dict) -> List[str]:
        out: List[str] = []
        for n in ran:
            dsts = [b for a, b in self.g.edges if a == n]
            if n in self.g.cond:
                router, pmap = self.g.cond[n]
                r = router(copy.deepcopy(state))
                for t in (r if isinstance(r, (list, tuple)) else [r]):
                    t = pmap[t] if pmap and t in pmap else t
                    if t != END and t not in self.g.nodes:
                        raise ValueError("router of %r returned unknown target %r" % (n, t))
                    dsts.append(t)
            for d in dsts:
                if d != END and d not in out:
                    out.append(d)
        return out

    # -- running ----------------------------------------------------------------------
    def _save(self, thread, ckpt):
        if self.saver and thread is not None:
            self.saver.put(thread, ckpt)

    def invoke(self, input: Any = None, config: Optional[dict] = None) -> dict:
        cfg = config or {}
        thread = (cfg.get("configurable") or {}).get("thread_id")
        limit = cfg.get("recursion_limit", self.limit)
        latest = self.saver.latest(thread) if (self.saver and thread is not None) else None
        if isinstance(input, Command):
            if not latest or latest["status"] != "interrupted":
                raise ValueError("nothing to resume on thread %r" % thread)
            ck = latest
            node = ck["interrupt"]["node"]
            ck["resume"].setdefault(node, []).append(input.resume)
            ck["status"], ck["interrupt"] = "running", None
        elif input is None:
            if not latest or latest["status"] == "done":
                raise ValueError("no unfinished run on thread %r" % thread)
            if latest["status"] == "interrupted":
                raise ValueError("thread %r is waiting: pass Command(resume=...)" % thread)
            ck = latest
        else:
            if latest and latest["status"] == "interrupted":
                raise ValueError("thread %r is waiting: pass Command(resume=...)" % thread)
            base = latest["state"] if latest else {}
            state = self._apply(base, [(START, self._check_update(START, dict(input)))])
            ck = {"step": latest["step"] if latest else 0, "state": state, "next": [], "writes": {},
                  "resume": {}, "status": "running", "interrupt": None}
            ck["next"] = self._targets([START], state)
            self._save(thread, ck)
        return self._loop(thread, ck, limit)

    def _loop(self, thread, ck, limit) -> dict:
        run_steps = 0
        while ck["next"]:
            if run_steps >= limit:
                raise GraphRecursionError("hit the limit of %d super-steps" % limit)
            writes: List[Tuple[str, dict]] = []
            for node in sorted(ck["next"]):
                if node in ck["writes"]:
                    writes.append((node, ck["writes"][node]))
                    self.events.append(("skip", node))
                    continue
                _ictx.append(_ICtx(node, ck["resume"].get(node, [])))
                try:
                    self.events.append(("run", node))
                    upd = self._check_update(node, self.g.nodes[node](copy.deepcopy(ck["state"])))
                except GraphInterrupt as gi:
                    ck["status"] = "interrupted"
                    ck["interrupt"] = {"node": node, "value": gi.interrupt.value}
                    self._save(thread, ck)
                    return {**ck["state"], "__interrupt__": [gi.interrupt]}
                except Exception:
                    self._save(thread, ck)      # keep the writes of nodes that already finished
                    raise
                finally:
                    _ictx.pop()
                ck["writes"][node] = upd
                writes.append((node, upd))
            state = self._apply(ck["state"], writes)
            ran = sorted(ck["next"])
            ck = {"step": ck["step"] + 1, "state": state, "next": self._targets(ran, state), "writes": {},
                  "resume": {}, "status": "running", "interrupt": None}
            run_steps += 1
            if not ck["next"]:
                ck["status"] = "done"
            self._save(thread, ck)
        ck["status"] = "done"
        return ck["state"]

    # -- inspection ---------------------------------------------------------------------
    def get_state(self, config: dict) -> Snapshot:
        ck = self.saver.latest(config["configurable"]["thread_id"])
        if ck is None:
            raise KeyError("no checkpoint for that thread")
        return Snapshot(ck["state"], ck["next"], ck["step"], ck["status"])

    def history(self, config: dict) -> List[dict]:
        return self.saver.list(config["configurable"]["thread_id"])

    def to_mermaid(self) -> str:
        lines = ["graph TD"]
        for a, b in self.g.edges:
            lines.append("  %s --> %s" % (a, b))
        for a, (router, pmap) in sorted(self.g.cond.items()):
            targets = list(pmap.values()) if pmap else []
            if not targets:
                lit = get_type_hints(router).get("return")
                targets = list(get_args(lit)) if lit is not None and get_origin(lit) is not None else []
            for t in targets:
                lines.append("  %s -.-> %s" % (a, t))
        return "\n".join(lines)


# =====================================================================================
# 3. A mini-SWE-agent-style harness
# =====================================================================================
DEFAULT_ALLOW = ("ls", "cat", "echo", "grep", "head", "tail", "wc", "mkdir", "touch", "cp", "mv",
                 "sort", "pwd", "printf", "diff")
DONE_MARK = "TASK_COMPLETE"      # my own marker for "I am finished"
_FENCE = re.compile(r"```bash\n(.*?)\n```", re.S)


class SandboxError(Exception):
    """A command was refused before it ran."""


@dataclass
class Observation:
    returncode: int
    output: str
    truncated: int = 0       # number of characters cut


def split_commands(cmd: str) -> List[List[str]]:
    """Split a command line into simple commands. Refuses things the toy allowlist cannot judge."""
    if "\n" in cmd or "`" in cmd or "$(" in cmd or "<<" in cmd or "<(" in cmd or ">(" in cmd:
        raise SandboxError("newlines, backticks, $( ), heredocs and process substitution are not allowed")
    lex = shlex.shlex(cmd, posix=True, punctuation_chars=True)
    lex.whitespace_split = True
    try:
        toks = list(lex)
    except ValueError as e:
        raise SandboxError("cannot parse command: %s" % e)
    cmds, cur = [], []
    for t in toks:
        if t in (";", "&&", "||", "|", "&"):
            if cur:
                cmds.append(cur)
            cur = []
        else:
            cur.append(t)
    if cur:
        cmds.append(cur)
    if not cmds:
        raise SandboxError("empty command")
    return cmds


class Sandbox:
    """Runs one bash command at a time in a directory, after an allowlist check.

    This is a TOY. It refuses obvious escapes, but a real harness uses a container or a VM.
    Each command is a fresh process, so `cd` does not carry over to the next command.
    """

    def __init__(self, workdir: str, allow=DEFAULT_ALLOW, timeout: float = 5.0, max_output: int = 2000):
        self.workdir, self.allow, self.timeout, self.max_output = os.path.realpath(workdir), tuple(allow), timeout, max_output

    def check(self, cmd: str):
        for argv in split_commands(cmd):
            toks = [t for t in argv if t not in (">", ">>", "<", "2>", "2>>", ">&", "2>&1")]
            if not toks:
                raise SandboxError("empty command")
            if toks[0] not in self.allow:
                raise SandboxError("%r is not on the allowlist %s" % (toks[0], list(self.allow)))
            for t in toks[1:]:
                pathlike = "/" in t or t.startswith(".")
                if (t.startswith("/") or t.startswith("~") or ".." in t.split("/") or "$" in t or "`" in t
                        or (pathlike and any(c in t for c in "[{?"))):
                    raise SandboxError("argument %r looks like a path or expansion outside the work directory" % t)

    def run(self, cmd: str) -> Observation:
        self.check(cmd)
        p = subprocess.Popen(["bash", "-c", cmd], cwd=self.workdir, env={"PATH": "/usr/bin:/bin", "LC_ALL": "C"},
                             stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, start_new_session=True)
        try:
            out, _ = p.communicate(timeout=self.timeout)
            rc = p.returncode
        except subprocess.TimeoutExpired:
            try:
                os.killpg(p.pid, 9)
            except ProcessLookupError:
                pass
            out, _ = p.communicate()
            return Observation(124, "timed out after %ss" % self.timeout)
        cut = max(0, len(out) - self.max_output)
        return Observation(rc, out[: self.max_output] + ("\n[truncated %d characters]" % cut if cut else ""), cut)


@dataclass
class Event:
    id: int
    kind: str          # system, user, assistant, observation, format_error, refusal
    text: str


_ROLE = {"system": "system", "user": "user", "assistant": "assistant",
         "observation": "user", "format_error": "user", "refusal": "user"}


def rebuild_messages(events: List[Event]) -> List[dict]:
    """The model's messages are rebuilt from the event log on every step (an OpenHands-style idea)."""
    return [{"role": _ROLE[e.kind], "content": e.text} for e in events]


def events_to_jsonl(events: List[Event]) -> str:
    return "\n".join(json.dumps({"id": e.id, "kind": e.kind, "text": e.text}) for e in events)


def events_from_jsonl(text: str) -> List[Event]:
    return [Event(**json.loads(l)) for l in text.splitlines() if l.strip()]


@dataclass
class RunResult:
    status: str          # submitted, step_limit, stuck, model_error
    steps: int
    events: List[Event]


class ScriptedModel:
    """A mock model that replays a list of replies. It keeps every message list it was shown."""

    def __init__(self, replies: List[str]):
        self.replies, self.seen = list(replies), []

    def __call__(self, messages: List[dict]) -> str:
        self.seen.append(copy.deepcopy(messages))
        if not self.replies:
            raise RuntimeError("script ended")
        return self.replies.pop(0)


class _NullSpan:
    def __init__(self):
        self.attributes: Dict[str, Any] = {}


@contextlib.contextmanager
def _span(tracer, name: str, **attrs):
    """A tracer span if there is a tracer, else a throwaway object (so callers need no ifs)."""
    if tracer is None:
        yield _NullSpan()
    else:
        with tracer.span(name, **attrs) as sp:
            yield sp


class BashAgent:
    """A bash-only agent loop with a linear history.

    Each step: show the history to the model, take its one ```bash block, run it, add the
    output. Ends on the done marker, the step limit, or `stuck` identical (action, output) pairs.
    """

    def __init__(self, model: Callable, sandbox: Sandbox, system: str = "You fix files. Reply with one ```bash block.",
                 step_limit: int = 10, stuck: int = 3, tracer=None):
        self.model, self.sandbox, self.system, self.step_limit, self.stuck = model, sandbox, system, step_limit, stuck
        self.tracer = tracer
        self.events: List[Event] = []

    def _add(self, kind: str, text: str):
        self.events.append(Event(len(self.events), kind, text))

    def run(self, task: str) -> RunResult:
        with _span(self.tracer, "invoke_agent bash-agent", **{"gen_ai.operation.name": "invoke_agent",
                                                            "gen_ai.agent.name": "bash-agent"}) as sp:
            res = self._run(task)
            sp.attributes["app.status"] = res.status
            sp.attributes["app.steps"] = res.steps
            return res

    def _run(self, task: str) -> RunResult:
        self.events = []
        self._add("system", self.system)
        self._add("user", task)
        recent: List[Tuple[str, str]] = []
        for step in range(1, self.step_limit + 1):
            try:
                msgs = rebuild_messages(self.events)
                with _span(self.tracer, "chat mock-model", **{"gen_ai.operation.name": "chat", "gen_ai.provider.name": "mock",
                                                              "gen_ai.request.model": "mock-model"}) as sp:
                    reply = self.model(msgs)
                    _usage(sp, "mock-model", json.dumps(msgs), reply)
            except Exception as e:
                self._add("format_error", "model error: %s" % e)
                return RunResult("model_error", step, self.events)
            self._add("assistant", reply)
            blocks = _FENCE.findall(reply)
            if len(blocks) != 1:
                self._add("format_error", "Reply with exactly one ```bash block. I found %d." % len(blocks))
                continue
            cmd = blocks[0].strip()
            if cmd == "echo " + DONE_MARK:
                return RunResult("submitted", step, self.events)
            try:
                with _span(self.tracer, "execute_tool bash", **{"gen_ai.operation.name": "execute_tool",
                                                                "gen_ai.tool.name": "bash"}) as sp:
                    try:
                        obs = self.sandbox.run(cmd)
                    except SandboxError:
                        sp.attributes["app.refused"] = True
                        raise
                text = "exit %d\n%s" % (obs.returncode, obs.output)
                self._add("observation", text)
            except SandboxError as e:
                text = "refused: %s" % e
                self._add("refusal", text)
            recent.append((cmd, text))
            if len(recent) >= self.stuck and len(set(recent[-self.stuck:])) == 1:
                return RunResult("stuck", step, self.events)
        return RunResult("step_limit", self.step_limit, self.events)


class SessionTree:
    """History as a tree (a Pi-style idea): go back to an earlier message and branch from there."""

    def __init__(self, system: str = "system"):
        self.nodes: Dict[int, dict] = {0: {"parent": None, "role": "system", "content": system}}
        self.head = 0

    def add(self, role: str, content: str) -> int:
        nid = len(self.nodes)
        self.nodes[nid] = {"parent": self.head, "role": role, "content": content}
        self.head = nid
        return nid

    def branch_from(self, node_id: int):
        if node_id not in self.nodes:
            raise KeyError(node_id)
        self.head = node_id

    def path(self, node_id: Optional[int] = None) -> List[dict]:
        nid, out = self.head if node_id is None else node_id, []
        while nid is not None:
            n = self.nodes[nid]
            out.append({"role": n["role"], "content": n["content"]})
            nid = n["parent"]
        return out[::-1]

    def leaves(self) -> List[int]:
        parents = {n["parent"] for n in self.nodes.values()}
        return sorted(i for i in self.nodes if i not in parents)


class FileTools:
    """Four tools in a directory: read, write, edit, bash (the surface Pi's README lists).

    `edit` replaces exactly one occurrence, so an ambiguous edit fails loudly.
    """

    def __init__(self, workdir: str, sandbox: Optional[Sandbox] = None):
        self.root = os.path.realpath(workdir)
        self.sandbox = sandbox or Sandbox(workdir)

    def _path(self, rel: str) -> str:
        p = os.path.realpath(os.path.join(self.root, rel))
        if os.path.commonpath([p, self.root]) != self.root:
            raise SandboxError("path %r is outside the work directory" % rel)
        return p

    def read(self, path: str) -> str:
        with open(self._path(path), encoding="utf-8") as f:
            return f.read()

    def write(self, path: str, text: str) -> str:
        p = self._path(path)
        os.makedirs(os.path.dirname(p), exist_ok=True)
        with open(p, "w", encoding="utf-8") as f:
            f.write(text)
        return "wrote %d characters" % len(text)

    def edit(self, path: str, old: str, new: str) -> str:
        text = self.read(path)
        n = text.count(old)
        if n != 1:
            raise ValueError("edit needs exactly one match for %r, found %d" % (old, n))
        self.write(path, text.replace(old, new))
        return "edited"

    def bash(self, cmd: str) -> Observation:
        return self.sandbox.run(cmd)

    def call(self, name: str, **kw):
        if name not in ("read", "write", "edit", "bash"):
            raise KeyError("no tool named %r" % name)
        return getattr(self, name)(**kw)


# =====================================================================================
# 4. A Temporal-style durable runtime
# =====================================================================================
class SimulatedCrash(Exception):
    """The worker 'died' at a chosen point. Used to test replay."""


class NonDeterminismError(RuntimeError):
    """On replay the workflow code did something different from its recorded history."""


class WorkflowSuspended(Exception):
    """The workflow is waiting for a timer or a signal."""


class NonRetryable(Exception):
    """An activity raises this to say: do not retry."""


class ActivityError(Exception):
    """An activity failed for good (out of attempts, or not retryable)."""


@dataclass
class RetryPolicy:
    """Backoff between attempts. Temporal's defaults are 1 s, x2, a cap of 100 x the first
    interval, and unlimited attempts [from its docs]. My default of 3 attempts is my own."""
    initial_interval: float = 1.0
    backoff: float = 2.0
    max_interval: float = 100.0
    max_attempts: int = 3
    non_retryable: Tuple[type, ...] = ()

    def delay(self, failed_attempt: int) -> float:
        return min(self.initial_interval * self.backoff ** (failed_attempt - 1), self.max_interval)


class HistoryStore:
    """An append-only event history per workflow. Optionally mirrored to a JSON Lines file."""

    def __init__(self, path: Optional[str] = None):
        self.path, self.events = path, {}
        if path and os.path.exists(path):
            with open(path, encoding="utf-8") as f:
                for line in f:
                    if line.strip():
                        row = json.loads(line)
                        self.events.setdefault(row["wf"], []).append(row["event"])

    def append(self, wf: str, event: dict) -> dict:
        evs = self.events.setdefault(wf, [])
        event = dict(event, id=len(evs) + 1)
        evs.append(event)
        if self.path:
            with open(self.path, "a", encoding="utf-8") as f:
                f.write(json.dumps({"wf": wf, "event": event}) + "\n")
        return event

    def for_workflow(self, wf: str) -> List[dict]:
        return list(self.events.get(wf, []))


_activity_local = threading.local()


def activity_info() -> dict:
    """Inside an activity: {'key': a stable idempotency key, 'attempt': 1-based attempt number}."""
    return dict(getattr(_activity_local, "info", {}))


@dataclass
class RunStatus:
    status: str                # completed, failed, suspended
    result: Any = None
    detail: str = ""


def _jsonable(x):
    return json.loads(json.dumps(x))


def _hash(args) -> str:
    return hashlib.sha1(json.dumps(args, sort_keys=True).encode()).hexdigest()[:10]


class Runtime:
    """Runs workflows by REPLAY. The workflow function is plain Python that calls ctx.activity,
    ctx.sleep, ctx.wait_for_signal and ctx.side_effect. Each call gets a sequence number. If the
    history already has the answer for that number, it is returned and nothing runs again.
    Time here is virtual: sleeping costs nothing, so tests are fast."""

    def __init__(self, store: Optional[HistoryStore] = None):
        self.store = store or HistoryStore()
        self.activities: Dict[str, Tuple[Callable, RetryPolicy]] = {}
        self.workflows: Dict[str, Callable] = {}
        self.now = 0.0
        self.crash_at: Optional[Tuple[int, str]] = None      # (event id, "before" | "after")
        self.executions: List[Tuple[str, int]] = []           # every real activity attempt: (name, attempt)

    def register_activity(self, name: str, fn: Callable, retry: Optional[RetryPolicy] = None):
        self.activities[name] = (fn, retry or RetryPolicy())

    def register_workflow(self, name: str, fn: Callable):
        self.workflows[name] = fn

    def _record(self, wf: str, type_: str, **data) -> dict:
        nxt = len(self.store.for_workflow(wf)) + 1
        if self.crash_at and self.crash_at == (nxt, "before"):
            self.crash_at = None
            raise SimulatedCrash("crash before event %d (%s)" % (nxt, type_))
        ev = self.store.append(wf, dict(type=type_, t=self.now, **data))
        if self.crash_at and self.crash_at == (nxt, "after"):
            self.crash_at = None
            raise SimulatedCrash("crash after event %d (%s)" % (nxt, type_))
        return ev

    def start(self, wf: str, name: str, *args) -> RunStatus:
        if self.store.for_workflow(wf):
            raise ValueError("workflow id %r already exists" % wf)
        if name not in self.workflows:
            raise KeyError("no workflow named %r" % name)
        self._record(wf, "WorkflowStarted", name=name, args=_jsonable(list(args)))
        return self.run(wf)

    def signal(self, wf: str, name: str, payload: Any = None):
        self._record(wf, "SignalReceived", name=name, payload=_jsonable(payload))

    def advance(self, seconds: float):
        self.now += seconds

    def status(self, wf: str) -> str:
        evs = self.store.for_workflow(wf)
        if not evs:
            raise KeyError(wf)
        return {"WorkflowCompleted": "completed", "WorkflowFailed": "failed"}.get(evs[-1]["type"], "running")

    def run(self, wf: str) -> RunStatus:
        """Run or replay a workflow until it finishes, fails or must wait."""
        evs = self.store.for_workflow(wf)
        if not evs:
            raise KeyError(wf)
        last = evs[-1]
        if last["type"] == "WorkflowCompleted":
            return RunStatus("completed", last["result"])
        if last["type"] == "WorkflowFailed":
            return RunStatus("failed", None, last["error"])
        started = evs[0]
        ctx = Ctx(self, wf)
        try:
            result = self.workflows[started["name"]](ctx, *started["args"])
        except WorkflowSuspended as s:
            return RunStatus("suspended", None, str(s))
        except ActivityError as e:
            self._record(wf, "WorkflowFailed", error=str(e))
            return RunStatus("failed", None, str(e))
        self._record(wf, "WorkflowCompleted", result=_jsonable(result))
        return RunStatus("completed", _jsonable(result))


class Ctx:
    """What a workflow function sees. Every method here is a 'command' with a sequence number."""

    def __init__(self, rt: Runtime, wf: str):
        self.rt, self.wf, self.seq = rt, wf, -1

    def _events(self, seq: int) -> List[dict]:
        return [e for e in self.rt.store.for_workflow(self.wf) if e.get("seq") == seq]

    def _next(self) -> int:
        self.seq += 1
        return self.seq

    def activity(self, name: str, *args) -> Any:
        seq = self._next()
        fn, policy = self.rt.activities[name]
        evs = self._events(seq)
        sched = [e for e in evs if e["type"] == "ActivityScheduled"]
        if sched:
            if sched[0]["name"] != name or sched[0]["args_hash"] != _hash(_jsonable(list(args))):
                raise NonDeterminismError("command %d was %s%s in history but the code now calls %s%s" % (
                    seq, sched[0]["name"], "(...)", name, "(...)"))
        else:
            self.rt._record(self.wf, "ActivityScheduled", seq=seq, name=name, args_hash=_hash(_jsonable(list(args))))
        for e in evs:
            if e["type"] == "ActivityCompleted":
                return e["result"]
            if e["type"] == "ActivityFailed":
                raise ActivityError(e["error"])
        attempt = 1 + sum(1 for e in evs if e["type"] == "ActivityAttemptFailed")
        while True:
            _activity_local.info = {"key": "%s:%d" % (self.wf, seq), "attempt": attempt}
            self.rt.executions.append((name, attempt))
            try:
                result = _jsonable(fn(*args))
            except Exception as ex:
                final = isinstance(ex, NonRetryable) or isinstance(ex, policy.non_retryable) or attempt >= policy.max_attempts
                if final:
                    self.rt._record(self.wf, "ActivityFailed", seq=seq, error="%s: %s" % (type(ex).__name__, ex), attempts=attempt)
                    raise ActivityError("%s failed after %d attempt(s): %s" % (name, attempt, ex))
                d = policy.delay(attempt)
                self.rt._record(self.wf, "ActivityAttemptFailed", seq=seq, attempt=attempt, error=str(ex), retry_in=d)
                self.rt.advance(d)
                attempt += 1
                continue
            finally:
                _activity_local.info = {}
            self.rt._record(self.wf, "ActivityCompleted", seq=seq, result=result, attempts=attempt)
            return result

    def sleep(self, seconds: float):
        seq = self._next()
        evs = self._events(seq)
        if any(e["type"] == "TimerFired" for e in evs):
            return
        started = [e for e in evs if e["type"] == "TimerStarted"]
        started = started[0] if started else self.rt._record(self.wf, "TimerStarted", seq=seq, fire_at=self.rt.now + seconds)
        if self.rt.now >= started["fire_at"]:
            self.rt._record(self.wf, "TimerFired", seq=seq)
            return
        raise WorkflowSuspended("sleeping until t=%g" % started["fire_at"])

    def wait_for_signal(self, name: str) -> Any:
        seq = self._next()
        evs = self._events(seq)
        done = [e for e in evs if e["type"] == "SignalConsumed"]
        all_ev = self.rt.store.for_workflow(self.wf)
        if done:
            return next(e["payload"] for e in all_ev if e["id"] == done[0]["signal_id"])
        taken = {e["signal_id"] for e in all_ev if e["type"] == "SignalConsumed"}
        for e in all_ev:
            if e["type"] == "SignalReceived" and e["name"] == name and e["id"] not in taken:
                self.rt._record(self.wf, "SignalConsumed", seq=seq, signal_id=e["id"])
                return e["payload"]
        raise WorkflowSuspended("waiting for signal %r" % name)

    def side_effect(self, fn: Callable) -> Any:
        seq = self._next()
        evs = [e for e in self._events(seq) if e["type"] == "SideEffect"]
        if evs:
            return evs[0]["value"]
        value = _jsonable(fn())
        self.rt._record(self.wf, "SideEffect", seq=seq, value=value)
        return value

    def now(self) -> float:
        """The virtual time, recorded once, so a replay sees the same value."""
        return self.side_effect(lambda: self.rt.now)


def format_history(events: List[dict]) -> str:
    """One line per event, like a short event-history view."""
    lines = []
    for e in events:
        extra = {k: v for k, v in e.items() if k not in ("id", "type", "t")}
        lines.append("%2d  t=%-7g %-22s %s" % (e["id"], e["t"], e["type"], json.dumps(extra, sort_keys=True)[:70]))
    return "\n".join(lines)


# =====================================================================================
# 5. Tracing, cost, and a little statistics
# =====================================================================================
PRICES = {"mock-imitator": (0.15, 0.60), "mock-model": (0.15, 0.60)}   # made-up USD per 1M input / output tokens


def approx_tokens(text: str) -> int:
    """A rough estimate: characters / 4, rounded up. Real tokenizers differ."""
    return max(1, math.ceil(len(text) / 4))


def cost_usd(model: str, in_tokens: int, out_tokens: int, prices: Dict[str, Tuple[float, float]] = PRICES) -> float:
    pin, pout = prices[model]
    return in_tokens * pin / 1e6 + out_tokens * pout / 1e6


def _usage(sp, model: str, prompt: str, reply: str):
    i, o = approx_tokens(prompt), approx_tokens(reply)
    sp.attributes.update({"gen_ai.usage.input_tokens": i, "gen_ai.usage.output_tokens": o,
                          "gen_ai.response.finish_reasons": ["stop"], "app.cost_usd": cost_usd(model, i, o)})


@dataclass
class Span:
    trace_id: str
    span_id: str
    parent_id: Optional[str]
    name: str
    start: float
    end: float = 0.0
    attributes: Dict[str, Any] = field(default_factory=dict)
    status: str = "OK"
    events: List[dict] = field(default_factory=list)

    @property
    def duration(self) -> float:
        return self.end - self.start


class Tracer:
    """Collects spans. Parent and child come from `with` nesting. The clock can be injected."""

    def __init__(self, clock: Callable[[], float] = time.time, seed: int = 0, capture_content: bool = False):
        self.clock, self.rng, self.capture_content = clock, random.Random(seed), capture_content
        self.spans: List[Span] = []
        self._local = threading.local()

    def _stack(self) -> List[Span]:
        if not hasattr(self._local, "stack"):
            self._local.stack = []
        return self._local.stack

    @contextlib.contextmanager
    def span(self, name: str, **attributes):
        st = self._stack()
        parent = st[-1] if st else None
        sp = Span(parent.trace_id if parent else "%032x" % self.rng.getrandbits(128), "%016x" % self.rng.getrandbits(64),
                  parent.span_id if parent else None, name, self.clock(), attributes=dict(attributes))
        st.append(sp)
        try:
            yield sp
        except Exception as e:
            sp.status = "ERROR"
            sp.attributes["error.type"] = type(e).__name__
            sp.events.append({"name": "exception", "message": str(e)[:200]})
            raise
        finally:
            sp.end = self.clock()
            st.pop()
            self.spans.append(sp)

    def content(self, sp: Span, key: str, value: Any):
        """Record prompt or argument text. Off by default: it can hold private data (the spec makes it opt-in)."""
        if self.capture_content:
            sp.attributes[key] = value


class TracedLM:
    """Wrap any LM so each call becomes a span named 'chat <model>' with token and cost attributes."""

    def __init__(self, lm, tracer: Tracer, model: str = "mock-imitator"):
        self.lm, self.tracer, self.model = lm, tracer, model

    @property
    def calls(self) -> int:
        return self.lm.calls

    def __call__(self, prompt: str) -> str:
        with self.tracer.span("chat " + self.model, **{"gen_ai.operation.name": "chat", "gen_ai.provider.name": "mock",
                                                     "gen_ai.request.model": self.model}) as sp:
            self.tracer.content(sp, "gen_ai.input.messages", prompt)
            reply = self.lm(prompt)
            _usage(sp, self.model, prompt, reply)
            return reply


def traced_tool(tracer: Tracer, name: str, fn: Callable, *args, call_id: str = "", **kw):
    """Run a tool inside an 'execute_tool <name>' span."""
    with tracer.span("execute_tool " + name, **{"gen_ai.operation.name": "execute_tool", "gen_ai.tool.name": name,
                                                "gen_ai.tool.call.id": call_id}) as sp:
        tracer.content(sp, "gen_ai.tool.call.arguments", {"args": list(args), "kw": kw})
        return fn(*args, **kw)


def traced_node(tracer: Tracer, name: str, fn: Callable) -> Callable:
    """Wrap a graph node. The span name 'node <name>' is my own; the GenAI spec has no graph-node span."""
    def wrapper(state):
        with tracer.span("node " + name, **{"graph.node": name}):
            return fn(state)
    return wrapper


def export_jsonl(spans: List[Span]) -> str:
    """One JSON object per line. Field names follow OTLP loosely, but this is not OTLP."""
    rows = []
    for s in spans:
        rows.append(json.dumps({"trace_id": s.trace_id, "span_id": s.span_id, "parent_span_id": s.parent_id, "name": s.name,
                                "start_time_unix_nano": int(s.start * 1e9), "end_time_unix_nano": int(s.end * 1e9),
                                "attributes": s.attributes, "status": {"code": s.status}, "events": s.events}, sort_keys=True))
    return "\n".join(rows)


def load_jsonl(text: str) -> List[Span]:
    out = []
    for line in text.splitlines():
        if line.strip():
            d = json.loads(line)
            out.append(Span(d["trace_id"], d["span_id"], d["parent_span_id"], d["name"], d["start_time_unix_nano"] / 1e9,
                            d["end_time_unix_nano"] / 1e9, d["attributes"], d["status"]["code"], d["events"]))
    return out


def summarize(spans: List[Span]) -> dict:
    by_op: Dict[str, dict] = {}
    for s in spans:
        op = s.attributes.get("gen_ai.operation.name", "other")
        row = by_op.setdefault(op, {"count": 0, "seconds": 0.0, "errors": 0})
        row["count"] += 1
        row["seconds"] += s.duration
        row["errors"] += s.status == "ERROR"
    return {"spans": len(spans), "errors": sum(s.status == "ERROR" for s in spans),
            "input_tokens": sum(s.attributes.get("gen_ai.usage.input_tokens", 0) for s in spans),
            "output_tokens": sum(s.attributes.get("gen_ai.usage.output_tokens", 0) for s in spans),
            "cost_usd": sum(s.attributes.get("app.cost_usd", 0.0) for s in spans),
            "by_operation": by_op}


def render_trace(spans: List[Span]) -> str:
    """A text tree: name, duration in ms, tokens, cost, and errors."""
    kids: Dict[Optional[str], List[Span]] = {}
    ids = {s.span_id for s in spans}
    for s in sorted(spans, key=lambda s: (s.start, s.end)):
        kids.setdefault(s.parent_id if s.parent_id in ids else None, []).append(s)
    lines: List[str] = []

    def walk(parent, depth):
        for s in kids.get(parent, []):
            bits = ["%.1f ms" % (s.duration * 1000)]
            a = s.attributes
            if "gen_ai.usage.input_tokens" in a:
                bits.append("tok %d/%d" % (a["gen_ai.usage.input_tokens"], a["gen_ai.usage.output_tokens"]))
            if "app.cost_usd" in a:
                bits.append("$%.6f" % a["app.cost_usd"])
            if s.status == "ERROR":
                bits.append("ERROR " + str(a.get("error.type", "")))
            if a.get("app.refused"):
                bits.append("refused")
            lines.append("%s%s  [%s]" % ("  " * depth, s.name, ", ".join(bits)))
            walk(s.span_id, depth + 1)

    walk(None, 0)
    return "\n".join(lines)


def bootstrap_ci(values: List[float], n_resamples: int = 1000, alpha: float = 0.05, seed: int = 0) -> Tuple[float, float, float]:
    """Mean and a percentile bootstrap interval. Resample with replacement, take the mean each time."""
    if not values:
        raise ValueError("no values")
    rng = random.Random(seed)
    means = sorted(sum(rng.choice(values) for _ in values) / len(values) for _ in range(n_resamples))
    lo = means[int((alpha / 2) * n_resamples)]
    hi = means[min(n_resamples - 1, int((1 - alpha / 2) * n_resamples))]
    return sum(values) / len(values), lo, hi


def pareto_front(points: List[Tuple[str, float, float]]) -> List[str]:
    """points are (name, cost, score). Keep a point unless another has cost <= and score >=, and is better on one."""
    keep = []
    for n, c, s in points:
        dominated = any((c2 <= c and s2 >= s) and (c2 < c or s2 > s) for n2, c2, s2 in points if n2 != n)
        if not dominated:
            keep.append(n)
    return keep


# =====================================================================================
# Demo
# =====================================================================================
class FakeClock:
    """A clock that moves 12.5 ms on every read, so trace output is the same on every run."""

    def __init__(self, step: float = 0.0125):
        self.t, self.step = 1000.0, step

    def __call__(self) -> float:
        self.t += self.step
        return self.t


def _h(title: str):
    print("\n== %s ==" % title)


def demo():
    from typing import TypedDict

    train, dev = make_tickets(40, seed=1), make_tickets(40, seed=2)

    _h("1. DSPy-like: signature, prompt, and optimizers (made-up tickets, mock LM)")
    sig = Signature.parse("ticket -> category")
    print("signature:", sig, "| with CoT:", sig.prepend_output("reasoning"))
    p = Predict(sig, make_lm(0))
    p.demos = [{"ticket": "refund invoice 1177, the amount is wrong", "category": "billing"}]
    print(render_prompt(sig, p.demos, {"ticket": "why is there an extra charge"}))
    k, seeds, table = 8, range(10), {}
    makers = [("no demos", lambda s: None), ("labeled few-shot", lambda s: LabeledFewShot(k, s)),
              ("bootstrap, 1 round", lambda s: BootstrapFewShot(category_match, k, 0, 1, s)),
              ("bootstrap, 3 rounds", lambda s: BootstrapFewShot(category_match, k, 0, 3, s)),
              ("bootstrap + labeled", lambda s: BootstrapFewShot(category_match, k, k, 3, s)),
              ("random search", lambda s: RandomSearch(category_match, 4, k, 0, 3, s))]
    for name, mk in makers:
        for s in seeds:
            lm = make_lm(s)
            configure(lm)
            opt, prog = mk(s), Classify()
            if opt:
                prog = opt.compile(prog, train)
            configure(make_lm(100 + s))
            table.setdefault(name, []).append(evaluate(prog, dev, category_match).score)
    print("dev accuracy, mean over 10 seeds (k=%d demos, 40 dev tickets):" % k)
    for name, v in table.items():
        print("  %-22s %.3f" % (name, sum(v) / len(v)))
    lm = make_lm(3)
    configure(lm)
    tri = BootstrapFewShot(triage_match, 6, 0, 3, 0).compile(Triage(), train)
    print("two-stage Triage: bootstrapped %d training examples; demos per predictor: %s" % (
        tri._bootstrapped, {n: len(q.demos) for n, q in tri.named_predictors()}))
    best = BootstrapFewShot(category_match, k, k, 3, 0).compile(Classify(), train)

    _h("2. LangGraph-like: state, conditional edge, checkpoint, interrupt, resume")

    class TicketState(TypedDict):
        ticket: str
        category: str
        reply: str
        log: Annotated[list, operator.add]

    runs = {"approve": 0}
    classify_lm = make_lm(7)
    configure(classify_lm)

    def classify(s):
        return {"category": best(ticket=s["ticket"]).category, "log": ["classify"]}

    def draft(s):
        return {"reply": "Hi, about your %s question: we are on it." % s["category"], "log": ["draft"]}

    def approve(s):
        runs["approve"] += 1                      # this line runs again on resume
        answer = interrupt({"question": "Send this reply?", "reply": s["reply"]})
        return {"log": ["approved" if answer else "rejected"]}

    def send(s):
        return {"log": ["send"]}

    def route(s) -> Literal["draft", "__end__"]:
        return END if s["category"] == "feedback" else "draft"

    b = StateGraph(TicketState)
    for n, f in [("classify", classify), ("draft", draft), ("approve", approve), ("send", send)]:
        b.add_node(n, f)
    b.add_edge(START, "classify").add_conditional_edges("classify", route)
    b.add_edge("draft", "approve").add_edge("approve", "send").add_edge("send", END)
    path = os.path.join(tempfile.mkdtemp(), "checkpoints.jsonl")
    cfg = {"configurable": {"thread_id": "ticket-1"}}
    g1 = b.compile(checkpointer=FileSaver(path))
    r = g1.invoke({"ticket": "I was charged twice for my pro plan this month"}, cfg)
    print("first call stops at:", r["__interrupt__"][0])
    g2 = b.compile(checkpointer=FileSaver(path))              # a new compiled graph, as after a restart
    print("snapshot after restart:", g2.get_state(cfg).status, g2.get_state(cfg).next)
    r = g2.invoke(Command(resume=True), cfg)
    print("final log:", r["log"], "| approve node ran", runs["approve"], "times | checkpoints:", len(g2.history(cfg)))
    print(g1.to_mermaid())

    _h("3. mini-SWE-agent-style harness: one bash action per step, in a temp directory")
    d = tempfile.mkdtemp()
    with open(os.path.join(d, "config.ini"), "w") as f:
        f.write("[server]\nhost = localhost\ntimout = 30\nport = 8080\n")
    fence = lambda c, t="Next.": "%s\n```bash\n%s\n```" % (t, c)
    script = [fence("ls"), fence("grep -n timout config.ini"),
              fence("printf '[server]\\nhost = localhost\\ntimeout = 30\\nport = 8080\\n' > config.ini"),
              fence("grep -n timeout config.ini"), fence("echo " + DONE_MARK)]
    agent = BashAgent(ScriptedModel(script), Sandbox(d), step_limit=8)
    res = agent.run("The key 'timout' in config.ini is a typo. Fix it.")
    for e in res.events:
        if e.kind == "assistant":
            print("  action      %s" % _FENCE.findall(e.text)[0][:66])
        elif e.kind in ("observation", "refusal"):
            print("  %-11s %s" % (e.kind, e.text.splitlines()[0][:66]))
    print("status:", res.status, "| steps:", res.steps, "| file fixed:", "timeout = 30" in open(os.path.join(d, "config.ini")).read())
    bad = [fence(c) for c in ("cat ../../etc/passwd", "rm -rf .", "cat $HOME/.profile", "curl http://example.com")] + [fence("echo " + DONE_MARK)]
    res2 = BashAgent(ScriptedModel(bad), Sandbox(d)).run("misbehave")
    print("adversarial script: %d of 4 commands refused; status %s" % (sum(e.kind == "refusal" for e in res2.events), res2.status))
    st = BashAgent(ScriptedModel([fence("ls")] * 6), Sandbox(d), stuck=3).run("loop")
    print("a model that repeats itself: status %s after %d steps" % (st.status, st.steps))
    tree = SessionTree()
    a = tree.add("user", "fix the bug")
    tree.add("assistant", "try plan A")
    tree.branch_from(a)
    tree.add("assistant", "try plan B")
    print("session tree: %d nodes, leaves %s, current path: %s" % (len(tree.nodes), tree.leaves(), [m["content"] for m in tree.path()]))

    _h("4. Temporal-style durable run: crash, replay, and the at-least-once catch")
    notified = []

    def classify_activity(text):
        return {"category": best(ticket=text).category}

    def notify_naive(category):
        notified.append(("naive", category))
        return "sent"

    def notify_keyed(category):
        key = activity_info()["key"]
        if key not in [k for k, _ in notified]:
            notified.append((key, category))
        return "sent"

    def remind(ctx, text):
        found = ctx.activity("classify", text)        # an LM call: recorded once, never repeated on replay
        ctx.sleep(3 * 86400)                          # a timer: costs nothing while it waits
        return ctx.activity("notify", found["category"])

    def make_rt(path, notify):
        rt = Runtime(HistoryStore(path))
        rt.register_activity("classify", classify_activity)
        rt.register_activity("notify", notify)
        rt.register_workflow("remind", remind)
        return rt

    for label, notify in (("without a key", notify_naive), ("with an idempotency key", notify_keyed)):
        del notified[:]
        hp = os.path.join(tempfile.mkdtemp(), "history.jsonl")
        classify_lm.calls = 0
        rt = make_rt(hp, notify)
        s1 = rt.start("wf-1", "remind", "I cannot log in, the password reset email never came")
        rt.advance(3 * 86400)
        rt.crash_at = (7, "before")                    # die after notify ran, before its result is recorded
        try:
            rt.run("wf-1")
        except SimulatedCrash as e:
            crash = str(e)
        rt2 = make_rt(hp, notify)                      # a new worker reads the same history file
        rt2.now = rt.now
        s2 = rt2.run("wf-1")
        print("%-24s first run: %s | crash: %s | after replay: %s | LM calls: %d | notify ran: %d, effect applied: %d" % (
            label, s1.status, crash, s2.status, classify_lm.calls, sum(1 for n, _ in rt.executions + rt2.executions if n == "notify"), len(notified)))
    print(format_history(rt2.store.for_workflow("wf-1")))

    _h("5. Tracing: spans, tokens, cost (made-up prices), JSONL, and a text viewer")
    tr = Tracer(clock=FakeClock(), seed=1)
    tlm = TracedLM(make_lm(9), tr)
    configure(tlm)
    d2 = tempfile.mkdtemp()
    with open(os.path.join(d2, "config.ini"), "w") as f:
        f.write("[server]\ntimout = 30\n")
    with tr.span("invoke_workflow ticket-flow", **{"gen_ai.operation.name": "invoke_workflow"}):
        best(ticket="the delivery of order 1177 is late")
        BashAgent(ScriptedModel([fence("grep -n timout config.ini"), fence("cat ../x"), fence("echo " + DONE_MARK)]),
                  Sandbox(d2), tracer=tr).run("find the typo")
        traced_tool(tr, "lookup_order", lambda n: {"order": n, "status": "late"}, "1177", call_id="call_1")
    print(render_trace(tr.spans))
    sm = summarize(tr.spans)
    print("totals: %d spans, %d errors, tokens in/out %d/%d, cost $%.6f" % (sm["spans"], sm["errors"], sm["input_tokens"], sm["output_tokens"], sm["cost_usd"]))
    print("by operation:", {k: v["count"] for k, v in sorted(sm["by_operation"].items())})
    print("one JSONL line:", export_jsonl(tr.spans[:1])[:150] + " ...")

    _h("6. Cost and confidence: made-up numbers for two harness settings (not real measurements)")
    per_task = {"lean": ([1, 1, 0, 1, 1, 1, 0, 1, 1, 1], [0.6, 0.7, 0.9, 0.5, 0.6, 0.7, 1.1, 0.6, 0.5, 0.7]),
                "heavy": ([1, 1, 1, 1, 1, 1, 0, 0, 1, 1], [1.4, 1.5, 1.6, 1.2, 1.3, 1.5, 1.9, 1.3, 1.2, 1.4])}
    pts = []
    for name, (ok, cost) in per_task.items():
        m, lo, hi = bootstrap_ci(ok, 1000, 0.05, 0)
        c = sum(cost) / len(cost)
        pts.append((name, c, m))
        print("%-6s success %.2f (95%% interval %.2f to %.2f), mean cost $%.2f per task" % (name, m, lo, hi, c))
    print("Pareto front (lower cost, higher success):", pareto_front(pts))


def main(argv: List[str]) -> int:
    """`python frameworks_toolkit.py` runs the demo. `... view spans.jsonl` shows a saved trace as a tree."""
    if len(argv) == 3 and argv[1] == "view":
        with open(argv[2], encoding="utf-8") as f:
            spans = load_jsonl(f.read())
        print(render_trace(spans))
        sm = summarize(spans)
        print("totals: %d spans, %d errors, tokens in/out %d/%d, cost $%.6f" % (
            sm["spans"], sm["errors"], sm["input_tokens"], sm["output_tokens"], sm["cost_usd"]))
        return 0
    if len(argv) == 1:
        demo()
        return 0
    print("usage: python frameworks_toolkit.py [view spans.jsonl]")
    return 2


if __name__ == "__main__":
    import sys
    sys.exit(main(sys.argv))

Run the demo

$ python frameworks_toolkit.py
== 1. DSPy-like: signature, prompt, and optimizers (made-up tickets, mock LM) ==
signature: ticket -> category | with CoT: ticket -> reasoning, category
INSTRUCTIONS: Given the fields ticket, produce the fields category.
OUTPUT FIELDS: category
Reply with one line per output field, like "name: value".
--- demo
ticket: refund invoice 1177, the amount is wrong
category: billing
--- input
ticket: why is there an extra charge
dev accuracy, mean over 10 seeds (k=8 demos, 40 dev tickets):
  no demos               0.195
  labeled few-shot       0.608
  bootstrap, 1 round     0.622
  bootstrap, 3 rounds    0.657
  bootstrap + labeled    0.807
  random search          0.672
two-stage Triage: bootstrapped 6 training examples; demos per predictor: {'category': 6, 'urgency': 6}

== 2. LangGraph-like: state, conditional edge, checkpoint, interrupt, resume ==
first call stops at: Interrupt(value={'question': 'Send this reply?', 'reply': 'Hi, about your billing question: we are on it.'}, node='approve')
snapshot after restart: interrupted ['approve']
final log: ['classify', 'draft', 'approved', 'send'] | approve node ran 2 times | checkpoints: 6
graph TD
  __start__ --> classify
  draft --> approve
  approve --> send
  send --> __end__
  classify -.-> draft
  classify -.-> __end__

== 3. mini-SWE-agent-style harness: one bash action per step, in a temp directory ==
  action      ls
  observation exit 0
  action      grep -n timout config.ini
  observation exit 0
  action      printf '[server]\nhost = localhost\ntimeout = 30\nport = 8080\n' >
  observation exit 0
  action      grep -n timeout config.ini
  observation exit 0
  action      echo TASK_COMPLETE
status: submitted | steps: 5 | file fixed: True
adversarial script: 4 of 4 commands refused; status submitted
a model that repeats itself: status stuck after 3 steps
session tree: 4 nodes, leaves [2, 3], current path: ['system', 'fix the bug', 'try plan B']

== 4. Temporal-style durable run: crash, replay, and the at-least-once catch ==
without a key            first run: suspended | crash: crash before event 7 (ActivityCompleted) | after replay: completed | LM calls: 1 | notify ran: 2, effect applied: 2
with an idempotency key  first run: suspended | crash: crash before event 7 (ActivityCompleted) | after replay: completed | LM calls: 1 | notify ran: 2, effect applied: 1
 1  t=0       WorkflowStarted        {"args": ["I cannot log in, the password reset email never came"], "na
 2  t=0       ActivityScheduled      {"args_hash": "42180cd59c", "name": "classify", "seq": 0}
 3  t=0       ActivityCompleted      {"attempts": 1, "result": {"category": "login"}, "seq": 0}
 4  t=0       TimerStarted           {"fire_at": 259200.0, "seq": 1}
 5  t=259200  TimerFired             {"seq": 1}
 6  t=259200  ActivityScheduled      {"args_hash": "660d2ff597", "name": "notify", "seq": 2}
 7  t=259200  ActivityCompleted      {"attempts": 1, "result": "sent", "seq": 2}
 8  t=259200  WorkflowCompleted      {"result": "sent"}

== 5. Tracing: spans, tokens, cost (made-up prices), JSONL, and a text viewer ==
invoke_workflow ticket-flow  [212.5 ms]
  chat mock-imitator  [12.5 ms, tok 352/5, $0.000056]
  invoke_agent bash-agent  [137.5 ms]
    chat mock-model  [12.5 ms, tok 32/11, $0.000011]
    execute_tool bash  [12.5 ms]
    chat mock-model  [12.5 ms, tok 67/7, $0.000014]
    execute_tool bash  [12.5 ms, ERROR SandboxError, refused]
    chat mock-model  [12.5 ms, tok 112/9, $0.000022]
  execute_tool lookup_order  [12.5 ms]
totals: 9 spans, 1 errors, tokens in/out 563/32, cost $0.000104
by operation: {'chat': 4, 'execute_tool': 3, 'invoke_agent': 1, 'invoke_workflow': 1}
one JSONL line: {"attributes": {"app.cost_usd": 5.5799999999999994e-05, "gen_ai.operation.name": "chat", "gen_ai.provider.name": "mock", "gen_ai.request.model": "mock ...

== 6. Cost and confidence: made-up numbers for two harness settings (not real measurements) ==
lean   success 0.80 (95% interval 0.50 to 1.00), mean cost $0.69 per task
heavy  success 0.80 (95% interval 0.50 to 1.00), mean cost $1.43 per task
Pareto front (lower cost, higher success): ['lean']

Exercises

1. Your own signature. Parse "ticket, language -> category, urgency: bool". Print the prompt it builds. Then break it five ways (no arrow, two arrows, a duplicate name, a bad type, an empty side) and check that each raises SignatureError [4].
2. Where the optimizer stops. Make a metric that always fails. What does BootstrapFewShot return? Now give the mock a 0.9 chance of format slips. How many training examples do you need before you get 8 demonstrations? Connect your answer to the paper's statement that a well-decomposed program can usually find a few passing examples [1].
3. Dev set honesty. Write 20 tickets with new wording, not from the templates. Score the compiled programs on them. How far does the score fall? What does that say about the dev numbers in the demo?
4. A third stage. Add a BestOfN wrapper around Classify with a reward that prefers shorter replies. Does it change the mock's score? Why or why not?
5. A new reducer. Add a state key with a "max" reducer. Write a graph where two parallel nodes write to it. Then remove the reducer and watch InvalidUpdateError. Explain why this is my rule, and not LangGraph's. Check what the LangGraph docs say about concurrent updates [10].
6. Break the interrupt. Put a counter increment before interrupt() in a node. How many times does it run after two resumes? Fix it by moving the side effect to its own node, and show the new count [12].
7. Time travel. Use history() to list the checkpoints for a thread. Copy the checkpoint after step 1 into a new thread with saver.put(...), then continue it with invoke(None, config). What is missing compared with the real feature in the docs [11]?
8. Stronger sandbox. Find a command that passes Sandbox.check but still reads outside the work directory (hint: think about what an allowed program can do with its arguments, or with files that are already in the directory). Write a test, then a fix, then write down what the fix still cannot stop. Why is a container better [13]?
9. Tool shapes. Change the agent so that it exposes three tools (read, write, bash) instead of bash only. Count how many steps the config-file task takes in each form. Does the answer match what you expect from the lecture's comparison of mini-SWE-agent and Pi [13] [16]?
10. A crash in the retry loop. Make an activity that fails twice, then succeeds. Put a crash after the first failure. Print the attempt numbers across the crash. Then explain what a real Temporal worker would do differently [24].
11. Determinism. Write a workflow that calls random.random() directly and decides which activity to run. Show a replay that fails with NonDeterminismError. Fix it with ctx.side_effect and re-run [22].
12. Signals. Write a workflow that waits for an "approve" signal or for a 3-day timer, whichever comes first. The lab's runtime cannot do this directly. What would you add? (Hint: look at how sleep and wait_for_signal record events.)
13. Trace what matters. Wrap a graph node with traced_node and a tool with traced_tool. Export JSONL, then use python frameworks_toolkit.py view on the file. Add an attribute that you would want when a run goes wrong, and say whether it is safe to keep in production [25].
14. A real price table. Replace the made-up price table with the current list for a model you use. What does the lab's cost calculation ignore (cached tokens, tiers, reasoning tokens)? How big could the error be?
15. Reproduce a small HarnessTax. Pick two setups of your own (for example, a short and a long system prompt). Run 20 mock tasks with 3 runs each. Report success, cost, a bootstrap interval, and the Pareto front. State what you cannot conclude [3].

The tests: test_frameworks_toolkit.py

The tests check the signature parser, prompt building and reply parsing, the modules, the mock model, every optimizer, the state graph (reducers, routing, fan-out, recursion limit, interrupts, resume, file checkpoints and crash recovery), the sandbox (including real timeouts, a FIFO and escape attempts), the agent loop, the durable runtime (retries, timers, signals, replay, non-determinism, at-least-once and idempotency), the tracer, the JSONL round trip, the cost arithmetic (hand-computed), the bootstrap interval, the Pareto front, the command line and the demo.

"""Tests for frameworks_toolkit.py. Run: python test_frameworks_toolkit.py -v
Standard library only. No network and no API key."""
import contextlib
import io
import json
import operator
import os
import shutil
import tempfile
import time
import unittest
from typing import Annotated, TypedDict, Literal

import frameworks_toolkit as ft


def scripted_lm(replies):
    """An LM that returns the given replies in order."""
    class S(ft.LM):
        def __call__(self, prompt):
            self.calls += 1
            self.history.append(prompt)
            return replies[(self.calls - 1) % len(replies)]
    return S()


class TmpDirCase(unittest.TestCase):
    def setUp(self):
        self.dir = tempfile.mkdtemp()
        self.addCleanup(shutil.rmtree, self.dir, True)
        ft.configure(None)

    def path(self, name):
        return os.path.join(self.dir, name)


# ----------------------------------------------------------------------------- 1. DSPy-like
class TestSignature(unittest.TestCase):
    def test_parse_simple(self):
        s = ft.Signature.parse("question -> answer")
        self.assertEqual([f.name for f in s.inputs], ["question"])
        self.assertEqual([f.name for f in s.outputs], ["answer"])
        self.assertEqual(str(s), "question -> answer")

    def test_parse_typed_and_multiple(self):
        s = ft.Signature.parse("context, question: str -> answer: int, ok: bool")
        self.assertEqual([(f.name, f.type) for f in s.inputs], [("context", "str"), ("question", "str")])
        self.assertEqual([(f.name, f.type) for f in s.outputs], [("answer", "int"), ("ok", "bool")])
        self.assertEqual(str(s), "context, question -> answer: int, ok: bool")

    def test_bad_signatures(self):
        for bad in ["question answer", "a -> b -> c", " -> b", "a -> ", "1a -> b", "a -> b: dict", "a, a -> b", "a -> a"]:
            with self.assertRaises(ft.SignatureError, msg=bad):
                ft.Signature.parse(bad)

    def test_prepend_output_keeps_original(self):
        s = ft.Signature.parse("q -> a")
        c = s.prepend_output("reasoning")
        self.assertEqual([f.name for f in c.outputs], ["reasoning", "a"])
        self.assertEqual([f.name for f in s.outputs], ["a"])

    def test_default_and_custom_instructions(self):
        self.assertIn("produce the fields a", ft.Signature.parse("q -> a").instructions)
        self.assertEqual(ft.Signature.parse("q -> a", "Be brief.").instructions, "Be brief.")


class TestPromptAndParse(unittest.TestCase):
    def setUp(self):
        self.sig = ft.Signature.parse("q -> a: int")

    def test_render_has_demos_inputs_and_output_fields(self):
        p = ft.render_prompt(self.sig, [{"q": "1+1", "a": 2}], {"q": "2+2"})
        self.assertIn("OUTPUT FIELDS: a", p)
        self.assertIn("--- demo\nq: 1+1\na: 2", p)
        self.assertTrue(p.endswith("--- input\nq: 2+2"))

    def test_parse_reply_typed_and_ignores_noise(self):
        self.assertEqual(ft.parse_reply(self.sig, "thinking...\na: 42\nextra: x"), {"a": 42})

    def test_parse_reply_missing_field(self):
        with self.assertRaises(ft.ParseError):
            ft.parse_reply(self.sig, "the answer is 4")

    def test_coerce(self):
        self.assertEqual(ft._coerce("3.5", "float"), 3.5)
        self.assertIs(ft._coerce("Yes", "bool"), True)
        self.assertIs(ft._coerce("no", "bool"), False)
        for typ, bad in [("int", "x"), ("float", "x"), ("bool", "maybe")]:
            with self.assertRaises(ft.ParseError):
                ft._coerce(bad, typ)


class TestExample(unittest.TestCase):
    def test_inputs_labels_and_attrs(self):
        e = ft.Example(q="x", a="y").with_inputs("q")
        self.assertEqual(e.inputs(), {"q": "x"})
        self.assertEqual(e.labels(), {"a": "y"})
        self.assertEqual(e.a, "y")
        with self.assertRaises(AttributeError):
            e.nope

    def test_with_inputs_returns_a_copy(self):
        e = ft.Example(q="x", a="y")
        self.assertEqual(e.inputs(), {})
        self.assertEqual(e.with_inputs("q").inputs(), {"q": "x"})
        self.assertEqual(e.inputs(), {})


class TestPredict(TmpDirCase):
    def test_predict_reads_typed_output(self):
        p = ft.Predict("q -> a: int", scripted_lm(["a: 7"]))
        out = p(q="x")
        self.assertEqual((out.q, out.a), ("x", 7))

    def test_wrong_inputs_raise(self):
        p = ft.Predict("q -> a", scripted_lm(["a: 1"]))
        with self.assertRaises(TypeError):
            p(q="x", extra="y")
        with self.assertRaises(TypeError):
            p()

    def test_needs_an_lm(self):
        with self.assertRaises(RuntimeError):
            ft.Predict("q -> a")(q="x")

    def test_global_lm_and_own_lm(self):
        ft.configure(scripted_lm(["a: global"]))
        self.assertEqual(ft.Predict("q -> a")(q="x").a, "global")
        self.assertEqual(ft.Predict("q -> a", scripted_lm(["a: own"]))(q="x").a, "own")

    def test_demos_go_into_the_prompt(self):
        lm = scripted_lm(["a: 1"])
        p = ft.Predict("q -> a", lm)
        p.demos = [{"q": "one", "a": "uno"}]
        p(q="two")
        self.assertIn("q: one\na: uno", lm.history[0])

    def test_trace_records_predictor_calls(self):
        p = ft.Predict("q -> a", scripted_lm(["a: 1"]))
        ft._trace_stack.append([])
        try:
            p(q="x")
        finally:
            trace = ft._trace_stack.pop()
        self.assertEqual(len(trace), 1)
        self.assertIs(trace[0][0], p)
        self.assertEqual(trace[0][2], {"a": "1"})

    def test_chain_of_thought_adds_reasoning_first(self):
        lm = scripted_lm(["reasoning: because\na: 5"])
        cot = ft.ChainOfThought("q -> a", lm)
        out = cot(q="x")
        self.assertEqual((out.reasoning, out.a), ("because", "5"))
        self.assertIn("OUTPUT FIELDS: reasoning, a", lm.history[0])

    def test_named_predictors_found_in_nested_modules(self):
        t = ft.Triage()
        self.assertEqual([n for n, _ in t.named_predictors()], ["category", "urgency"])
        c = ft.Classify()
        self.assertEqual([n for n, _ in c.named_predictors()], ["predict"])
        self.assertEqual([n for n, _ in ft.Predict("q -> a").named_predictors()], ["self"])


class TestBestOfN(TmpDirCase):
    def test_keeps_best_reward_and_stops_early(self):
        lm = scripted_lm(["a: 1", "a: 9", "a: 5"])
        b = ft.BestOfN(ft.Predict("q -> a: int", lm), 3, lambda kw, p: p.a, threshold=9)
        self.assertEqual(b(q="x").a, 9)
        self.assertEqual(lm.calls, 2)        # stopped when the threshold was met

    def test_runs_all_n_when_threshold_is_not_met(self):
        lm = scripted_lm(["a: 1", "a: 2", "a: 3"])
        b = ft.BestOfN(ft.Predict("q -> a: int", lm), 3, lambda kw, p: p.a, threshold=100)
        self.assertEqual(b(q="x").a, 3)
        self.assertEqual(lm.calls, 3)

    def test_skips_bad_replies_and_raises_if_all_bad(self):
        lm = scripted_lm(["junk", "a: 4"])
        self.assertEqual(ft.BestOfN(ft.Predict("q -> a: int", lm), 2, lambda kw, p: p.a)(q="x").a, 4)
        with self.assertRaises(ft.ParseError):
            ft.BestOfN(ft.Predict("q -> a: int", scripted_lm(["junk"])), 2, lambda kw, p: 0)(q="x")

    def test_n_must_be_positive(self):
        with self.assertRaises(ValueError):
            ft.BestOfN(ft.Predict("q -> a"), 0, lambda kw, p: 0)


class TestImitatorLM(TmpDirCase):
    def test_copies_the_closest_demo(self):
        lm = ft.make_lm(0, p_format_slip=0.0)
        p = ft.Predict("ticket -> category", lm)
        p.demos = [{"ticket": "refund invoice 1177", "category": "billing"},
                   {"ticket": "package tracking is stuck", "category": "shipping"}]
        self.assertEqual(p(ticket="the tracking of my package is late").category, "shipping")
        self.assertEqual(p(ticket="refund for invoice 9034").category, "billing")

    def test_same_seed_same_replies(self):
        a, b = ft.make_lm(5), ft.make_lm(5)
        prompt = ft.render_prompt(ft.Signature.parse("ticket -> category"), [], {"ticket": "hello"})
        self.assertEqual([a(prompt) for _ in range(6)], [b(prompt) for _ in range(6)])

    def test_slip_breaks_format_only_without_demos(self):
        lm = ft.make_lm(0, p_format_slip=1.0)
        p = ft.Predict("ticket -> category", lm)
        with self.assertRaises(ft.ParseError):
            p(ticket="hello there")
        p.demos = [{"ticket": "hello there", "category": "feedback"}]
        self.assertEqual(p(ticket="hello there").category, "feedback")

    def test_counts_calls_and_keeps_prompts(self):
        lm = ft.make_lm(0)
        p = ft.Predict("ticket -> category", lm)
        p.demos = [{"ticket": "x", "category": "login"}]
        p(ticket="a")
        p(ticket="b")
        self.assertEqual((lm.calls, len(lm.history)), (2, 2))

    def test_overlap_ignores_stop_words(self):
        self.assertEqual(ft.overlap("the a of", "to is it"), 0.0)
        self.assertAlmostEqual(ft.overlap("refund invoice", "refund invoice"), 1.0)
        self.assertAlmostEqual(ft.overlap("refund invoice", "refund order"), 1 / 3)


class TestTicketsAndEvaluate(TmpDirCase):
    def test_make_tickets_is_deterministic_and_balanced(self):
        a, b = ft.make_tickets(40, 3), ft.make_tickets(40, 3)
        self.assertEqual([e.toDict() for e in a], [e.toDict() for e in b])
        counts = {c: sum(e.category == c for e in a) for c in ft.CATEGORIES}
        self.assertEqual(set(counts.values()), {10})
        self.assertEqual(a[0].inputs().keys(), {"ticket"})
        self.assertIn(a[0].urgency, ("low", "high"))

    def test_evaluate_scores_and_counts_calls(self):
        lm = scripted_lm(["category: billing"])
        ft.configure(lm)
        dev = [ft.Example(ticket="x", category="billing").with_inputs("ticket"),
               ft.Example(ticket="y", category="login").with_inputs("ticket")]
        r = ft.evaluate(ft.Classify(), dev, ft.category_match)
        self.assertEqual((r.score, r.n, r.failures, r.lm_calls), (0.5, 2, 0, 2))
        self.assertEqual([ok for ok, _ in r.rows], [True, False])

    def test_exception_counts_as_a_miss(self):
        ft.configure(scripted_lm(["no format here"]))
        dev = ft.make_tickets(5, 1)
        r = ft.evaluate(ft.Classify(), dev, ft.category_match)
        self.assertEqual((r.score, r.failures), (0.0, 5))
        self.assertEqual(r.rows[0][1], "ParseError")

    def test_empty_devset(self):
        ft.configure(scripted_lm(["category: x"]))
        self.assertEqual(ft.evaluate(ft.Classify(), [], ft.category_match).score, 0.0)


class TestOptimizers(TmpDirCase):
    def setUp(self):
        super().setUp()
        self.train, self.dev = ft.make_tickets(40, 1), ft.make_tickets(40, 2)

    def test_labeled_few_shot_puts_k_demos_and_leaves_original(self):
        prog = ft.Classify()
        out = ft.LabeledFewShot(5, seed=1).compile(prog, self.train)
        self.assertEqual(len(out.predict.demos), 5)
        self.assertEqual(prog.predict.demos, [])
        self.assertEqual(set(out.predict.demos[0]), {"ticket", "category"})

    def test_labeled_few_shot_fills_every_stage_of_a_program(self):
        out = ft.LabeledFewShot(3).compile(ft.Triage(), self.train)
        self.assertEqual({n: len(p.demos) for n, p in out.named_predictors()}, {"category": 3, "urgency": 3})
        self.assertEqual(set(out.urgency.demos[0]), {"ticket", "category", "urgency"})

    def test_bootstrap_with_always_false_metric_keeps_nothing(self):
        ft.configure(ft.make_lm(0))
        out = ft.BootstrapFewShot(lambda e, p, trace=None: False, 4, 0, 2).compile(ft.Classify(), self.train)
        self.assertEqual(out.predict.demos, [])
        self.assertEqual(out._bootstrapped, 0)

    def test_bootstrap_with_always_true_metric_stops_at_max(self):
        ft.configure(ft.make_lm(0, p_format_slip=0.0))
        out = ft.BootstrapFewShot(lambda e, p, trace=None: True, 4, 0, 1).compile(ft.Classify(), self.train)
        self.assertEqual(len(out.predict.demos), 4)
        self.assertEqual(out._bootstrapped, 4)

    def test_bootstrapped_demos_come_from_passing_runs_only(self):
        ft.configure(ft.make_lm(2))
        out = ft.BootstrapFewShot(ft.category_match, 6, 0, 3).compile(ft.Classify(), self.train)
        gold = {e.ticket: e.category for e in self.train}
        self.assertGreater(len(out.predict.demos), 0)
        for d in out.predict.demos:
            self.assertEqual(d["category"], gold[d["ticket"]])

    def test_bootstrap_traces_every_stage_of_a_two_stage_program(self):
        ft.configure(ft.make_lm(3))
        out = ft.BootstrapFewShot(ft.triage_match, 5, 0, 4).compile(ft.Triage(), self.train)
        n = out._bootstrapped
        self.assertGreater(n, 0)
        self.assertEqual(len(out.category.demos), n)
        self.assertEqual(len(out.urgency.demos), n)
        self.assertEqual(set(out.urgency.demos[0]), {"ticket", "category", "urgency"})

    def test_bootstrap_does_not_touch_the_student_or_its_lm(self):
        lm = ft.make_lm(0)
        prog = ft.Classify()
        prog.set_lm(lm)
        out = ft.BootstrapFewShot(ft.category_match, 4, 0, 2).compile(prog, self.train)
        self.assertEqual(prog.predict.demos, [])
        self.assertIs(out.predict.lm, lm)          # the LM is shared, not copied
        self.assertGreater(lm.calls, 0)

    def test_more_rounds_never_bootstrap_fewer_examples(self):
        got = []
        for rounds in (1, 3):
            ft.configure(ft.make_lm(4))
            got.append(ft.BootstrapFewShot(ft.category_match, 8, 0, rounds, 1).compile(ft.Classify(), self.train)._bootstrapped)
        self.assertGreaterEqual(got[1], got[0])

    def test_max_labeled_demos_are_added_after_bootstrapped_ones(self):
        ft.configure(ft.make_lm(1))
        out = ft.BootstrapFewShot(ft.category_match, 3, 4, 2).compile(ft.Classify(), self.train)
        self.assertEqual(len(out.predict.demos), out._bootstrapped + 4)

    def test_a_teacher_can_be_given(self):
        good = ft.Classify()
        good.set_lm(scripted_lm(["category: billing"]))
        student = ft.Classify()
        student.set_lm(scripted_lm(["category: login"]))
        train = [ft.Example(ticket="t%d" % i, category="billing").with_inputs("ticket") for i in range(6)]
        out = ft.BootstrapFewShot(ft.category_match, 3, 0, 1).compile(student, train, teacher=good)
        self.assertEqual(len(out.predict.demos), 3)
        self.assertTrue(all(d["category"] == "billing" for d in out.predict.demos))

    def test_optimizing_beats_no_demos_on_average(self):
        base, tuned = [], []
        for s in range(5):
            ft.configure(ft.make_lm(s))
            prog = ft.BootstrapFewShot(ft.category_match, 8, 8, 3, s).compile(ft.Classify(), self.train)
            ft.configure(ft.make_lm(100 + s))
            tuned.append(ft.evaluate(prog, self.dev, ft.category_match).score)
            base.append(ft.evaluate(ft.Classify(), self.dev, ft.category_match).score)
        self.assertGreater(sum(tuned) / 5 - sum(base) / 5, 0.4)

    def test_random_search_returns_the_best_candidate(self):
        ft.configure(ft.make_lm(0))
        rs = ft.RandomSearch(ft.category_match, 3, 6, 0, 2)
        best = rs.compile(ft.Classify(), self.train, self.dev)
        self.assertEqual(len(rs.scores), 5)               # no demos, labeled, and 3 bootstraps
        i = rs.scores.index(max(rs.scores))
        self.assertEqual(rs.scores[i], max(rs.scores))
        self.assertTrue(all(isinstance(x, float) for x in rs.scores))
        if i == 0:
            self.assertEqual(best.predict.demos, [])
        else:
            self.assertGreater(len(best.predict.demos), 0)

    def test_random_search_ties_pick_the_earliest_candidate(self):
        ft.configure(ft.make_lm(0, p_format_slip=0.0))
        rs = ft.RandomSearch(lambda e, p, trace=None: True, 2, 3, 0, 1)
        best = rs.compile(ft.Classify(), self.train)
        self.assertEqual(best.predict.demos, [])          # candidate 0 is the program with no demos


# ----------------------------------------------------------------------------- 2. StateGraph
class S1(TypedDict):
    x: int
    log: Annotated[list, operator.add]


class S2(TypedDict):
    a: str
    note: str


def build_linear():
    b = ft.StateGraph(S1)
    b.add_node("inc", lambda s: {"x": s["x"] + 1, "log": ["inc"]})
    b.add_node("dbl", lambda s: {"x": s["x"] * 2, "log": ["dbl"]})
    b.add_edge(ft.START, "inc").add_edge("inc", "dbl").add_edge("dbl", ft.END)
    return b


class TestGraphBasics(unittest.TestCase):
    def test_reducers_are_read_from_annotations(self):
        g = ft.StateGraph(S1)
        self.assertEqual(g.keys, ["x", "log"])
        self.assertIn("log", g.reducers)
        self.assertNotIn("x", g.reducers)

    def test_linear_run(self):
        out = build_linear().compile().invoke({"x": 3, "log": []})
        self.assertEqual(out, {"x": 8, "log": ["inc", "dbl"]})

    def test_reducer_starts_from_an_empty_list(self):
        out = build_linear().compile().invoke({"x": 1})
        self.assertEqual(out["log"], ["inc", "dbl"])

    def test_node_returning_none_changes_nothing(self):
        b = ft.StateGraph(S2)
        b.add_node("n", lambda s: None).add_edge(ft.START, "n").add_edge("n", ft.END)
        self.assertEqual(b.compile().invoke({"a": "q"}), {"a": "q"})

    def test_nodes_get_a_copy_of_the_state(self):
        def sneaky(s):
            s["log"].append("mutated")
            return {"x": 1}
        b = ft.StateGraph(S1)
        b.add_node("n", sneaky).add_edge(ft.START, "n").add_edge("n", ft.END)
        self.assertEqual(b.compile().invoke({"x": 0, "log": ["a"]})["log"], ["a"])

    def test_bad_updates(self):
        for bad, exc in [({"nope": 1}, ft.InvalidUpdateError), ("text", ft.InvalidUpdateError)]:
            b = ft.StateGraph(S2)
            b.add_node("n", lambda s, bad=bad: bad).add_edge(ft.START, "n").add_edge("n", ft.END)
            with self.assertRaises(exc):
                b.compile().invoke({"a": "q"})

    def test_compile_checks(self):
        with self.assertRaises(ValueError):
            ft.StateGraph(S2).compile()                          # no START edge
        b = ft.StateGraph(S2)
        b.add_node("n", lambda s: None).add_edge(ft.START, "n")
        with self.assertRaises(ValueError):
            b.compile()                                          # n has no way out
        with self.assertRaises(ValueError):
            b.add_edge("n", "ghost")
        with self.assertRaises(ValueError):
            b.add_node("n", lambda s: None)                      # duplicate
        with self.assertRaises(ValueError):
            b.add_node(ft.END, lambda s: None)                   # reserved


class TestGraphRouting(unittest.TestCase):
    def make(self, router, pmap=None):
        b = ft.StateGraph(S1)
        b.add_node("a", lambda s: {"log": ["a"]}).add_node("b", lambda s: {"log": ["b"]}).add_node("c", lambda s: {"log": ["c"]})
        b.add_edge(ft.START, "a").add_conditional_edges("a", router, pmap)
        b.add_edge("b", ft.END).add_edge("c", ft.END)
        return b

    def test_conditional_edge_picks_a_branch(self):
        b = self.make(lambda s: "b" if s["x"] > 0 else "c")
        self.assertEqual(b.compile().invoke({"x": 1})["log"], ["a", "b"])
        self.assertEqual(b.compile().invoke({"x": 0})["log"], ["a", "c"])

    def test_path_map_translates_labels(self):
        b = self.make(lambda s: "yes", {"yes": "b", "no": "c"})
        self.assertEqual(b.compile().invoke({"x": 0})["log"], ["a", "b"])

    def test_router_can_return_end(self):
        b = self.make(lambda s: ft.END)
        self.assertEqual(b.compile().invoke({"x": 0})["log"], ["a"])

    def test_router_to_unknown_node_fails(self):
        with self.assertRaises(ValueError):
            self.make(lambda s: "ghost").compile().invoke({"x": 0})

    def test_fan_out_runs_in_one_super_step_and_merges(self):
        b = self.make(lambda s: ["c", "b"])
        g = b.compile()
        out = g.invoke({"x": 0})
        self.assertEqual(out["log"], ["a", "b", "c"])            # merged in sorted node order
        self.assertEqual(g.events, [("run", "a"), ("run", "b"), ("run", "c")])

    def test_two_writes_to_a_plain_key_are_refused(self):
        b = ft.StateGraph(S2)
        b.add_node("p", lambda s: {"note": "p"}).add_node("q", lambda s: {"note": "q"})
        b.add_edge(ft.START, "p").add_edge(ft.START, "q").add_edge("p", ft.END).add_edge("q", ft.END)
        with self.assertRaises(ft.InvalidUpdateError):
            b.compile().invoke({"a": "x"})

    def test_a_loop_hits_the_recursion_limit(self):
        b = ft.StateGraph(S1)
        b.add_node("n", lambda s: {"x": s["x"] + 1}).add_edge(ft.START, "n")
        b.add_conditional_edges("n", lambda s: "n" if s["x"] < 100 else ft.END)
        with self.assertRaises(ft.GraphRecursionError):
            b.compile(recursion_limit=5).invoke({"x": 0})
        self.assertEqual(b.compile().invoke({"x": 0}, {"recursion_limit": 200})["x"], 100)
        self.assertEqual(b.compile(recursion_limit=30).invoke({"x": 80})["x"], 100)

    def test_mermaid_shows_static_and_conditional_edges(self):
        def r(s) -> Literal["b", "c"]:
            return "b"
        text = self.make(r).compile().to_mermaid()
        self.assertIn("__start__ --> a", text)
        self.assertIn("a -.-> b", text)
        self.assertIn("a -.-> c", text)
        self.assertIn("b --> __end__", text)


class S3(TypedDict):
    draft: str
    approved: bool
    msgs: Annotated[list, operator.add]


def build_review(counter=None):
    """draft -> review (asks a human) -> send or discard."""
    counter = counter if counter is not None else []

    def draft(s):
        return {"draft": "refund 20 USD", "msgs": ["drafted"]}

    def review(s):
        counter.append("review started")                    # runs again on resume
        ok = ft.interrupt({"question": "send this?", "draft": s["draft"]})
        return {"approved": bool(ok), "msgs": ["reviewed"]}

    b = ft.StateGraph(S3)
    b.add_node("draft", draft).add_node("review", review)
    b.add_node("send", lambda s: {"msgs": ["sent"]}).add_node("discard", lambda s: {"msgs": ["discarded"]})
    b.add_edge(ft.START, "draft").add_edge("draft", "review")
    b.add_conditional_edges("review", lambda s: "send" if s["approved"] else "discard")
    b.add_edge("send", ft.END).add_edge("discard", ft.END)
    return b, counter


CFG = {"configurable": {"thread_id": "t1"}}


class TestInterruptResume(TmpDirCase):
    def test_pause_shows_the_question_and_saves_a_checkpoint(self):
        b, _ = build_review()
        g = b.compile(ft.InMemorySaver())
        out = g.invoke({"draft": "", "approved": False}, CFG)
        self.assertEqual(out["__interrupt__"][0].value["question"], "send this?")
        self.assertEqual(out["__interrupt__"][0].node, "review")
        snap = g.get_state(CFG)
        self.assertEqual((snap.status, snap.next), ("interrupted", ["review"]))
        self.assertEqual(snap.values["msgs"], ["drafted"])

    def test_resume_finishes_the_run_and_the_node_restarts(self):
        b, counter = build_review()
        g = b.compile(ft.InMemorySaver())
        g.invoke({"draft": "", "approved": False}, CFG)
        out = g.invoke(ft.Command(resume=True), CFG)
        self.assertEqual(out["msgs"], ["drafted", "reviewed", "sent"])
        self.assertEqual(counter, ["review started", "review started"])      # the node ran twice
        self.assertEqual(g.events.count(("run", "draft")), 1)                # earlier nodes did not rerun
        self.assertEqual(g.get_state(CFG).status, "done")

    def test_the_human_can_say_no(self):
        b, _ = build_review()
        g = b.compile(ft.InMemorySaver())
        g.invoke({"draft": "", "approved": False}, CFG)
        self.assertEqual(g.invoke(ft.Command(resume=False), CFG)["msgs"][-1], "discarded")

    def test_resume_needs_a_waiting_thread(self):
        b, _ = build_review()
        g = b.compile(ft.InMemorySaver())
        with self.assertRaises(ValueError):
            g.invoke(ft.Command(resume=True), CFG)

    def test_new_input_while_waiting_is_refused(self):
        b, _ = build_review()
        g = b.compile(ft.InMemorySaver())
        g.invoke({"draft": "", "approved": False}, CFG)
        with self.assertRaises(ValueError):
            g.invoke({"draft": "other"}, CFG)
        with self.assertRaises(ValueError):
            g.invoke(None, CFG)

    def test_interrupt_outside_a_graph_is_an_error(self):
        with self.assertRaises(RuntimeError):
            ft.interrupt("hello")

    def test_two_interrupts_in_one_node_resume_in_order(self):
        def node(s):
            a = ft.interrupt("first?")
            b = ft.interrupt("second?")
            return {"msgs": ["%s+%s" % (a, b)]}
        b = ft.StateGraph(S3)
        b.add_node("n", node).add_edge(ft.START, "n").add_edge("n", ft.END)
        g = b.compile(ft.InMemorySaver())
        self.assertEqual(g.invoke({"draft": ""}, CFG)["__interrupt__"][0].value, "first?")
        self.assertEqual(g.invoke(ft.Command(resume="A"), CFG)["__interrupt__"][0].value, "second?")
        self.assertEqual(g.invoke(ft.Command(resume="B"), CFG)["msgs"], ["A+B"])

    def test_interrupt_needs_a_checkpointer_to_resume(self):
        b, _ = build_review()
        g = b.compile()                         # no saver
        out = g.invoke({"draft": "", "approved": False}, CFG)
        self.assertIn("__interrupt__", out)
        with self.assertRaises(ValueError):
            g.invoke(ft.Command(resume=True), CFG)


class TestCheckpoints(TmpDirCase):
    def test_one_checkpoint_per_super_step_plus_the_input(self):
        g = build_linear().compile(ft.InMemorySaver())
        g.invoke({"x": 1}, CFG)
        hist = g.history(CFG)
        self.assertEqual(len(hist), 3)                 # input, after inc, after dbl (done)
        self.assertEqual([h["step"] for h in hist], [0, 1, 2])
        self.assertEqual(hist[-1]["status"], "done")
        self.assertEqual(hist[1]["state"]["x"], 2)     # time travel: the state after the first node

    def test_threads_are_separate(self):
        g = build_linear().compile(ft.InMemorySaver())
        g.invoke({"x": 1}, {"configurable": {"thread_id": "a"}})
        g.invoke({"x": 10}, {"configurable": {"thread_id": "b"}})
        self.assertEqual(g.get_state({"configurable": {"thread_id": "a"}}).values["x"], 4)
        self.assertEqual(g.get_state({"configurable": {"thread_id": "b"}}).values["x"], 22)
        with self.assertRaises(KeyError):
            g.get_state({"configurable": {"thread_id": "zzz"}})

    def test_a_second_input_on_the_same_thread_continues_the_state(self):
        g = build_linear().compile(ft.InMemorySaver())
        g.invoke({"x": 1}, CFG)
        out = g.invoke({"x": 5}, CFG)                  # x is replaced, log is appended to
        self.assertEqual(out["log"], ["inc", "dbl", "inc", "dbl"])

    def test_file_saver_survives_a_new_process(self):
        path = self.path("ckpt.jsonl")
        b, _ = build_review()
        b.compile(ft.FileSaver(path)).invoke({"draft": "", "approved": False}, CFG)
        b2, counter2 = build_review()                                  # a "restart": new objects, same file
        g2 = b2.compile(ft.FileSaver(path))
        self.assertEqual(g2.get_state(CFG).status, "interrupted")
        out = g2.invoke(ft.Command(resume=True), CFG)
        self.assertEqual(out["msgs"], ["drafted", "reviewed", "sent"])
        with open(path) as f:
            self.assertTrue(all(json.loads(line) for line in f))       # every line is JSON

    def test_in_memory_saver_forgets_on_restart(self):
        b, _ = build_review()
        b.compile(ft.InMemorySaver()).invoke({"draft": "", "approved": False}, CFG)
        b2, _ = build_review()
        with self.assertRaises(ValueError):
            b2.compile(ft.InMemorySaver()).invoke(ft.Command(resume=True), CFG)

    def test_crash_in_a_node_then_continue_without_redoing_finished_work(self):
        calls = {"ok": 0, "flaky": 0}

        def ok(s):
            calls["ok"] += 1
            return {"msgs": ["ok"]}

        def flaky(s):
            calls["flaky"] += 1
            if calls["flaky"] == 1:
                raise RuntimeError("boom")
            return {"msgs": ["flaky"]}
        b = ft.StateGraph(S3)
        b.add_node("a_ok", ok).add_node("b_flaky", flaky).add_node("end", lambda s: {"msgs": ["end"]})
        b.add_edge(ft.START, "a_ok").add_edge(ft.START, "b_flaky").add_edge("a_ok", "end").add_edge("b_flaky", "end").add_edge("end", ft.END)
        g = b.compile(ft.InMemorySaver())
        with self.assertRaises(RuntimeError):
            g.invoke({"draft": ""}, CFG)
        out = g.invoke(None, CFG)                        # continue from the last checkpoint
        self.assertEqual(out["msgs"], ["ok", "flaky", "end"])
        self.assertEqual(calls, {"ok": 1, "flaky": 2})   # "ok" finished before the crash and was not redone

    def test_continue_on_a_finished_thread_is_an_error(self):
        g = build_linear().compile(ft.InMemorySaver())
        g.invoke({"x": 1}, CFG)
        with self.assertRaises(ValueError):
            g.invoke(None, CFG)


# ----------------------------------------------------------------------------- 3. Harness
def fence(cmd):
    return "I will run this.\n```bash\n%s\n```" % cmd


class TestSplitAndSandbox(TmpDirCase):
    def test_split_commands(self):
        self.assertEqual(ft.split_commands("ls -l | wc -l"), [["ls", "-l"], ["wc", "-l"]])
        self.assertEqual(ft.split_commands("echo 'a b' && cat x; ls"), [["echo", "a b"], ["cat", "x"], ["ls"]])
        self.assertEqual(ft.split_commands("echo hi > out.txt"), [["echo", "hi", ">", "out.txt"]])

    def test_split_refuses_what_it_cannot_judge(self):
        for bad in ["echo `id`", "echo $(id)", "cat <<EOF", "ls\nls", "echo 'unclosed", "   ", "cat <(ls)"]:
            with self.assertRaises(ft.SandboxError, msg=bad):
                ft.split_commands(bad)

    def test_runs_allowed_commands_in_the_work_directory(self):
        sb = ft.Sandbox(self.dir)
        self.assertEqual(sb.run("echo hello > a.txt").returncode, 0)
        obs = sb.run("cat a.txt")
        self.assertEqual((obs.returncode, obs.output), (0, "hello\n"))
        self.assertTrue(os.path.exists(self.path("a.txt")))

    def test_not_on_the_allowlist(self):
        for cmd in ["rm a.txt", "python3 -c 1", "curl http://example.com", "ls && rm x", "echo hi | sh"]:
            with self.assertRaises(ft.SandboxError, msg=cmd):
                ft.Sandbox(self.dir).run(cmd)

    def test_escape_attempts_are_refused(self):
        sb = ft.Sandbox(self.dir)
        for cmd in ["cat /etc/passwd", "cat ../x", "cat a/../../x", "ls ~", "echo $HOME", "echo hi > /tmp/x",
                    "echo hi >> ../x", "cat ./.[.]/x", "ls ..", "cat ${HOME}", "echo $PATH", "cp x /tmp/y"]:
            with self.assertRaises(ft.SandboxError, msg=cmd):
                sb.run(cmd)

    def test_brackets_in_quoted_text_are_fine(self):
        sb = ft.Sandbox(self.dir)
        sb.run("printf '[server]\\nport = 80\\n' > c.ini")
        self.assertEqual(sb.run("cat c.ini").output, "[server]\nport = 80\n")

    def test_every_command_is_a_fresh_process(self):
        sb = ft.Sandbox(self.dir, allow=("cd", "pwd", "mkdir"))
        sb.run("mkdir sub")
        sb.run("cd sub")
        self.assertEqual(sb.run("pwd").output.strip(), os.path.realpath(self.dir))

    def test_environment_is_minimal(self):
        out = ft.Sandbox(self.dir, allow=("env",)).run("env").output
        self.assertIn("PATH=", out)
        self.assertNotIn("HOME=", out)
        os.environ["SECRET_TOKEN_FOR_TEST"] = "x"
        self.addCleanup(os.environ.pop, "SECRET_TOKEN_FOR_TEST", None)
        self.assertNotIn("SECRET_TOKEN", ft.Sandbox(self.dir, allow=("env",)).run("env").output)

    def test_timeout_kills_a_blocked_command_quickly(self):
        os.mkfifo(self.path("pipe"))
        sb = ft.Sandbox(self.dir, timeout=0.5)
        t0 = time.monotonic()
        obs = sb.run("cat pipe")                     # blocks forever: nobody writes to the FIFO
        took = time.monotonic() - t0
        self.assertEqual(obs.returncode, 124)
        self.assertIn("timed out", obs.output)
        self.assertLess(took, 3.0)
        self.assertGreaterEqual(took, 0.4)

    def test_long_output_is_cut_with_a_marker(self):
        with open(self.path("big.txt"), "w") as f:
            f.write("x" * 5000)
        obs = ft.Sandbox(self.dir, max_output=100).run("cat big.txt")
        self.assertEqual(obs.truncated, 4900)
        self.assertIn("[truncated 4900 characters]", obs.output)
        self.assertLess(len(obs.output), 160)

    def test_nonzero_exit_is_reported_not_raised(self):
        obs = ft.Sandbox(self.dir).run("cat missing.txt")
        self.assertNotEqual(obs.returncode, 0)
        self.assertIn("No such file", obs.output)


class TestEventsAndAgent(TmpDirCase):
    def write_cfg(self, text="timout = 30\n"):
        with open(self.path("config.ini"), "w") as f:
            f.write(text)

    def test_messages_are_rebuilt_from_events(self):
        evs = [ft.Event(0, "system", "s"), ft.Event(1, "user", "u"), ft.Event(2, "assistant", "a"),
               ft.Event(3, "observation", "o"), ft.Event(4, "refusal", "r")]
        self.assertEqual([m["role"] for m in ft.rebuild_messages(evs)], ["system", "user", "assistant", "user", "user"])

    def test_events_round_trip_through_jsonl(self):
        evs = [ft.Event(0, "system", "line1\nline2"), ft.Event(1, "user", "q\"uote")]
        self.assertEqual(ft.events_from_jsonl(ft.events_to_jsonl(evs)), evs)

    def test_scripted_model_replays_and_records(self):
        m = ft.ScriptedModel(["one", "two"])
        self.assertEqual((m([{"role": "user", "content": "a"}]), m([])), ("one", "two"))
        self.assertEqual(len(m.seen[0]), 1)
        with self.assertRaises(RuntimeError):
            m([])

    def test_agent_fixes_a_file_and_submits(self):
        self.write_cfg()
        model = ft.ScriptedModel([fence("cat config.ini"), fence("echo 'timeout = 30' > config.ini"), fence("echo TASK_COMPLETE")])
        res = ft.BashAgent(model, ft.Sandbox(self.dir)).run("fix the typo")
        self.assertEqual((res.status, res.steps), ("submitted", 3))
        self.assertEqual(open(self.path("config.ini")).read(), "timeout = 30\n")

    def test_history_is_linear_each_prompt_extends_the_last(self):
        self.write_cfg()
        model = ft.ScriptedModel([fence("cat config.ini"), fence("ls"), fence("echo TASK_COMPLETE")])
        ft.BashAgent(model, ft.Sandbox(self.dir)).run("look around")
        for a, b in zip(model.seen, model.seen[1:]):
            self.assertEqual(b[:len(a)], a)
            self.assertEqual(len(b), len(a) + 2)          # the model's reply and the observation

    def test_wrong_format_is_reported_back_and_the_loop_continues(self):
        model = ft.ScriptedModel(["no block here", fence("ls") + "\n" + fence("ls"), fence("echo TASK_COMPLETE")])
        res = ft.BashAgent(model, ft.Sandbox(self.dir)).run("t")
        self.assertEqual(res.status, "submitted")
        self.assertEqual([e.kind for e in res.events].count("format_error"), 2)
        self.assertIn("found 0", [e.text for e in res.events if e.kind == "format_error"][0])

    def test_refusals_are_shown_to_the_model(self):
        model = ft.ScriptedModel([fence("rm -rf x"), fence("echo TASK_COMPLETE")])
        res = ft.BashAgent(model, ft.Sandbox(self.dir)).run("t")
        self.assertEqual(res.status, "submitted")
        self.assertIn("refused", [e.text for e in res.events if e.kind == "refusal"][0])
        self.assertIn("refused", model.seen[1][-1]["content"])

    def test_step_limit(self):
        model = ft.ScriptedModel([fence("echo %d" % i) for i in range(10)])
        res = ft.BashAgent(model, ft.Sandbox(self.dir), step_limit=4).run("t")
        self.assertEqual((res.status, res.steps), ("step_limit", 4))

    def test_stuck_detector(self):
        model = ft.ScriptedModel([fence("ls")] * 10)
        res = ft.BashAgent(model, ft.Sandbox(self.dir), stuck=3).run("t")
        self.assertEqual((res.status, res.steps), ("stuck", 3))

    def test_different_commands_are_not_stuck(self):
        model = ft.ScriptedModel([fence("ls"), fence("pwd"), fence("ls"), fence("echo TASK_COMPLETE")])
        self.assertEqual(ft.BashAgent(model, ft.Sandbox(self.dir), stuck=2).run("t").status, "submitted")

    def test_model_failure_ends_the_run_cleanly(self):
        res = ft.BashAgent(ft.ScriptedModel([]), ft.Sandbox(self.dir)).run("t")
        self.assertEqual(res.status, "model_error")
        self.assertEqual(res.events[-1].kind, "format_error")


class TestSessionTreeAndTools(TmpDirCase):
    def test_branching(self):
        t = ft.SessionTree("sys")
        a = t.add("user", "task")
        b = t.add("assistant", "plan A")
        t.branch_from(a)
        c = t.add("assistant", "plan B")
        self.assertEqual([m["content"] for m in t.path()], ["sys", "task", "plan B"])
        self.assertEqual([m["content"] for m in t.path(b)], ["sys", "task", "plan A"])
        self.assertEqual(t.leaves(), [b, c])
        with self.assertRaises(KeyError):
            t.branch_from(99)

    def test_four_tools(self):
        ft_ = ft.FileTools(self.dir)
        self.assertEqual(ft_.call("write", path="sub/a.txt", text="one two one"), "wrote 11 characters")
        self.assertEqual(ft_.call("read", path="sub/a.txt"), "one two one")
        self.assertEqual(ft_.call("edit", path="sub/a.txt", old="two", new="2"), "edited")
        self.assertEqual(ft_.call("read", path="sub/a.txt"), "one 2 one")
        self.assertEqual(ft_.call("bash", cmd="cat sub/a.txt").output, "one 2 one")

    def test_edit_needs_exactly_one_match(self):
        t = ft.FileTools(self.dir)
        t.write("a.txt", "x x y")
        for old in ("x", "zzz"):
            with self.assertRaises(ValueError):
                t.edit("a.txt", old, "q")
        self.assertEqual(t.read("a.txt"), "x x y")

    def test_paths_and_names_are_checked(self):
        t = ft.FileTools(self.dir)
        for rel in ("../x.txt", "/etc/passwd", "a/../../x"):
            with self.assertRaises(ft.SandboxError, msg=rel):
                t.read(rel)
        with self.assertRaises(KeyError):
            t.call("delete", path="x")
        os.symlink("/etc", self.path("link"))
        with self.assertRaises(ft.SandboxError):
            t.read("link/hostname")


# ----------------------------------------------------------------------------- 4. Durable execution
def make_runtime(store=None, flaky_fails=0, notes=None):
    """A runtime with: lookup (pure), flaky (fails N times), notify (side effect, logs to notes)."""
    rt = ft.Runtime(store)
    state = {"fails": flaky_fails}
    notes = notes if notes is not None else []

    def lookup(x):
        return {"value": x * 2, "pair": (x, x)}

    def flaky(x):
        if state["fails"] > 0:
            state["fails"] -= 1
            raise IOError("temporary failure")
        return x + 100

    def notify(msg):
        notes.append((msg, ft.activity_info()["key"], ft.activity_info()["attempt"]))
        return "sent"

    def broken(x):
        raise ft.NonRetryable("bad input")

    rt.register_activity("lookup", lookup)
    rt.register_activity("flaky", flaky, ft.RetryPolicy(initial_interval=1, backoff=2, max_attempts=3))
    rt.register_activity("notify", notify)
    rt.register_activity("broken", broken, ft.RetryPolicy(max_attempts=5))
    return rt, notes


def wf_simple(ctx, x):
    return ctx.activity("lookup", x)["value"] + 1


def wf_timer(ctx, x):
    a = ctx.activity("lookup", x)
    ctx.sleep(100)
    b = ctx.activity("notify", "done %d" % a["value"])
    return [a["value"], b]


class TestRetryPolicy(unittest.TestCase):
    def test_delays_double_and_are_capped(self):
        p = ft.RetryPolicy(initial_interval=1, backoff=2, max_interval=5)
        self.assertEqual([p.delay(i) for i in (1, 2, 3, 4)], [1, 2, 4, 5])


class TestDurable(unittest.TestCase):
    def test_simple_workflow_and_its_history(self):
        rt, _ = make_runtime()
        rt.register_workflow("w", wf_simple)
        st = rt.start("a", "w", 5)
        self.assertEqual((st.status, st.result), ("completed", 11))
        self.assertEqual([e["type"] for e in rt.store.for_workflow("a")],
                         ["WorkflowStarted", "ActivityScheduled", "ActivityCompleted", "WorkflowCompleted"])
        self.assertEqual([e["id"] for e in rt.store.for_workflow("a")], [1, 2, 3, 4])
        self.assertEqual(rt.status("a"), "completed")

    def test_results_go_through_json(self):
        rt, _ = make_runtime()
        rt.register_workflow("w", lambda ctx: ctx.activity("lookup", 2))
        self.assertEqual(rt.start("a", "w").result, {"value": 4, "pair": [2, 2]})     # the tuple became a list

    def test_running_a_finished_workflow_runs_nothing_again(self):
        rt, _ = make_runtime()
        rt.register_workflow("w", wf_simple)
        rt.start("a", "w", 5)
        n = len(rt.executions)
        self.assertEqual(rt.run("a").result, 11)
        self.assertEqual(len(rt.executions), n)

    def test_start_errors(self):
        rt, _ = make_runtime()
        rt.register_workflow("w", wf_simple)
        rt.start("a", "w", 1)
        with self.assertRaises(ValueError):
            rt.start("a", "w", 1)
        with self.assertRaises(KeyError):
            rt.start("b", "nope")
        with self.assertRaises(KeyError):
            rt.status("zzz")

    def test_retry_then_success_records_each_failed_attempt(self):
        rt, _ = make_runtime(flaky_fails=2)
        rt.register_workflow("w", lambda ctx: ctx.activity("flaky", 1))
        st = rt.start("a", "w")
        self.assertEqual((st.status, st.result), ("completed", 101))
        types = [e["type"] for e in rt.store.for_workflow("a")]
        self.assertEqual(types.count("ActivityAttemptFailed"), 2)
        self.assertEqual(rt.store.for_workflow("a")[-2]["attempts"], 3)
        self.assertEqual(rt.now, 3.0)                      # waited 1 s then 2 s (virtual time)
        self.assertEqual(rt.executions, [("flaky", 1), ("flaky", 2), ("flaky", 3)])

    def test_out_of_attempts_fails_the_workflow(self):
        rt, _ = make_runtime(flaky_fails=10)
        rt.register_workflow("w", lambda ctx: ctx.activity("flaky", 1))
        st = rt.start("a", "w")
        self.assertEqual(st.status, "failed")
        self.assertIn("after 3 attempt", st.detail)
        self.assertEqual(rt.status("a"), "failed")
        self.assertEqual(rt.store.for_workflow("a")[-2]["type"], "ActivityFailed")
        self.assertEqual(rt.run("a").status, "failed")           # stays failed on replay

    def test_non_retryable_stops_at_once(self):
        rt, _ = make_runtime()
        rt.register_workflow("w", lambda ctx: ctx.activity("broken", 1))
        st = rt.start("a", "w")
        self.assertEqual(st.status, "failed")
        self.assertEqual(rt.executions, [("broken", 1)])

    def test_policy_can_name_non_retryable_types(self):
        rt = ft.Runtime()

        def f():
            raise KeyError("x")
        rt.register_activity("f", f, ft.RetryPolicy(max_attempts=5, non_retryable=(KeyError,)))
        rt.register_workflow("w", lambda ctx: ctx.activity("f"))
        self.assertEqual(rt.start("a", "w").status, "failed")
        self.assertEqual(len(rt.executions), 1)

    def test_timer_suspends_then_completes_without_redoing_activities(self):
        rt, notes = make_runtime()
        rt.register_workflow("w", wf_timer)
        self.assertEqual(rt.start("a", "w", 4).status, "suspended")
        self.assertEqual(rt.run("a").status, "suspended")          # still not time
        rt.advance(99)
        self.assertEqual(rt.run("a").status, "suspended")
        rt.advance(1)
        st = rt.run("a")
        self.assertEqual((st.status, st.result), ("completed", [8, "sent"]))
        self.assertEqual([n for n, _ in rt.executions].count("lookup"), 1)    # lookup ran once in total
        self.assertEqual(len([e for e in rt.store.for_workflow("a") if e["type"] == "TimerStarted"]), 1)

    def test_crash_after_a_recorded_result_does_not_repeat_the_activity(self):
        store = ft.HistoryStore()
        rt, notes = make_runtime(store)
        rt.register_workflow("w", wf_timer)
        rt.crash_at = (3, "after")                         # right after lookup's ActivityCompleted
        with self.assertRaises(ft.SimulatedCrash):
            rt.start("a", "w", 4)
        rt2, _ = make_runtime(store, notes=notes)          # a new worker with the same history
        rt2.register_workflow("w", wf_timer)
        self.assertEqual(rt2.run("a").status, "suspended")      # replay reached the timer
        rt2.advance(100)
        st = rt2.run("a")
        self.assertEqual((st.status, st.result), ("completed", [8, "sent"]))
        self.assertEqual([n for n, _ in rt.executions + rt2.executions].count("lookup"), 1)

    def test_crash_before_the_result_is_recorded_runs_the_activity_again(self):
        store = ft.HistoryStore()
        notes = []
        rt, _ = make_runtime(store, notes=notes)
        rt.register_workflow("w", wf_timer)
        rt.start("a", "w", 4)
        rt.advance(100)
        rt.crash_at = (7, "before")                        # notify ran, but its ActivityCompleted was lost
        with self.assertRaises(ft.SimulatedCrash):
            rt.run("a")
        rt2, _ = make_runtime(store, notes=notes)
        rt2.register_workflow("w", wf_timer)
        rt2.now = rt.now
        self.assertEqual(rt2.run("a").status, "completed")
        self.assertEqual(len(notes), 2)                    # at-least-once: the side effect happened twice
        self.assertEqual(notes[0][1], notes[1][1])         # ...but both calls carried the same idempotency key
        self.assertEqual(notes[0][1], "a:2")

    def test_an_idempotent_effect_applied_once_despite_two_runs(self):
        store, applied = ft.HistoryStore(), set()
        rt = ft.Runtime(store)
        runs = []

        def charge(amount):
            runs.append(amount)
            applied.add(ft.activity_info()["key"])        # the "payment provider" dedupes by key
            return "ok"
        rt.register_activity("charge", charge)
        rt.register_workflow("w", lambda ctx: ctx.activity("charge", 5))
        rt.crash_at = (3, "before")
        with self.assertRaises(ft.SimulatedCrash):
            rt.start("a", "w")
        rt2 = ft.Runtime(store)
        rt2.register_activity("charge", charge)
        rt2.register_workflow("w", lambda ctx: ctx.activity("charge", 5))
        rt2.run("a")
        self.assertEqual((len(runs), len(applied)), (2, 1))

    def test_attempt_numbers_continue_after_a_crash_in_the_retry_loop(self):
        store = ft.HistoryStore()
        rt, _ = make_runtime(store, flaky_fails=1)
        rt.register_workflow("w", lambda ctx: ctx.activity("flaky", 1))
        rt.crash_at = (3, "after")                          # after the first AttemptFailed
        with self.assertRaises(ft.SimulatedCrash):
            rt.start("a", "w")
        rt2, _ = make_runtime(store, flaky_fails=0)
        rt2.register_workflow("w", lambda ctx: ctx.activity("flaky", 1))
        self.assertEqual(rt2.run("a").result, 101)
        self.assertEqual(rt2.executions, [("flaky", 2)])

    def test_history_survives_in_a_file(self):
        path = os.path.join(tempfile.mkdtemp(), "h.jsonl")
        self.addCleanup(shutil.rmtree, os.path.dirname(path), True)
        rt, _ = make_runtime(ft.HistoryStore(path))
        rt.register_workflow("w", wf_simple)
        rt.start("a", "w", 5)
        again = ft.HistoryStore(path)
        self.assertEqual(again.for_workflow("a"), rt.store.for_workflow("a"))
        self.assertEqual(again.append("a", {"type": "X", "t": 0})["id"], 5)


class TestDeterminism(unittest.TestCase):
    def test_changing_the_order_of_commands_is_caught(self):
        store = ft.HistoryStore()
        rt, _ = make_runtime(store)
        rt.register_workflow("w", lambda ctx: (ctx.activity("lookup", 1), ctx.sleep(10), ctx.activity("lookup", 2))[2])
        rt.start("a", "w")
        rt2, _ = make_runtime(store)
        rt2.register_workflow("w", lambda ctx: (ctx.activity("notify", "x"), ctx.sleep(10))[0])      # a different program
        with self.assertRaises(ft.NonDeterminismError):
            rt2.run("a")

    def test_changing_the_arguments_is_caught(self):
        store = ft.HistoryStore()
        rt, _ = make_runtime(store)
        rt.register_workflow("w", lambda ctx: (ctx.activity("lookup", 1), ctx.sleep(10))[0])
        rt.start("a", "w")
        rt2, _ = make_runtime(store)
        rt2.register_workflow("w", lambda ctx: (ctx.activity("lookup", 2), ctx.sleep(10))[0])
        with self.assertRaises(ft.NonDeterminismError):
            rt2.run("a")

    def test_random_in_workflow_code_breaks_replay_but_side_effect_does_not(self):
        import random
        store = ft.HistoryStore()
        rt, _ = make_runtime(store)
        picked = []

        def bad(ctx):
            n = random.choice([1, 2, 3, 4, 5, 6, 7, 8])         # not deterministic
            picked.append(n)
            ctx.activity("lookup", n)
            ctx.sleep(10)
            return n
        rt.register_workflow("w", bad)
        random.seed(123)
        rt.start("a", "w")
        rt.advance(10)
        # pick a seed whose first choice differs from the one in the history
        other = next(s for s in range(100) if random.Random(s).choice([1, 2, 3, 4, 5, 6, 7, 8]) != picked[0])
        random.seed(other)
        with self.assertRaises(ft.NonDeterminismError):
            rt.run("a")

        def good(ctx):
            n = ctx.side_effect(lambda: random.choice([1, 2, 3, 4, 5, 6, 7, 8]))
            ctx.activity("lookup", n)
            ctx.sleep(10)
            return n
        rt3, _ = make_runtime(ft.HistoryStore())
        rt3.register_workflow("w", good)
        first = rt3.start("g", "w")
        rt3.advance(10)
        results = set()
        for seed in range(10):
            random.seed(seed)
            results.add(rt3.run("g").result)
        self.assertEqual(len(results), 1)
        self.assertEqual(first.status, "suspended")

    def test_now_is_recorded_once(self):
        rt, _ = make_runtime()

        def w(ctx):
            t = ctx.now()
            ctx.sleep(50)
            return t
        rt.register_workflow("w", w)
        rt.now = 7
        rt.start("a", "w")
        rt.advance(50)
        self.assertEqual(rt.run("a").result, 7)


class TestSignals(unittest.TestCase):
    def wf(self, ctx):
        a = ctx.wait_for_signal("approve")
        b = ctx.wait_for_signal("approve")
        return [a, b]

    def test_waits_until_a_signal_arrives(self):
        rt, _ = make_runtime()
        rt.register_workflow("w", self.wf)
        self.assertEqual(rt.start("a", "w").status, "suspended")
        rt.signal("a", "approve", "yes")
        self.assertEqual(rt.run("a").status, "suspended")        # the second wait is still open
        rt.signal("a", "approve", "again")
        self.assertEqual(rt.run("a").result, ["yes", "again"])

    def test_a_signal_that_arrives_early_is_kept(self):
        rt, _ = make_runtime()
        rt.register_workflow("w", self.wf)
        rt.register_workflow("v", lambda ctx: (ctx.sleep(5), ctx.wait_for_signal("go"))[1])
        rt.start("a", "v")
        rt.signal("a", "go", {"n": 1})
        rt.advance(5)
        self.assertEqual(rt.run("a").result, {"n": 1})

    def test_other_signal_names_do_not_match_and_each_signal_is_used_once(self):
        rt, _ = make_runtime()
        rt.register_workflow("w", lambda ctx: ctx.wait_for_signal("go"))
        rt.start("a", "w")
        rt.signal("a", "stop", 1)
        self.assertEqual(rt.run("a").status, "suspended")
        rt.signal("a", "go", 2)
        self.assertEqual(rt.run("a").result, 2)
        self.assertEqual(rt.run("a").result, 2)                  # replay gives the same answer

    def test_format_history(self):
        rt, _ = make_runtime()
        rt.register_workflow("w", wf_simple)
        rt.start("a", "w", 1)
        text = ft.format_history(rt.store.for_workflow("a"))
        self.assertEqual(len(text.splitlines()), 4)
        self.assertIn("ActivityCompleted", text)


# ----------------------------------------------------------------------------- 5. Tracing and statistics
class TickClock:
    """Moves by 1 second on every read, so durations are easy to compute by hand."""
    def __init__(self):
        self.t = 0.0

    def __call__(self):
        self.t += 1.0
        return self.t


class TestCost(unittest.TestCase):
    def test_approx_tokens(self):
        self.assertEqual((ft.approx_tokens(""), ft.approx_tokens("abcd"), ft.approx_tokens("abcde")), (1, 1, 2))

    def test_cost_by_hand(self):
        # 100 input tokens at 0.15 USD per million, plus 10 output tokens at 0.60 USD per million
        self.assertAlmostEqual(ft.cost_usd("mock-model", 100, 10), 100 * 0.15 / 1e6 + 10 * 0.60 / 1e6)
        self.assertAlmostEqual(ft.cost_usd("mock-model", 100, 10), 2.1e-5)
        self.assertEqual(ft.cost_usd("mock-model", 0, 0), 0.0)
        with self.assertRaises(KeyError):
            ft.cost_usd("unknown", 1, 1)


class TestTracer(unittest.TestCase):
    def test_nesting_gives_parent_ids_and_one_trace_id(self):
        tr = ft.Tracer(TickClock())
        with tr.span("root") as r:
            with tr.span("child"):
                pass
            with tr.span("child2"):
                pass
        by = {s.name: s for s in tr.spans}
        self.assertEqual([s.name for s in tr.spans], ["child", "child2", "root"])      # spans close inner first
        self.assertEqual(by["child"].parent_id, by["root"].span_id)
        self.assertIsNone(by["root"].parent_id)
        self.assertEqual(len({s.trace_id for s in tr.spans}), 1)

    def test_separate_roots_get_separate_trace_ids(self):
        tr = ft.Tracer(TickClock())
        with tr.span("a"):
            pass
        with tr.span("b"):
            pass
        self.assertNotEqual(tr.spans[0].trace_id, tr.spans[1].trace_id)

    def test_ids_depend_on_the_seed_only(self):
        def ids(seed):
            tr = ft.Tracer(TickClock(), seed=seed)
            with tr.span("a"):
                with tr.span("b"):
                    pass
            return [(s.trace_id, s.span_id) for s in tr.spans]
        self.assertEqual(ids(1), ids(1))
        self.assertNotEqual(ids(1), ids(2))
        self.assertEqual(len(ids(1)[0][0]), 32)
        self.assertEqual(len(ids(1)[0][1]), 16)

    def test_durations_with_a_fake_clock(self):
        tr = ft.Tracer(TickClock())
        with tr.span("outer"):          # start=1
            with tr.span("inner"):      # start=2, end=3
                pass
        by = {s.name: s for s in tr.spans}   # outer end=4
        self.assertEqual((by["inner"].duration, by["outer"].duration), (1.0, 3.0))

    def test_an_exception_marks_the_span_and_is_re_raised(self):
        tr = ft.Tracer(TickClock())
        with self.assertRaises(ZeroDivisionError):
            with tr.span("outer"):
                with tr.span("bad"):
                    1 / 0
        by = {s.name: s for s in tr.spans}
        self.assertEqual((by["bad"].status, by["bad"].attributes["error.type"]), ("ERROR", "ZeroDivisionError"))
        self.assertEqual(by["outer"].status, "ERROR")            # the error passes through the parent too
        self.assertEqual(len(tr.spans), 2)                       # both spans were still recorded
        with tr.span("after"):                                   # and the stack is clean
            pass
        self.assertIsNone(tr.spans[-1].parent_id)

    def test_content_is_off_unless_asked_for(self):
        for capture, expect in [(False, False), (True, True)]:
            tr = ft.Tracer(TickClock(), capture_content=capture)
            lm = ft.TracedLM(scripted_lm(["a: 1"]), tr, "mock-model")
            lm("secret prompt")
            self.assertEqual("gen_ai.input.messages" in tr.spans[0].attributes, expect)

    def test_traced_lm_attributes_and_cost(self):
        tr = ft.Tracer(TickClock())
        lm = ft.TracedLM(scripted_lm(["x" * 40]), tr, "mock-model")
        self.assertEqual(lm("p" * 400), "x" * 40)
        sp = tr.spans[0]
        self.assertEqual(sp.name, "chat mock-model")
        a = sp.attributes
        self.assertEqual((a["gen_ai.operation.name"], a["gen_ai.request.model"]), ("chat", "mock-model"))
        self.assertEqual((a["gen_ai.usage.input_tokens"], a["gen_ai.usage.output_tokens"]), (100, 10))
        self.assertAlmostEqual(a["app.cost_usd"], 2.1e-5)
        self.assertEqual(lm.calls, 1)

    def test_traced_tool(self):
        tr = ft.Tracer(TickClock(), capture_content=True)
        self.assertEqual(ft.traced_tool(tr, "add", lambda a, b: a + b, 2, 3, call_id="c1"), 5)
        sp = tr.spans[0]
        self.assertEqual(sp.name, "execute_tool add")
        self.assertEqual((sp.attributes["gen_ai.tool.name"], sp.attributes["gen_ai.tool.call.id"]), ("add", "c1"))
        self.assertEqual(sp.attributes["gen_ai.tool.call.arguments"], {"args": [2, 3], "kw": {}})

    def test_traced_node_inside_a_graph(self):
        tr = ft.Tracer(TickClock())
        b = ft.StateGraph(S1)
        b.add_node("inc", ft.traced_node(tr, "inc", lambda s: {"x": s["x"] + 1}))
        b.add_edge(ft.START, "inc").add_edge("inc", ft.END)
        with tr.span("invoke_workflow demo"):
            b.compile().invoke({"x": 0})
        self.assertEqual([s.name for s in tr.spans], ["node inc", "invoke_workflow demo"])
        self.assertEqual(tr.spans[0].parent_id, tr.spans[1].span_id)

    def test_agent_run_produces_an_agent_span_with_chat_and_tool_children(self):
        d = tempfile.mkdtemp()
        self.addCleanup(shutil.rmtree, d, True)
        tr = ft.Tracer(TickClock())
        model = ft.ScriptedModel([fence("rm x"), fence("ls"), fence("echo TASK_COMPLETE")])
        ft.BashAgent(model, ft.Sandbox(d), tracer=tr).run("t")
        names = [s.name for s in tr.spans]
        self.assertEqual(names.count("chat mock-model"), 3)
        self.assertEqual(names.count("execute_tool bash"), 2)
        root = tr.spans[-1]
        self.assertEqual((root.name, root.attributes["app.status"], root.attributes["app.steps"]), ("invoke_agent bash-agent", "submitted", 3))
        self.assertTrue(all(s.parent_id == root.span_id for s in tr.spans[:-1]))
        refused = [s for s in tr.spans if s.attributes.get("app.refused")]
        self.assertEqual((len(refused), refused[0].status), (1, "ERROR"))


class TestExportAndView(unittest.TestCase):
    def make(self):
        tr = ft.Tracer(TickClock(), capture_content=True)
        lm = ft.TracedLM(scripted_lm(["x" * 40, "y" * 8]), tr, "mock-model")
        with tr.span("invoke_agent demo", **{"gen_ai.operation.name": "invoke_agent"}):
            lm("p" * 400)
            lm("q" * 40)
            try:
                with tr.span("execute_tool boom", **{"gen_ai.operation.name": "execute_tool"}):
                    raise ValueError("bad")
            except ValueError:
                pass
        return tr

    def test_jsonl_round_trip(self):
        tr = self.make()
        text = ft.export_jsonl(tr.spans)
        rows = [json.loads(l) for l in text.splitlines()]
        self.assertEqual(len(rows), 4)
        self.assertEqual(set(rows[0]), {"trace_id", "span_id", "parent_span_id", "name", "start_time_unix_nano",
                                        "end_time_unix_nano", "attributes", "status", "events"})
        back = ft.load_jsonl(text)
        self.assertEqual([(s.name, s.span_id, s.parent_id, s.status, s.attributes) for s in back],
                         [(s.name, s.span_id, s.parent_id, s.status, s.attributes) for s in tr.spans])
        self.assertAlmostEqual(back[0].duration, tr.spans[0].duration)
        self.assertEqual(ft.load_jsonl(""), [])

    def test_summary_totals_by_hand(self):
        s = ft.summarize(self.make().spans)
        self.assertEqual(s["spans"], 4)
        self.assertEqual(s["errors"], 1)
        self.assertEqual((s["input_tokens"], s["output_tokens"]), (100 + 10, 10 + 2))
        self.assertAlmostEqual(s["cost_usd"], ft.cost_usd("mock-model", 100, 10) + ft.cost_usd("mock-model", 10, 2))
        self.assertEqual(s["by_operation"]["chat"]["count"], 2)
        self.assertEqual(s["by_operation"]["execute_tool"]["errors"], 1)

    def test_viewer_shows_a_tree(self):
        text = ft.render_trace(self.make().spans)
        lines = text.splitlines()
        self.assertTrue(lines[0].startswith("invoke_agent demo"))
        self.assertTrue(lines[1].startswith("  chat mock-model"))
        self.assertIn("tok 100/10", lines[1])
        self.assertIn("$0.000021", lines[1])
        self.assertIn("ERROR ValueError", lines[3])
        self.assertEqual(len(lines), 4)

    def test_viewer_handles_a_partial_trace(self):
        spans = self.make().spans[1:]                    # the root is missing: children show as roots
        self.assertEqual(len(ft.render_trace(spans).splitlines()), 3)


class TestStats(unittest.TestCase):
    def test_bootstrap_ci_basic_properties(self):
        m, lo, hi = ft.bootstrap_ci([1, 1, 0, 1, 0, 1, 1, 1, 0, 1])
        self.assertAlmostEqual(m, 0.7)
        self.assertLessEqual(lo, m)
        self.assertGreaterEqual(hi, m)
        self.assertEqual(ft.bootstrap_ci([1, 1, 0, 1]), ft.bootstrap_ci([1, 1, 0, 1]))     # same seed, same answer

    def test_constant_data_gives_a_point_interval(self):
        self.assertEqual(ft.bootstrap_ci([0.5] * 8), (0.5, 0.5, 0.5))

    def test_more_data_gives_a_tighter_interval(self):
        small = [1, 1, 1, 0, 1, 1, 0, 1, 1, 1]
        big = small * 10
        _, l1, h1 = ft.bootstrap_ci(small)
        _, l2, h2 = ft.bootstrap_ci(big)
        self.assertLess(h2 - l2, h1 - l1)

    def test_empty_input(self):
        with self.assertRaises(ValueError):
            ft.bootstrap_ci([])

    def test_pareto_front(self):
        pts = [("a", 1.0, 0.8), ("b", 2.0, 0.8), ("c", 2.0, 0.9), ("d", 0.5, 0.5), ("e", 3.0, 0.7)]
        self.assertEqual(ft.pareto_front(pts), ["a", "c", "d"])          # b and e are beaten by a (cheaper, no worse)

    def test_pareto_ties_are_both_kept(self):
        self.assertEqual(ft.pareto_front([("a", 1.0, 0.5), ("b", 1.0, 0.5)]), ["a", "b"])
        self.assertEqual(ft.pareto_front([]), [])


class TestDemo(unittest.TestCase):
    def run_demo(self):
        buf = io.StringIO()
        with contextlib.redirect_stdout(buf):
            ft.demo()
        return buf.getvalue()

    def test_demo_runs_and_is_deterministic(self):
        a, b = self.run_demo(), self.run_demo()
        self.assertEqual(a, b)
        for n in range(1, 7):
            self.assertIn("== %d. " % n, a)

    def test_demo_makes_the_teaching_points(self):
        out = self.run_demo()
        self.assertIn("file fixed: True", out)
        self.assertIn("notify ran: 2, effect applied: 2", out)
        self.assertIn("notify ran: 2, effect applied: 1", out)
        self.assertIn("Pareto front (lower cost, higher success): ['lean']", out)


class TestCommandLine(TmpDirCase):
    def run_cli(self, *args):
        import subprocess
        import sys
        here = os.path.dirname(os.path.abspath(ft.__file__))
        return subprocess.run([sys.executable, os.path.join(here, "frameworks_toolkit.py"), *args],
                              capture_output=True, text=True, timeout=60)

    def test_view_prints_a_saved_trace(self):
        tr = ft.Tracer(TickClock())
        lm = ft.TracedLM(scripted_lm(["x" * 40]), tr, "mock-model")
        with tr.span("invoke_agent demo", **{"gen_ai.operation.name": "invoke_agent"}):
            lm("p" * 400)
        path = self.path("spans.jsonl")
        with open(path, "w") as f:
            f.write(ft.export_jsonl(tr.spans))
        r = self.run_cli("view", path)
        self.assertEqual(r.returncode, 0)
        self.assertIn("invoke_agent demo", r.stdout)
        self.assertIn("  chat mock-model", r.stdout)
        self.assertIn("totals: 2 spans, 0 errors, tokens in/out 100/10", r.stdout)

    def test_bad_usage_gives_a_message(self):
        r = self.run_cli("nonsense")
        self.assertEqual(r.returncode, 2)
        self.assertIn("usage", r.stdout)


if __name__ == "__main__":
    unittest.main()

$ python test_frameworks_toolkit.py -v
test_keeps_best_reward_and_stops_early ... ok
test_n_must_be_positive ... ok
test_runs_all_n_when_threshold_is_not_met ... ok
test_skips_bad_replies_and_raises_if_all_bad ... ok
test_a_second_input_on_the_same_thread_continues_the_state ... ok
test_continue_on_a_finished_thread_is_an_error ... ok
test_crash_in_a_node_then_continue_without_redoing_finished_work ... ok
test_file_saver_survives_a_new_process ... ok
test_in_memory_saver_forgets_on_restart ... ok
test_one_checkpoint_per_super_step_plus_the_input ... ok
test_threads_are_separate ... ok
test_bad_usage_gives_a_message ... ok
test_view_prints_a_saved_trace ... ok
test_approx_tokens ... ok
test_cost_by_hand ... ok
test_demo_makes_the_teaching_points ... /workspace/course/frameworks_toolkit.py:1718: ResourceWarning: unclosed file <_io.TextIOWrapper name='/tmp/tmpvowkcu5u/config.ini' mode='r' encoding='UTF-8'>
  print("status:", res.status, "| steps:", res.steps, "| file fixed:", "timeout = 30" in open(os.path.join(d, "config.ini")).read())
ResourceWarning: Enable tracemalloc to get the object allocation traceback
ok
test_demo_runs_and_is_deterministic ... /workspace/course/frameworks_toolkit.py:1718: ResourceWarning: unclosed file <_io.TextIOWrapper name='/tmp/tmptjh7ic0n/config.ini' mode='r' encoding='UTF-8'>
  print("status:", res.status, "| steps:", res.steps, "| file fixed:", "timeout = 30" in open(os.path.join(d, "config.ini")).read())
ResourceWarning: Enable tracemalloc to get the object allocation traceback
/workspace/course/frameworks_toolkit.py:1718: ResourceWarning: unclosed file <_io.TextIOWrapper name='/tmp/tmp36w7ycq5/config.ini' mode='r' encoding='UTF-8'>
  print("status:", res.status, "| steps:", res.steps, "| file fixed:", "timeout = 30" in open(os.path.join(d, "config.ini")).read())
ResourceWarning: Enable tracemalloc to get the object allocation traceback
ok
test_changing_the_arguments_is_caught ... ok
test_changing_the_order_of_commands_is_caught ... ok
test_now_is_recorded_once ... ok
test_random_in_workflow_code_breaks_replay_but_side_effect_does_not ... ok
test_an_idempotent_effect_applied_once_despite_two_runs ... ok
test_attempt_numbers_continue_after_a_crash_in_the_retry_loop ... ok
test_crash_after_a_recorded_result_does_not_repeat_the_activity ... ok
test_crash_before_the_result_is_recorded_runs_the_activity_again ... ok
test_history_survives_in_a_file ... ok
test_non_retryable_stops_at_once ... ok
test_out_of_attempts_fails_the_workflow ... ok
test_policy_can_name_non_retryable_types ... ok
test_results_go_through_json ... ok
test_retry_then_success_records_each_failed_attempt ... ok
test_running_a_finished_workflow_runs_nothing_again ... ok
test_simple_workflow_and_its_history ... ok
test_start_errors ... ok
test_timer_suspends_then_completes_without_redoing_activities ... ok
test_agent_fixes_a_file_and_submits ... /workspace/course/test_frameworks_toolkit.py:763: ResourceWarning: unclosed file <_io.TextIOWrapper name='/tmp/tmp0gzswy_j/config.ini' mode='r' encoding='UTF-8'>
  self.assertEqual(open(self.path("config.ini")).read(), "timeout = 30\n")
ResourceWarning: Enable tracemalloc to get the object allocation traceback
ok
test_different_commands_are_not_stuck ... ok
test_events_round_trip_through_jsonl ... ok
test_history_is_linear_each_prompt_extends_the_last ... ok
test_messages_are_rebuilt_from_events ... ok
test_model_failure_ends_the_run_cleanly ... ok
test_refusals_are_shown_to_the_model ... ok
test_scripted_model_replays_and_records ... ok
test_step_limit ... ok
test_stuck_detector ... ok
test_wrong_format_is_reported_back_and_the_loop_continues ... ok
test_inputs_labels_and_attrs ... ok
test_with_inputs_returns_a_copy ... ok
test_jsonl_round_trip ... ok
test_summary_totals_by_hand ... ok
test_viewer_handles_a_partial_trace ... ok
test_viewer_shows_a_tree ... ok
test_bad_updates ... ok
test_compile_checks ... ok
test_linear_run ... ok
test_node_returning_none_changes_nothing ... ok
test_nodes_get_a_copy_of_the_state ... ok
test_reducer_starts_from_an_empty_list ... ok
test_reducers_are_read_from_annotations ... ok
test_a_loop_hits_the_recursion_limit ... ok
test_conditional_edge_picks_a_branch ... ok
test_fan_out_runs_in_one_super_step_and_merges ... ok
test_mermaid_shows_static_and_conditional_edges ... ok
test_path_map_translates_labels ... ok
test_router_can_return_end ... ok
test_router_to_unknown_node_fails ... ok
test_two_writes_to_a_plain_key_are_refused ... ok
test_copies_the_closest_demo ... ok
test_counts_calls_and_keeps_prompts ... ok
test_overlap_ignores_stop_words ... ok
test_same_seed_same_replies ... ok
test_slip_breaks_format_only_without_demos ... ok
test_interrupt_needs_a_checkpointer_to_resume ... ok
test_interrupt_outside_a_graph_is_an_error ... ok
test_new_input_while_waiting_is_refused ... ok
test_pause_shows_the_question_and_saves_a_checkpoint ... ok
test_resume_finishes_the_run_and_the_node_restarts ... ok
test_resume_needs_a_waiting_thread ... ok
test_the_human_can_say_no ... ok
test_two_interrupts_in_one_node_resume_in_order ... ok
test_a_teacher_can_be_given ... ok
test_bootstrap_does_not_touch_the_student_or_its_lm ... ok
test_bootstrap_traces_every_stage_of_a_two_stage_program ... ok
test_bootstrap_with_always_false_metric_keeps_nothing ... ok
test_bootstrap_with_always_true_metric_stops_at_max ... ok
test_bootstrapped_demos_come_from_passing_runs_only ... ok
test_labeled_few_shot_fills_every_stage_of_a_program ... ok
test_labeled_few_shot_puts_k_demos_and_leaves_original ... ok
test_max_labeled_demos_are_added_after_bootstrapped_ones ... ok
test_more_rounds_never_bootstrap_fewer_examples ... ok
test_optimizing_beats_no_demos_on_average ... ok
test_random_search_returns_the_best_candidate ... ok
test_random_search_ties_pick_the_earliest_candidate ... ok
test_chain_of_thought_adds_reasoning_first ... ok
test_demos_go_into_the_prompt ... ok
test_global_lm_and_own_lm ... ok
test_named_predictors_found_in_nested_modules ... ok
test_needs_an_lm ... ok
test_predict_reads_typed_output ... ok
test_trace_records_predictor_calls ... ok
test_wrong_inputs_raise ... ok
test_coerce ... ok
test_parse_reply_missing_field ... ok
test_parse_reply_typed_and_ignores_noise ... ok
test_render_has_demos_inputs_and_output_fields ... ok
test_delays_double_and_are_capped ... ok
test_branching ... ok
test_edit_needs_exactly_one_match ... ok
test_four_tools ... ok
test_paths_and_names_are_checked ... ok
test_a_signal_that_arrives_early_is_kept ... ok
test_format_history ... ok
test_other_signal_names_do_not_match_and_each_signal_is_used_once ... ok
test_waits_until_a_signal_arrives ... ok
test_bad_signatures ... ok
test_default_and_custom_instructions ... ok
test_parse_simple ... ok
test_parse_typed_and_multiple ... ok
test_prepend_output_keeps_original ... ok
test_brackets_in_quoted_text_are_fine ... ok
test_environment_is_minimal ... ok
test_escape_attempts_are_refused ... ok
test_every_command_is_a_fresh_process ... ok
test_long_output_is_cut_with_a_marker ... ok
test_nonzero_exit_is_reported_not_raised ... ok
test_not_on_the_allowlist ... ok
test_runs_allowed_commands_in_the_work_directory ... ok
test_split_commands ... ok
test_split_refuses_what_it_cannot_judge ... ok
test_timeout_kills_a_blocked_command_quickly ... ok
test_bootstrap_ci_basic_properties ... ok
test_constant_data_gives_a_point_interval ... ok
test_empty_input ... ok
test_more_data_gives_a_tighter_interval ... ok
test_pareto_front ... ok
test_pareto_ties_are_both_kept ... ok
test_empty_devset ... ok
test_evaluate_scores_and_counts_calls ... ok
test_exception_counts_as_a_miss ... ok
test_make_tickets_is_deterministic_and_balanced ... ok
test_agent_run_produces_an_agent_span_with_chat_and_tool_children ... ok
test_an_exception_marks_the_span_and_is_re_raised ... ok
test_content_is_off_unless_asked_for ... ok
test_durations_with_a_fake_clock ... ok
test_ids_depend_on_the_seed_only ... ok
test_nesting_gives_parent_ids_and_one_trace_id ... ok
test_separate_roots_get_separate_trace_ids ... ok
test_traced_lm_attributes_and_cost ... ok
test_traced_node_inside_a_graph ... ok
test_traced_tool ... ok

----------------------------------------------------------------------
Ran 150 tests in 1.568s

OK

5. Discussion questions

Use these in a study group or a class. Try to answer with an example and a source. Give each about 5 to 10 minutes.

1. DSPy's compiler keeps a trace only if the metric passes on the final output [1]. Name a task where a weak metric would let bad traces through. What would you change in the metric?
2. In the GSM8K table, bootstrapping lifts a plain Predict program a lot, and a ChainOfThought program less [1]. What does that suggest about when an optimizer matters most?
3. The lab's bootstrapping was limited by what the zero-shot program already got right. How would a stronger teacher change that? What would you give up in cost or in honesty about what the student can do [1]?
4. LangGraph "does not abstract prompts or architecture", and DSPy abstracts prompts [2] [1]. Which would you pick for a support bot with a review step, and which for a classifier with 20 labeled examples? Why not both?
5. A node restarts from its first line after an interrupt [12]. List three kinds of side effect, and say for each whether it is safe before the interrupt.
6. mini-SWE-agent has bash only, OpenHands has a sandbox and an event stream, and Pi has four tools and no permission system [13] [15] [17]. For a tool that edits a customer's files, which one would you start from, and what would you add?
7. HarnessTax found little change in success but large changes in cost [3]. Its tasks are two open benchmarks that the models may have seen. What kind of harness feature could matter on private, longer tasks and not show up there?
8. Claude Code's first-call context is over 10 times Pi's in the post [3]. Name one reason a larger prompt could still be worth it, and say how you would test that on your own work.
9. Temporal replays workflow code from a log, and activities may run more than once [20] [23]. A teammate says "durable means exactly once". Explain what is true and what is not, using Figure 7.
10. The OpenTelemetry GenAI conventions make prompt and argument content opt-in [25]. When would you turn it on? What would you do about customer data in the prompts?
11. Anthropic suggests starting with direct API calls and a simple pattern [30]. You are on a team that wants a framework "to move fast". What would you ask before agreeing, and what would you measure after a month [31]?
12. The lecture says to run custom domain evals [37]. Pick a product you know. Write three cases for it, one exact-match, one state-based and one that needs a judge. Which is the hardest to trust, and why?

6. Quiz (10 questions)

Pick one answer for each question, then press Check answers. Check answers 0 of 10 answered.

1. A team edits a prompt every week and cannot tell whether the product is getting better. Which of the lecture's three pillars is missing?
   • Data: they need to collect much more training data first
   • System: they need to switch to an agent framework
   • Evaluation: a metric, a set of cases from their own domain, and cost and latency numbers, so that each change can be compared with the last
   • None. Prompt edits are enough, because a better prompt always shows up in the output
2. What is a DSPy signature [1] [4]?
   • A typed declaration of a step's inputs and outputs, such as question -> answer, that says what the step does and leaves the prompt text to a module and an optimizer
   • A cryptographic hash that proves a prompt was not changed
   • The full prompt string, written by hand, including all instructions
   • A decorator that sends a function to a remote worker
3. What does DSPy's BootstrapFewShot do with a training input [1]?
   • It fine-tunes the model weights on every training pair
   • It keeps every trace, correct or not, and relies on the model to ignore the bad ones
   • It asks a human to write a chain of thought for each input
   • It runs a teacher program (by default the zero-shot program), records the traces of every module, and keeps a trace as a demonstration only if the metric accepts the final result
4. In the paper's Table 1 (GPT-3.5, development set), what happens to the ChainOfThought program when it goes from no compilation to one round of bootstrapping [1]?
   • It goes from 24.0 to 64.7
   • It goes from 50.0 to 80.3
   • It stays at 50.0, because chain of thought already uses reasoning
   • It goes from 80.3 to 50.0, because bootstrapping overfits
5. A LangGraph node sends an email, then calls interrupt() to ask for approval. The person approves and the run resumes. What happens [12]?
   • The node continues from the line after interrupt(), so the email is sent once
   • The graph restarts from the first node, and every earlier node runs again
   • The node restarts from its first line, so code before the interrupt runs again and the email can be sent twice unless that step is idempotent or moved after the interrupt
   • Nothing resumes, because an interrupt ends the thread for good
6. A Temporal worker sends an email inside an activity. It crashes after the email is sent but before the result is recorded in the event history. What should you expect, and what is the fix [23] [24]?
   • Temporal guarantees the activity ran exactly once, so no fix is needed
   • The result is read from the email provider's logs and recorded
   • The workflow fails for good, and a person must restart it
   • The activity may run again, so it should be idempotent, for example by sending a key that stays the same across attempts so the email system can drop a repeat
7. In mini-SWE-agent, an action cd sub has no effect on the next action. Why [13]?
   • Each action runs with subprocess.run, so every action is independent and there is no stateful shell session that remembers the directory
   • Bash blocks cd when it is run by a model
   • The agent deletes the directory after each step
   • Only OpenHands is allowed to change directories
8. Which statement matches the HarnessTax post [3]?
   • Harness choice changed task success by about 30 points, and cost hardly moved
   • Claude Code was the cheapest harness for every model
   • Harness choice had little effect on task success on the two benchmarks, but the same model could reach similar success at up to 5 times the cost
   • The study proves that a harness with more tools is always better
9. Why must Temporal workflow code be deterministic, and where does an LLM call belong [22]?
   • Because the model is deterministic, so the code must be as well. LLM calls belong in the workflow
   • Because the workflow is replayed against its event history after a failure, and replay needs the same steps in the same order. LLM calls belong in activities, whose results are recorded and reused
   • Because Temporal forbids all network access, so LLM calls cannot be used
   • Because determinism makes the workflow run faster. LLM calls belong in a timer
10. In the lab, a chat span has a prompt of 400 characters and a reply of 40 characters. Tokens are estimated as characters divided by 4. The made-up prices are $0.15 per million input tokens and $0.60 per million output tokens. What is the span's cost, and does the OpenTelemetry GenAI convention define a cost attribute [25]?
    • $0.000021 (100 × 0.15 / 1,000,000 + 10 × 0.60 / 1,000,000, my own arithmetic), and no, the lab's app.cost_usd is its own name
    • $0.00021, and yes, gen_ai.usage.cost
    • $0.000015, and yes, it is set from the model name
    • $0.0000021, and no cost attribute is allowed in any trace

Answer key

7. Graded assignment and rubric

Title: Orchestrate, replay and measure: extend the toolkit, break it on purpose, and report cost with uncertainty.
Time: 3 to 4 hours. Total: 100 points. Work: solo.

The task

Extend the lab code and write a short memo (three pages at most). Mock models are fine. You may use a real model, DSPy, LangGraph or Temporal if you have access, but say clearly which parts used them. Build your own small task and data for each part. Say how you made them.

Part A. A program and its optimizer (25 points)

• Pick a small task that is not support tickets (for example, sorting short log lines into severity levels). Write 60 labeled examples. Split them into train, dev and a test set that you do not touch until the end [1].
• Write a two-stage program with Predict or ChainOfThought, and a metric. Say why the metric is a fair proxy [4] [5].
• Run BootstrapFewShot with at least two settings and the zero-shot program. Use at least 5 seeds. Report the mean and a bootstrap interval for dev, and the test result for your final choice only [6].
• Add one new feature to the mini library, for example a new optimizer rule, an assertion that retries on a bad format, or a new module. Write at least five unit tests, including an edge case [8].
• Show one compiled prompt and say what the optimizer changed.

Part B. A graph that can pause and survive a crash (25 points)

• Build a graph with at least four nodes, a conditional edge, a reducer on one key, and an interrupt before an irreversible step [10] [12].
• Use FileSaver. Show a run that pauses, a "restart" with a new compiled graph, and a resume. Print the checkpoint list [11].
• Show the interrupt re-run problem: put a counter before interrupt(), then fix it. Report the counts before and after.
• Wrap one step of the same flow as an activity in the durable runtime. Inject a crash after the side effect and before its result is recorded. Show the duplicate, then fix it with the idempotency key from activity_info() [23].
• Write at least five unit tests for your flow, including one that proves the fix.

Part C. A harness comparison (25 points)

• Write at least five small tasks in the style of the config-file task, each with a check that decides pass or fail [13].
• Write two scripted agents that differ in one design choice (for example bash only versus a FileTools style with an edit tool, or a short versus a long system prompt). A real model is welcome as a third.
• Run each agent on each task 3 times. Use the tracer to count tokens and cost per run, with a price table you state and label as made up or real [25].
• Report success and cost per setting with bootstrap intervals, and draw the Pareto front using bootstrap_ci and pareto_front [3] [32].
• State plainly what you can and cannot conclude from your sample, in the spirit of the HarnessTax limits.

Part D. Observability and a design memo (25 points)

• Export one trace from Part C as JSONL and show it with the viewer. Mark the slowest span and the costliest span [25].
• Add one attribute that you would want when a run fails. Say whether it is safe to store in production, and what you would redact [25].
• Write a one-page memo for this decision: a team of three wants to build a ticket-triage agent that can send refunds, with human approval for refunds above a limit. Choose which of DSPy, LangGraph, Temporal and a plain loop you would use for which part. Say what you would not use, and why [30] [31].
• Name three best practices from section 2.10 that your design follows, and one it breaks. Say what you would do about the one you break.
• List three limits of your experiments in this assignment.

What to hand in

• Your frameworks_toolkit.py (or your own file names), your tasks and data, and the tests. All tests must pass.
• A memo in PDF or Markdown with the tables from Parts A to C, the trace from Part D and the design memo.
• A sources list for any reading you cite.

Rubric

Criterion	Full marks	Partial	Points
A. Program and optimizer	Own task, 60 examples, a clean train, dev and test split. Metric justified. Two optimizer settings and zero-shot over 5 or more seeds with intervals. One new feature with five or more passing tests. Compiled prompt shown and explained.	Test set reused for tuning, or one seed only, or no intervals, or thin tests.	25
B. Graph and durability	Four or more nodes, conditional edge, reducer, interrupt. Pause, restart and resume with a file saver. Re-run problem shown and fixed with counts. Crash in the gap shown, duplicate shown, key fix proved by a test.	Missing the restart, or the duplicate is not shown, or the fix has no test.	25
C. Harness comparison	Five or more tasks with checks, two setups, 3 runs each. Tokens and cost from traces with a stated price table. Intervals and a Pareto front. Honest conclusion about the sample.	Fewer tasks, or no intervals, or cost not from traces, or over-claims.	25
D. Observability and design	Trace exported and viewed, slowest and costliest spans marked. Attribute with a privacy note. A clear one-page memo with choices and non-choices. Three practices followed and one broken. Three honest limits.	Vague memo, or no privacy thought, or limits missing.	25
Total			100

Grade bands: 90 to 100 excellent. 75 to 89 good. 60 to 74 needs work. Under 60: redo and resubmit. Plain writing is rewarded: short sentences, no filler. Label your own arithmetic as your own.

8. Glossary

Term	Meaning
Activity (Temporal)	A unit of work that touches the outside world, such as an API call or an LLM call. It is retried by default and its result is recorded in the event history [20] [24].
Agent-computer interface (ACI)	An interface designed for a language-model agent to use a computer, as an IDE is designed for a person. SWE-agent's central idea [14].
Bootstrap (confidence interval)	Resampling the per-task results with replacement many times to see how much the average moves. HarnessTax used 10,000 resamples [3].
BootstrapFewShot	A DSPy optimizer. It runs a teacher on training inputs and keeps traces that pass the metric as demonstrations [1].
Checkpoint	A saved copy of a graph's state after a super-step, stored under a thread id [11].
Compound AI system	A system made of several model calls and other parts, such as retrieval and code, not one call. The lecture's middle level of the "system" pillar.
Determinism (workflow)	The property that workflow code takes the same steps in the same order on replay. It is required by Temporal [22].
Durable execution	Saving a program's progress so a new process can continue after a crash [20] [21].
Event history	Temporal's append-only log of what a workflow run did, used to replay it [21].
Harness	The code around a model that manages its tools, context and task execution [3] [18].
Idempotent	Safe to repeat: running it twice has the same effect as running it once. Usually done with a key that stays the same across retries [23].
Interrupt (LangGraph)	A call that pauses a node and waits for a person. On resume the node restarts from its first line [12].
Module (DSPy)	A reusable strategy for a signature, such as Predict or ChainOfThought [1] [5].
Optimizer (DSPy)	A procedure that takes a program, a metric and a few examples and returns a better program. Once called a teleprompter [1] [6].
Pareto front	The set of options that no other option beats on both cost and success. Used to compare setups [3].
Reducer	A function that says how a node's update is combined with the old value of a state key. The default replaces it [10].
Replay	Running workflow code again against its history, returning recorded results instead of repeating work [21].
Signature (DSPy)	A typed declaration of a step's inputs and outputs, such as question -> answer [1] [4].
Span / trace	A span is one timed operation with attributes. A trace is the tree of spans for one request or run [25].
State graph	A program defined as nodes that update shared state, joined by edges, some of them conditional [10].
Super-step	One round of a graph run. Nodes that run in parallel share a super-step [10].
Workflow (two senses)	In Anthropic's wording, a system where code sets the path through model calls [30]. In Temporal, the deterministic function that holds the control flow of a run [20].

9. References

Numbers in square brackets in the text point here. "(required)" marks the three readings from the course schedule. "(added)" marks sources I found myself. All links were opened on October 28, 2026. Docs pages and blog posts change, so I state what I read and do not quote their numbers beyond what the notes say.

• [1] Khattab, O., Singhvi, A., Maheshwari, P., Zhang, Z., Santhanam, K., Vardhamanan, S., Haq, S., Sharma, A., Joshi, T. T., Moazam, H., Miller, H., Zaharia, M., and Potts, C. "DSPy: Compiling Declarative Language Model Calls into Self-Improving Pipelines." ICLR 2024. arXiv:2310.03714. https://arxiv.org/abs/2310.03714 (required)
• [2] LangChain. "LangGraph overview." LangChain Docs. https://docs.langchain.com/oss/python/langgraph/overview. Docs change often. I read it on October 28, 2026. (required)
• [3] Arena.ai. "HarnessTax: How Much Does the Harness Matter for Coding Agents?" Arena Blog, published September 16, 2026, last updated September 18, 2026. https://arena.ai/blog/coding-agents-harness-tax. The post's citation block names Pan, Yang, Arabzadeh, Chiang, Stoica and Zaharia, and points to the project page https://harnesstax.github.io/. A blog post, not peer reviewed. (required)
• [4] DSPy. "Signatures." DSPy documentation. https://dspy.ai/learn/programming/signatures/ (added)
• [5] DSPy. "Modules." DSPy documentation. https://dspy.ai/learn/programming/modules/ (added)
• [6] DSPy. "DSPy Optimizers (formerly Teleprompters)." DSPy documentation. https://dspy.ai/learn/optimization/optimizers/. I read the docs source text. The site lists version 3.4.1 as current. (added)
• [7] Opsahl-Ong, K., Ryan, M. J., Purtell, J., Broman, D., Potts, C., Zaharia, M., and Khattab, O. "Optimizing Instructions and Demonstrations for Multi-Stage Language Model Programs." arXiv:2406.11695, 2024. https://arxiv.org/abs/2406.11695 (abstract level) (added)
• [8] Singhvi, A., Shetty, M., Tan, S., Potts, C., Sen, K., Zaharia, M., and Khattab, O. "DSPy Assertions: Computational Constraints for Self-Refining Language Model Pipelines." arXiv:2312.13382, 2023. https://arxiv.org/abs/2312.13382 (abstract level) (added)
• [9] Agrawal, L. A., Tan, S., Soylu, D., Ziems, N., Khare, R., Opsahl-Ong, K., Singhvi, A., Shandilya, H., et al. "GEPA: Reflective Prompt Evolution Can Outperform Reinforcement Learning." arXiv:2507.19457, 2025. The arXiv page notes acceptance at ICLR 2026. https://arxiv.org/abs/2507.19457 (abstract level) (added)
• [10] LangChain. "Graph API overview." LangGraph docs. https://docs.langchain.com/oss/python/langgraph/graph-api (added)
• [11] LangChain. "Persistence." LangGraph docs. https://docs.langchain.com/oss/python/langgraph/persistence (added)
• [12] LangChain. "Interrupts." LangGraph docs. https://docs.langchain.com/oss/python/langgraph/interrupts (added)
• [13] SWE-agent project. "mini-SWE-agent: The 100 line AI agent that solves GitHub issues or helps you in your command line." README. https://github.com/SWE-agent/mini-swe-agent. Site: https://mini-swe-agent.com/. The score claim is the project's own. (added)
• [14] Yang, J., Jimenez, C. E., Wettig, A., Lieret, K., Yao, S., Narasimhan, K., and Press, O. "SWE-agent: Agent-Computer Interfaces Enable Automated Software Engineering." NeurIPS 2024. arXiv:2405.15793. https://arxiv.org/abs/2405.15793 (abstract level) (added)
• [15] Wang, X., Li, B., Song, Y., Xu, F. F., Tang, X., Zhuge, M., Pan, J., Song, Y., et al. "OpenHands: An Open Platform for AI Software Developers as Generalist Agents." ICLR 2025. arXiv:2407.16741. https://arxiv.org/abs/2407.16741. Code: https://github.com/OpenHands/OpenHands (added)
• [16] Pi contributors. "pi" (README: "a minimal, extensible agent harness"). https://github.com/earendil-works/pi. The repository moved here from an earlier address in 2026. Site: https://pi.dev (added)
• [17] Zechner, M. "What I learned building an opinionated and minimal coding agent." November 30, 2025. https://mariozechner.at/posts/2025-11-30-pi-coding-agent/ (added)
• [18] Lee, Y., Nair, R., Zhang, Q., Lee, K., et al. "Meta-Harness: End-to-End Optimization of Model Harnesses." arXiv:2603.28052, 2026. https://arxiv.org/abs/2603.28052 (abstract level) (added)
• [19] Yu, S., Chong, D., Nandi, A., Soylu, D., et al. "Shepherd: Enabling Programmable Meta-Agents via Reversible Agentic Execution Traces." arXiv:2605.10913, 2026. https://arxiv.org/abs/2605.10913 (abstract level) (added)
• [20] Temporal Technologies. "Temporal Workflow." Temporal Documentation. https://docs.temporal.io/workflows (added)
• [21] Temporal Technologies. "Event History." Temporal Documentation. https://docs.temporal.io/encyclopedia/event-history (added)
• [22] Temporal Technologies. "Temporal Workflow Definition." Temporal Documentation. https://docs.temporal.io/workflow-definition (added)
• [23] Temporal Technologies. "Activity Definition" and "What is a Temporal Activity?" Temporal Documentation. https://docs.temporal.io/activity-definition, https://docs.temporal.io/activities (added)
• [24] Temporal Technologies. "What is a Temporal Retry Policy?" Temporal Documentation. https://docs.temporal.io/encyclopedia/retry-policies (added)
• [25] OpenTelemetry. "Semantic Conventions for Generative AI." Repository https://github.com/open-telemetry/semantic-conventions-genai, status Development. The older docs page at opentelemetry.io/docs/specs/semconv/gen-ai/ now says the content moved. I read the client inference, agent spans and metrics pages. Names may change. (added)
• [26] Langfuse. "LLM Observability & Application Tracing." Langfuse docs. https://langfuse.com/docs/observability/overview (added)
• [27] Arize AI. "What is Arize Phoenix?" Phoenix docs. https://arize.com/docs/phoenix (added)
• [28] LangChain. "LangSmith Observability." LangSmith docs. https://docs.langchain.com/langsmith/observability (added)
• [29] MLflow. "LLM Tracing and Agent Observability." MLflow docs. https://mlflow.org/docs/latest/genai/tracing/ (added)
• [30] Anthropic. "Building effective agents." Anthropic engineering blog, December 2024. https://www.anthropic.com/engineering/building-effective-agents (the address redirects). A note on the page says much of its tooling landscape has changed since December 2024. (added)
• [31] HumanLayer. "12-factor-agents." GitHub repository and README. https://github.com/humanlayer/12-factor-agents. An opinion piece, not a standard. (added)
• [32] Kapoor, S., Stroebl, B., Siegel, Z. S., Nadgir, N., and Narayanan, A. "AI Agents That Matter." arXiv:2407.01502, 2024. https://arxiv.org/abs/2407.01502 (abstract level) (added)
• [33] OpenAI. "OpenAI Agents SDK." Documentation. https://openai.github.io/openai-agents-python/ (added)
• [34] CrewAI. "CrewAI Documentation." https://docs.crewai.com/ (added)
• [35] LlamaIndex. "Workflows." LlamaIndex documentation. https://developers.llamaindex.ai/python/llamaagents/workflows/ (added)
• [36] Pydantic. "Pydantic AI." Documentation. https://pydantic.dev/docs/ai/overview/. The older address ai.pydantic.dev redirects here. (added)
• [37] Husain, H. "Your AI Product Needs Evals." Hamel's Blog. https://hamel.dev/blog/posts/evals/. The lecture slides cite this blog. (added)
• [38] Husain, H. "Fuck You, Show Me The Prompt." Hamel's Blog (the title is blunt, and it is the post's real title). https://hamel.dev/blog/posts/prompt/ (added)
• [39] METR. "Task-Completion Time Horizons of Frontier AI Models." https://metr.org/time-horizons/. The page says it is no longer actively updated. I use only its definition of a time horizon. (added)
• [40] Yang, D., Ryan, M., and Yang, J. "CS329Z: Engineering AI Agents." Stanford University, Fall 2026. Course site and schedule: https://cs329z.stanford.edu/
• [41] Yang, D., Ryan, M., and Yang, J. "Frameworks & Orchestration." CS329Z lecture slides (Lecture 5, Week 3, October 7), Stanford University. https://web.stanford.edu/class/cs329z/slides/lecture05.pdf. Used only to check topic coverage and to point to what the lecture covers. This pack does not reproduce the slides.
• [42] CS329Z course staff. "lecture5-frameworks." Course code repository for Lecture 5. https://github.com/cs329z/lecture5-frameworks. Reviewed only to see which topics and library versions the lecture uses. This pack uses none of its code and a different example.