from fastcore.test import *
from importlib.resources import files
from aidialog.msg_parts import fmt2hist
from claude_agent_sdk import fork_session
from collections import Counter
import tempfile, shutilant
Read, write, and build Claude Code session transcripts
Find, read, search, and edit Claude Code session transcripts and prompt history on this machine.
Session layout
Claude Code stores each conversation as JSONL at ~/.claude/projects/<dir>/<session-id>.jsonl. <dir> is the resolved project path with every non-alphanumeric character replaced by -, including underscores. On macOS, a project in /tmp/foo therefore appears under -private-tmp-foo. sess_dir and sess_file compute these paths. A transcript is named by the conversation’s first session id. claude --resume and post-compaction restarts can advertise a fresh id in CLAUDE_CODE_SESSION_ID while appending records to the original file, so the environment variable may name an id with no file behind it. cur_sess finds the project’s most recently modified transcript, and resolve_session accepts an id or a /rename title.
Each line is one JSON object. The main fields are type, uuid, parentUuid, sessionId, timestamp, and message. Conversation records are user and assistant; a tool result is a user record containing tool_result blocks. rec_role reports those as tool. The active conversation is the parentUuid chain walked back from the last record by sess_thread. Compaction starts a new chain, while load_sess returns every record. conv_recs drops bookkeeping and rec_txt extracts readable text.
Claude also appends every typed prompt to ~/.claude/history.jsonl, with its project directory and a millisecond timestamp. prompt_hist reads it into filterable rows. This can recover the user’s side when a transcript has been lost or rewritten.
Reading workflow
To find something that was said - an earlier discussion, a decision, work lost to compaction - start with sess2dlg, not with the records. It costs a fraction of a second even on a 50MB transcript, and turns tens of thousands of records into a dialog of a few dozen messages: d.summary() is the map (one sized row per message, a prompt’s reply on its own line), d.find_msgs(pat) the search, view_msg/view_msgs the read. doc(aidialog.dlgskill) covers that layer, and a dialog written with write_ipynb is an ordinary ipynb whose prompt sources carry their replies, so rgapi’s nbrg searches saved sessions across files, replies included.
Work at the record level for surgery, or when a record’s envelope is itself the question: locate with sess_file, cur_sess, or resolve_session; load with load_sess, or load_recs for an arbitrary path; select the active thread with sess_thread and conv_recs; search readable text with sess_search, whose hits each carry their record on .rec; read a slice with show_recs. Prefer sess_search to grepping JSONL, since raw files contain base64 data, signatures, and envelope noise.
Reading functions do not modify transcripts. save_sess, append_sess, fork_sess, and the compaction functions do. Read their docs and inspect the target records before calling them; save_sess replaces a whole session file.
Claude Code stores every conversation as a JSONL transcript, and claude --resume <session-id> rebuilds a conversation from one. It does not care who wrote the file. A transcript assembled by hand, including tool calls that never really ran, resumes like any other. So sessions can be mined for data, saved as templates, or built synthetically to give a fresh session worked examples of tool use already in its context. This module finds, reads, and writes them.
Where sessions live
Each project gets a folder under ~/.claude/projects, named by the project’s absolute path with every character that is not a letter or digit replaced by -. The path is resolved first, which matters on macOS, where /tmp and /var are symlinks into /private. The folder for a project in /tmp/foo is therefore -private-tmp-foo.
sess_dir
def sess_dir(
cwd:NoneType=None, # Project directory; the current directory if None
):The folder where Claude Code keeps session transcripts for the project at cwd
sess_dir()Path('/Users/jhoward/.claude/projects/-Users-jhoward-aai-ws-llmsurgery-nbs')Underscores are replaced too, which is easy to get wrong when sanitizing by hand:
test_eq(sess_dir('/a/b_c').name, '-a-b-c')
test_eq(sess_dir('~'), sess_dir(Path.home()))Claude Code exports a session id as CLAUDE_CODE_SESSION_ID to every process it spawns, shell commands and MCP servers alike. But it identifies the current run, not the conversation: claude --resume and post-compaction restarts put a fresh id in the environment, while records keep appending to the original transcript, which is named by the conversation’s first session id. So the env var names the transcript only while a conversation is on its first run, and can advertise an id that names no file at all. The reliable name for “this conversation” is instead the most recently modified transcript in the project’s session folder: appends keep the live transcript’s mtime freshest through restarts and resumes alike. cur_sess uses that heuristic, keeping the env var only as a fallback for when no transcript exists yet. One caveat: two conversations open on the same project trade the newest spot on every write, so under concurrent sessions pass an explicit sid (clikernel sidesteps this by resolving its host conversation once, at worker spawn, when the spawning conversation’s transcript is freshest).
cur_sess
def cur_sess(
cwd:NoneType=None, # Project directory; the current directory if None
):The current conversation’s session id: the most recent transcript for the project at cwd, else the advertised id
cur_sess()The newest transcript wins; with no transcripts at all, the advertised id is used:
tp = Path(tempfile.mkdtemp())
sd = sess_dir(tp)
sd.mkdir(parents=True)
for i,n in enumerate(['older','newer']):
(sd/f'{n}.jsonl').touch()
os.utime(sd/f'{n}.jsonl', (i,i))
test_eq(cur_sess(tp), 'newer')
test_eq(cur_sess(tempfile.mkdtemp()), os.environ.get('CLAUDE_CODE_SESSION_ID'))sess_file
def sess_file(
sid:NoneType=None, # Session id or unique id prefix; `cur_sess(cwd)` if None
cwd:NoneType=None, # Project directory; the current directory, then all projects, if None
):Path to the transcript of session sid for the project at cwd
Under Claude Code, sess_file() with no arguments is therefore the running conversation’s own transcript, surviving resumes and restarts. Session ids are unique across projects, so when an explicit sid’s file is not in the current directory’s folder, sess_file looks across all project folders, and the defaults work from anywhere, including a notebook kernel whose working directory is not the project root.
test_eq(sess_file('abc', '/a/b_c'), SESSIONS/'-a-b-c'/'abc.jsonl')A sid that names no transcript is treated as an id prefix, so the eight characters that identify a session at a glance are enough to open it. An exact id still resolves without globbing, and a prefix matching more than one transcript raises rather than choosing: silently opening the wrong session is the failure worth spending an error on.
test_eq(sess_file('old', tp), sd/'older.jsonl')
for n in ['dup-a','dup-b']: (sd/f'{n}.jsonl').touch()
with expect_fail(contains='matches 2 sessions'): sess_file('dup', tp)
sess_file('new', tp)load_recs
def load_recs(
path
):Session records read directly from JSONL path
Creating dummy data
The examples below use real Claude Code transcripts rather than approximating its record format. Each fixture builder returns immediately when its target exists, so a normal notebook run is deterministic; delete the generated file or directory before rerunning when the fixture needs refreshing.
ant_data = Path(files('llmsurgery')/'data'/'ant')
def mk_ant_fixture_project(path=None):
"Create the minimal Claude project used by the ant fixtures"
assert ant_data.is_dir(), f'ant fixtures ship as package data; missing {ant_data}'
root = Path(path) if path else ant_data/'project'
skill = root/'.claude/skills/ant-fixture/SKILL.md'
if skill.exists(): return root
skill.parent.mkdir(parents=True, exist_ok=True)
skill.write_text("""---
name: ant-fixture
description: Report the fixed ant fixture fact when asked.
---
When invoked, report exactly: `The ant fixture skill is loaded.`
""")
return rootThe fixture project supplies the smallest useful project environment: one local skill with a fixed response. The fixtures live inside the package (llmsurgery/data/ant) and ship with it, anchored via importlib.resources so any working directory finds them.
mk_ant_fixture_project()Path('data/ant/project')The source fixture is a real Agent SDK session with a deliberately narrow surface. Claude loads the one project skill, runs one Bash command, evaluates one expression through the persistent clikernel MCP server, and finishes with fixed text. This captures genuine skill injection, tool calls, tool results, and Claude Code bookkeeping without unrelated session noise.
async def query_sid(msgs):
"Session id from an Agent SDK message stream"
sid = None
async for m in msgs:
if isinstance(m, ResultMessage): sid = m.session_id
if not sid: raise RuntimeError('Claude did not return a session id')
return sidasync def mk_ant_source(path=None):
"Create the real Claude session used by the ant fixtures"
path = Path(path) if path else ant_data/'source.jsonl'
if path.exists(): return path
root = mk_ant_fixture_project(path.parent/'project')
cmd = shutil.which('clikernel-mcp')
if not cmd: raise FileNotFoundError('clikernel-mcp')
opts = ClaudeAgentOptions(cwd=root.resolve(), model='haiku', system_prompt='Follow the requested tool sequence exactly.',
tools=['Skill','Bash','mcp__clikernel__execute'], allowed_tools=['Bash','mcp__clikernel__execute'],
skills=['ant-fixture'], setting_sources=['project'], strict_mcp_config=True,
mcp_servers=dict(clikernel=dict(type='stdio', command=cmd)), max_turns=8)
prompt = "Use the ant-fixture skill. Then use Bash to run `printf 'bash fixture\\n'`. Then use clikernel to evaluate `6*7`. After all tools finish, reply exactly: fixture complete."
sid = await query_sid(query(prompt=prompt, options=opts))
src = sess_file(sid, root)
if not src.exists(): raise FileNotFoundError(src)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(src.read_bytes())
return pathOn the first run, mk_ant_source performs the model call and copies Claude Code’s transcript into the packaged data/ant/source.jsonl. Later runs reuse that checked-in transcript, while deleting the file explicitly requests a fresh capture against the installed Claude Code and Agent SDK versions.
await mk_ant_source()Path('data/ant/source.jsonl')A real transcript contains both conversation and Claude Code bookkeeping. Counting record types first shows what the file actually contains before we decide which subset each reader needs.
source_path = ant_data/'source.jsonl'
source_recs = load_recs(source_path)
len(source_recs),Counter(source_recs.attrgot('type'))(20,
Counter({'assistant': 8,
'user': 5,
'attachment': 3,
'queue-operation': 2,
'ai-title': 1,
'last-prompt': 1}))The user and assistant records carry the conversation. Attachments inject context such as skill contents; queue operations, titles, and prompt markers are bookkeeping retained in the JSONL but not ordinary chat messages.
The parent chain
sess_thread
def sess_thread(
recs, # Session records, e.g. from `load_sess`
):The records on the active conversation chain, walking parentUuid back from the last record
Claude reconstructs a conversation by starting at the final linked conversation record and walking parentUuid backwards. Some bookkeeping records, including custom-title, carry UUIDs but are not part of that chain, so sess_thread considers only user, assistant, attachment, and system records.
source_thread = sess_thread(source_recs)
len(source_recs),len(source_thread),Counter(source_thread.attrgot('type'))(20, 16, Counter({'assistant': 8, 'user': 5, 'attachment': 3}))The four omitted records are unlinked bookkeeping. The active chain retains conversation records and injected-context attachments, and every consecutive pair must link correctly.
assert all(b.parentUuid==a.uuid for a,b in zip(source_thread,source_thread[1:]))Breaking a parent link makes everything before it unreachable, which mirrors what resume does with abandoned or malformed branches. A cyclic chain (possible only in a corrupt file, e.g. one holding duplicate uuids) raises instead of walking forever - and the writers refuse to create such a file in the first place:
broken = L(obj2dict(r) for r in source_thread)
broken[-2]['parentUuid'] = None
test_eq(len(sess_thread(L(dict2obj(r) for r in broken))), 2)
cyc = L(dict2obj(obj2dict(r)) for r in source_thread)
cyc[-1]['parentUuid'] = cyc[-1]['uuid']
test_fail(lambda: sess_thread(cyc), contains='cycle')rec_txt
def rec_txt(
r, # A session record
):Every readable string in r’s message content, joined, for finding records by text
Forking
sess_id
def sess_id(
recs
):The session id in transcript records recs
The fork fixture is derived from the checked-in source transcript using the Agent SDK’s native fork_session. The source is first installed into the fixture project’s Claude session directory, because the SDK operates on normal Claude Code storage rather than arbitrary JSONL paths.
from claude_agent_sdk import fork_sessionasync def mk_ant_fork(path=None, source=None):
"Create the native Claude fork used by the ant fixtures"
path,source = Path(path) if path else ant_data/'fork.jsonl',Path(source) if source else ant_data/'source.jsonl'
if path.exists(): return path
await mk_ant_source(source)
root = mk_ant_fixture_project(source.parent/'project')
recs = load_recs(source)
sid = sess_id(recs)
live = sess_file(sid, root)
live.parent.mkdir(parents=True, exist_ok=True)
live.write_bytes(source.read_bytes())
forked = fork_session(sid, directory=str(root.resolve()), title='ant fixture fork')
src = sess_file(forked.session_id, root)
if not src.exists(): raise FileNotFoundError(src)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(src.read_bytes())
return pathmk_ant_fork stores the native fork beside it as fork.jsonl. Later runs reuse that fixture; deleting it requests a fresh fork from the installed SDK.
The stored fixture path confirms that the native fork is available for the examples below.
await mk_ant_fork()Path('data/ant/fork.jsonl')Loading both transcripts shows the new session identity and the top-level records produced by the fork.
fork_path = ant_data/'fork.jsonl'
fork_recs = load_recs(fork_path)
source_sid,fork_sid = sess_id(source_recs),sess_id(fork_recs)
source_sid,fork_sid,len(source_recs),len(fork_recs),Counter(fork_recs.attrgot('type'))('e12b4bd3-4f36-4dda-873d-2ff25d1f1044',
'5d2b9564-f7f8-48c3-b611-f33fbe39c577',
20,
17,
Counter({'assistant': 8, 'user': 5, 'attachment': 3, 'custom-title': 1}))The fork has a fresh session ID. Its transcript contains the copied session records and the requested custom-title; unlinked bookkeeping from the captured source is absent.
source_thread,fork_thread = sess_thread(source_recs),sess_thread(fork_recs)
len(source_thread),len(fork_thread),Counter(source_thread.attrgot('type')),Counter(fork_thread.attrgot('type'))(16,
16,
Counter({'assistant': 8, 'user': 5, 'attachment': 3}),
Counter({'assistant': 8, 'user': 5, 'attachment': 3}))The active chains have the same record-type structure. Native forking assigns fresh record UUIDs and rebuilds the parent links under the fork’s session ID.
test_ne(source_sid, fork_sid)
test_eq(source_thread.attrgot('type'), fork_thread.attrgot('type'))
test_eq(set(fork_thread.attrgot('sessionId')), {fork_sid})
test_ne(source_thread.attrgot('uuid'), fork_thread.attrgot('uuid'))
test_eq(fork_thread.attrgot('parentUuid'), [None,*fork_thread.attrgot('uuid')[:-1]])
[(a.uuid,b.uuid) for a,b in zip(source_thread[:3],fork_thread[:3])][('f711b566-2404-41a5-a1b6-36f99d52e391',
'58718dc0-80ed-4298-befc-c933e7913a94'),
('071ddef8-7dfb-42dc-b32c-38506751d50e',
'1e84ec59-9412-44c1-8f12-16aeef9a201c'),
('05dd6db4-a8a1-422a-9617-ea7b19a00873',
'2773c527-c1a1-4816-9a3f-435a9cde9f45')]Record envelopes change, while the readable conversation remains the same. Rendering the fork gives a direct view of the session that later compaction examples will use.
test_eq([rec_txt(r) for r in source_thread], [rec_txt(r) for r in fork_thread])Writing a session
Records are plain dicts, so writing a session comes down to filling the envelope and linking the chain. mk_rec fills the envelope for one message. It writes the optional bookkeeping a real transcript carries (version, gitBranch, userType, permissionMode, and API metadata on assistant records), not only the six required fields: what Claude Code’s LLM side makes of a sparse-but-valid record is close to untestable, so we err towards realistic.
Two records with the same content get different files by default, since ids and timestamps are fresh each call. Sometimes the opposite is wanted: the same history should produce byte-identical records, so the same session id maps to the same file however many times it is rebuilt. canon gives a canonical JSON rendering to hash, and stable_uuid turns any string into a deterministic uuid. fastllm_claude_code.core derives its session and record ids this way, and a session template built from a fixed script can too.
stable_uuid
def stable_uuid(
s
):A uuid deterministically derived from string s
canon
def canon(
o
):Canonical compact JSON for o, key-sorted, for stable hashing
Canonical JSON ignores dictionary insertion order, while stable_uuid maps the same key to the same UUID and different keys to different UUIDs.
test_eq(canon(dict(b=1, a=2)), canon(dict(a=2, b=1)))
test_eq(stable_uuid('x'), stable_uuid('x'))
assert stable_uuid('x') != stable_uuid('y')mk_rec
def mk_rec(
role, # 'user' or 'assistant'
content, # A string, or a list of content blocks
cwd:str='.', # Project directory recorded in the envelope
uid:NoneType=None, # Record uuid; random if None
ts:NoneType=None, # ISO8601 timestamp; the current time if None
model:str='claude-sonnet-4-6', # Recorded in assistant API metadata; None omits it, so resume uses the user's default
input_toks:int=0, # `input_tokens` recorded in assistant usage, e.g. an estimate of the context so far
**kwargs
):A transcript record for one conversation message, ready for save_sess
mk_rec('user', 'Hello!'){'type': 'user',
'uuid': '37c5febb-1af0-4afc-b44a-9df9eaed0fa4',
'parentUuid': None,
'sessionId': None,
'timestamp': '2026-07-17T05:27:38.969Z',
'cwd': '/Users/jhoward/aai-ws/llmsurgery/nbs',
'version': '2.1.206',
'gitBranch': 'HEAD',
'isSidechain': False,
'userType': 'external',
'permissionMode': 'default',
'message': {'type': 'message', 'role': 'user', 'content': 'Hello!'}}Assistant records get deterministic API metadata derived from the record id, and stop_reason reflects a trailing tool call:
tu = [dict(type='tool_use', id='toolu_01', name='probe', input={})]
r = mk_rec('assistant', tu, uid=stable_uuid('demo'), ts='2026-01-01T00:00:00.000Z')
test_eq(r['message']['stop_reason'], 'tool_use')
test_eq(r, mk_rec('assistant', tu, uid=stable_uuid('demo'), ts='2026-01-01T00:00:00.000Z'))
assert 'requestId' in r and 'requestId' not in mk_rec('user', 'hi')
test_eq(r['message']['usage']['output_tokens'], _est_toks(tu))
assert 'model' not in mk_rec('assistant', tu, model=None)['message']
test_eq(mk_rec('user', [dict(type='text', text='hi')])['message']['content'], 'hi') # lone text block: written as a string, as Claude Code does
test_eq(mk_rec('user', 'hi', cwd='~')['cwd'], str(Path.home()))API-side messages carry artifacts a real transcript never contains: cache_control markers, and text-only user turns split into blocks where Claude Code records one plain string. mk_rec scrubs both, so a written session is indistinguishable from one Claude Code recorded itself.
cb = [dict(type='text', text='Hi ', cache_control=dict(type='ephemeral')), dict(type='text', text='there.')]
test_eq(mk_rec('user', cb)['message']['content'], 'Hi there.')
tr = mk_rec('user', [dict(type='tool_result', tool_use_id='t1', content='ok', cache_control=dict(type='ephemeral'))])
test_eq(tr['message']['content'], [dict(type='tool_result', tool_use_id='t1', content='ok')])save_sess assigns a session id, chains each record to the one before, and writes the file where claude --resume will look for it. It re-links parentUuid unconditionally, so it is for writing linear conversations. To copy a session while keeping its branch structure, write the records yourself.
save_sess
def save_sess(
recs, # Records in conversation order, e.g. from `mk_rec`
sid:NoneType=None, # Session id; a fresh uuid if None
cwd:NoneType=None, # Project directory; the current directory if None
ts:NoneType=None, # If given, stamp every record's timestamp: True for the current time, or an ISO8601 string
):Chain recs, write them as session sid for the project at cwd, and return sid
append_sess
def append_sess(
recs, # Records to append, e.g. a munged template round
sid:NoneType=None, # Session to append to; `cur_sess()` if None
cwd:NoneType=None, # Project directory; the current directory if None
ts:NoneType=None, # If given, stamp each appended record's timestamp: True for the current time, or an ISO8601 string
):Chain recs onto the tail of session sid and append them to its transcript, returning sid
append_sess is the mid-life counterpart: it chains new records onto the transcript’s tail without rewriting the existing bytes, so structures already in the file (compaction records, sidechains) are left untouched. It’s how a worked round gets spliced into an existing conversation, e.g. llmdojo’s claudedojo -r refreshing a session after a compaction.
Whole conversations convert in one call: msgs2recs takes Anthropic-style messages (dicts with role and content, like claude_mk_msg or fastllm’s denorm_msgs produce) and builds one deterministic record per message, ready for save_sess. The same messages and key give the same ids, so a rebuilt session file is byte-identical.
msgs2recs
def msgs2recs(
msgs, # Anthropic-style messages: dicts with `role` and `content`
key:str='', # Salt: the same messages and key give the same ids
cwd:str='.', # Project directory recorded in the envelopes
ts:str='2026-01-01T00:00:00.000Z', # Timestamp for every record
model:str='claude-sonnet-4-6', # Recorded in assistant API metadata; None omits it, so resume uses the user's default
**kwargs
):Deterministic transcript records for msgs, one record per message
Converting the same messages with the same key is deterministic. Changing the key changes record ids, while roles and accumulated token estimates follow the message sequence.
den = [dict(role='user', content='Ping?'), dict(role='assistant', content=[dict(type='text', text='Pong.')])]
r1,r2 = msgs2recs(den, 'k'),msgs2recs(den, 'k')
test_eq(canon(r1), canon(r2))
test_ne(r1[0]['uuid'], msgs2recs(den, 'other')[0]['uuid'])
test_eq([r['type'] for r in r1], ['user','assistant'])
test_eq(r1[1]['message']['usage'], dict(input_tokens=1, output_tokens=3, cache_creation_input_tokens=0, cache_read_input_tokens=0))msgs2recs and save_sess compose into the one-call operation a stateless caller wants: file this message list as a session, and return the id to resume. The sid is derived from the messages and a salt, so the same history maps to the same file—replaying a conversation is idempotent rather than littering the project with near-duplicate transcripts.
msgs2sess
def msgs2sess(
msgs, # Anthropic-style messages: dicts with `role` and `content`
key:str='', # Salt: the same messages and key give the same session id
cwd:str='.', # Project directory the session is filed under
extra:NoneType=None, # Records appended after the messages, e.g. from `mk_deferred`
ts:str='2026-01-01T00:00:00.000Z', # Timestamp for every record
model:str='claude-sonnet-4-6', # Recorded in assistant API metadata; None omits it, so resume uses the user's default
):Write msgs as a resumable session with a content-stable id, returning the sid
tp2 = Path(tempfile.mkdtemp())
psid = msgs2sess(den, key='pingpong', cwd=tp2)
test_eq(msgs2sess(den, key='pingpong', cwd=tp2), psid)
back = load_recs(sess_file(psid, tp2))
test_eq(back[-1].message.content[0].text, 'Pong.')
test_eq(sess_thread(back).attrgot('uuid'), back.attrgot('uuid'))Synthetic tool calls
A worked tool call is two records joined by one id: a tool_use block in an assistant record, answered by a tool_result block in the user record that follows. mk_tu and mk_tr build the pair so the ids cannot drift, and tool_turn assembles the full exchange, from request to reply.
tool_turn
def tool_turn(
prompt, # The user request
name, # Tool name
input, # Tool arguments
output, # Tool result
answer, # The assistant's closing text
**kwargs
):A complete synthetic tool-use turn, as four records ready for save_sess
mk_tr
def mk_tr(
tu, # The `tool_use` block being answered
content, # The tool's output
):The tool_result content block answering tu
mk_tu
def mk_tu(
name, # Tool name, as the transcript records it
input:NoneType=None, # Tool arguments
tid:NoneType=None, # tool_use id; random if None
):A tool_use content block
The sample session in the next section is built from exactly one such turn.
A sample session
The smallest useful synthetic history is a tool call that never ran, whose result carries a fact the model could not know any other way. We write it against a scratch project directory.
proj = Path(tempfile.mkdtemp())
sample = tool_turn('Measure the flux please.', 'flux_meter', {}, 'flux: 41.7 kilofinches',
'The flux reading is 41.7 kilofinches.', cwd=proj)
sid = save_sess(sample, cwd=proj)
sid'6885e8fa-4b16-458d-9087-80eeff30e397'Reading a session
A transcript is one JSON object per line. load_sess wraps each in dict2obj so fields read as attributes.
load_sess
def load_sess(
sid:NoneType=None, # Session id; the current session if None
cwd:NoneType=None, # Project directory; the current directory if None
):The records of session sid, as an L of attribute-access dicts
Reading it back gives exactly what we wrote:
back = load_sess(sid, proj)
test_eq(len(back), 4)
test_eq(back[-1].message.content[0].text, 'The flux reading is 41.7 kilofinches.')
test_eq(back[2].message.content[0].tool_use_id, back[1].message.content[0].id)A record carries more than resume strictly needs. Only six fields are required: type, uuid, parentUuid, sessionId, timestamp, and message. The rest is optional bookkeeping. Strip timestamp and the session is not even found. message is shaped exactly as the Anthropic API shapes messages: a role, plus content as either a string or a list of content blocks (text, tool_use, tool_result, thinking). Assistant records in real transcripts also carry API metadata (requestId, message.id, model, usage), and none of it is needed on resume. In particular, synthetic histories work without thinking blocks.
Searching a session
A long conversation accumulates thousands of records, and sess_thread deliberately stops at the last compaction boundary, where the parent chain breaks. Finding where something was said or decided is a different job: search every record’s text, then read the records around the hit.
rec_role
def rec_role(
r, # A session record
):The conversational role of r: a user record carrying tool results counts as tool
conv_recs
def conv_recs(
recs, # Session records, e.g. from `load_sess`
):Just the records carrying conversation messages, dropping Claude Code’s bookkeeping
Content blocks nest—a tool result can itself contain text blocks—so rec_txt walks the whole message and skips bookkeeping fields that would otherwise pollute searches. The captured clikernel result is a user record whose readable text is simply 42:
result_42 = first(r for r in source_thread if rec_txt(r)=='42')
test_eq((result_42.type,rec_role(result_42)), ('user','tool'))conv_recs removes attachments and other bookkeeping. rec_role then separates human user messages from tool results, even though both are stored with record type user:
source_conv = conv_recs(source_thread)
len(source_thread),len(source_conv),Counter(map(rec_role,source_conv))(16, 13, Counter({'assistant': 8, 'tool': 3, 'user': 2}))The 16-record active chain becomes 13 conversation records: two user messages, eight assistant records, and three tool results. The second user message is injected skill content, demonstrating that canonical role and human authorship are not identical.
sess_search
def sess_search(
pat, # Regex to find in conversation text
sid:NoneType=None, # Session id; the current session if None
cwd:NoneType=None, # Project directory; the current directory if None
maxlen:int=160, # Preview characters shown around a hit's first match
recs:NoneType=None, # Already-loaded records; if given, `sid` and `cwd` are ignored
):Search conversation records: SessHits rows, each carrying its own record on .rec, with every searched record on .recs
SessHits
def SessHits(
items:NoneType=None, *rest, use_list:bool=False, match:NoneType=None
):Search hits with a match-centered preview per line, each carrying its record on .rec
The result displays one line per hit—index, role, timestamp, and text around the first match—and retains the searched records on .recs. Searching the captured fixture for the exact clikernel result finds one tool record:
hits = sess_search(r'\b42\b', recs=source_recs)
test_eq([h.role for h in hits], ['tool'])
test_eq(hits.attrgot('role'), ['tool']) # `SessHits` is an `L`, so selections keep the row display and `.recs`
test_eq((type(hits[:1]).__name__, hits[:1].recs), ('SessHits', hits.recs))
hits 10 tool 2026-07-17T02:51 42show_recs
def show_recs(
recs, # Session records, e.g. a slice of `SessHits.recs`
mx:int=500, # Characters of text shown per record
showall:bool=False, # Include bookkeeping records?
):A readable transcript of records in recs; conversation records only unless showall
show_recs is the reading half of the workflow: search, take a slice of .recs around a hit, and render it as a conversation. It is deliberately tolerant where recs2chat is strict—any stretch of any transcript renders, with long records truncated:
s = show_recs(hits.recs[hits[0].i-2:hits[0].i+3], mx=100)
assert s.count('---\n')==5 and '\n42' in s
s--- assistant 2026-07-17T02:51:37 ---
The user is providing me with context about the ant-fixture skill. They're telling me that when invo…[+884 chars]
--- assistant 2026-07-17T02:51:37 ---
mcp__clikernel__execute
6*7
--- tool 2026-07-17T02:51:38 ---
42
--- assistant 2026-07-17T02:51:40 ---
The user is providing me with MCP server instructions for clikernel. They're confirming that I shoul…[+249 chars]
--- assistant 2026-07-17T02:51:40 ---
fixture completeshowall=True also renders bookkeeping records. A synthetic compact boundary demonstrates the label used for a non-conversation record.
sysr = dict(type='system', subtype='compact_boundary', timestamp='2026-01-01T00:00:00.000Z')
sa = show_recs([source_conv[0],sysr], showall=True)
assert sa.count('---\n')==2 and 'system:compact_boundary' in sa
sa--- user 2026-07-17T02:51:28 ---
Use the ant-fixture skill. Then use Bash to run `printf 'bash fixture\n'`. Then use clikernel to evaluate `6*7`. After all tools finish, reply exactly: fixture complete.
--- system:compact_boundary 2026-01-01T00:00:00 ---Prompt history
prompt_hist
def prompt_hist(
project:NoneType=None, # Only prompts for the project at this directory; all projects if None
since:NoneType=None, # Only prompts at or after this datetime or ISO string; a TZ-less string is on the same local clock as the returned times, Z/offset strings are converted
pat:NoneType=None, # Only prompts whose text matches this regex
path:NoneType=None, # History file; Claude Code's own (`~/.claude/history.jsonl`) if None
):Claude Code’s global prompt history, oldest first: every prompt typed on this machine, kept even when its transcript is gone
PromptHist
def PromptHist(
*args, **kwargs
):Prompt-history rows in local time, one line per prompt
Transcripts are per-project and can be rewritten or deleted, but Claude Code also appends every prompt to one global file, ~/.claude/history.jsonl, with a millisecond timestamp and the project it was typed in. That file is the recovery source for the user side of any conversation whose transcript is gone. prompt_hist reads it into filterable rows whose repr shows one prompt per line.
Times are the machine’s local clock: rows display and compare in local time, and a TZ-less since string means that same clock, so a time read off the rows works as a filter unchanged. A timestamp copied from a transcript record instead is UTC with a Z suffix; those, and any other offset-carrying string, are converted to local before comparing.
hf = Path(tempfile.mkdtemp())/'history.jsonl'
hp = [dict(display='Fix the table widths', timestamp=1700000000000, project='/a/xhtml'), # chkstyle: ignore-node
dict(display='Now the CSS side', timestamp=1700000060000, project='/a/xhtml'),
dict(display='Release it', timestamp=1700000120000, project='/b/other')]
hf.write_text(''.join(json.dumps(r)+'\n' for r in hp))
test_eq(len(prompt_hist(path=hf)), 3)
test_eq(len(prompt_hist('/a/xhtml', path=hf)), 2)
test_eq(len(prompt_hist(since=datetime.fromtimestamp(1700000060), path=hf)), 2)
test_eq(len(prompt_hist(since=str(datetime.fromtimestamp(1700000060)), path=hf)), 2)
test_eq(len(prompt_hist(since='2023-11-14T22:14:20Z', path=hf)), 2)
test_eq(prompt_hist(pat='CSS', path=hf)[0].display, 'Now the CSS side')
prompt_hist(path=hf)Curating a captured session
Thinking blocks are usually the bulk of a captured assistant turn, and resume rebuilds a conversation without them. is_think_rec identifies records containing only thinking, while strip_think removes them; the fixture deliberately contains such records so this example proves that real work is removed.
strip_think
def strip_think(
recs, # Session records
):Drop records whose message content is only thinking blocks; resume does not need them
is_think_rec
def is_think_rec(
r, # Session record
):Whether r is an assistant record containing only thinking blocks
The captured fixture contains thinking-only records. Stripping them removes part of the active thread and leaves no thinking-only record behind.
source_clean = strip_think(source_thread)
assert 0<len(source_clean)<len(source_thread)
assert not source_clean.filter(is_think_rec)
len(source_thread),len(source_clean),Counter(source_clean.attrgot('type'))(16, 13, Counter({'user': 5, 'assistant': 5, 'attachment': 3}))After thinking, tool traffic is the bulk: file dumps in results, whole files in edit-tool inputs. trunc_tools caps every string inside tool_use inputs and tool_result content, leaving structure intact and marking each cut with the count of characters dropped. Records come back as fresh dicts, originals untouched.
trunc_tools
def trunc_tools(
recs, # Session records
mx:int=2000, # Maximum characters per string in tool inputs and results
):Copies of recs with strings in tool_use inputs and tool_result content truncated to mx characters
A big tool result shrinks to the cap plus a marker; the small input alongside it, and the original records, are untouched:
The fixture keeps tool traffic deliberately small, but a tight cap still shows the operation on genuine records. The first changed strings retain their opening text and report how many characters were removed:
source_trunc = trunc_tools(source_thread, 20)
[(i,rec_txt(a),rec_txt(b)) for i,(a,b) in enumerate(zip(source_thread,source_trunc)) if rec_txt(a)!=rec_txt(b)][(5, 'Launching skill: ant-fixture', 'Launching skill: ant…[+8 chars]'),
(8,
"Bash\nprintf 'bash fixture\\n'\nPrint bash fixture message",
"Bash\nprintf 'bash fixture…[+3 chars]\nPrint bash fixture m…[+6 chars]")]The real fixture demonstrates ordinary truncation, while a deliberately large synthetic result verifies the exact cap and that trunc_tools leaves its input untouched.
lt = tool_turn('Read it all.', 'reader', dict(path='/tmp/big.txt'), 'line\n'*500, 'Long.')
tr = trunc_tools(lt, 100)
assert rec_txt(tr[2]).endswith('…[+2400 chars]')
test_eq(tr[1]['message']['content'][0]['input'], dict(path='/tmp/big.txt'))
test_eq(lt[2]['message']['content'][0]['content'], 'line\n'*500)Captured ids are random, so saving the same curated span twice gives two files that differ throughout. reid_recs derives record, tool, and API ids from position and a salt instead, while preserving tool-call pairings and leaving the captured records untouched.
reid_recs
def reid_recs(
recs, # Records in conversation order
key:str='', # Salt: the same records and key give the same ids
ts:NoneType=None, # If given, set every record's timestamp to this
):Deterministically re-derive record uuids, tool_use ids, and API metadata, so one capture gives one file
Re-identifying the captured fixture twice with one key is deterministic; a different key changes the ids, tool results still answer the correct calls, and the source records are untouched:
r1,r2 = reid_recs(source_clean, 'fixture'),reid_recs(source_clean, 'fixture')
test_eq(canon(list(r1)), canon(list(r2)))
test_ne(r1[0]['uuid'], reid_recs(source_clean, 'other')[0]['uuid'])
test_ne(r1[0]['uuid'], source_clean[0].uuid)With the timestamps pinned too, the whole file is byte-for-byte reproducible. A captured record can also carry a session_id field alongside sessionId; save_sess keeps both in step:
t1,t2 = reid_recs(sample, 'tmpl', ts='2026-01-01T00:00:00.000Z'),reid_recs(sample, 'tmpl', ts='2026-01-01T00:00:00.000Z')
t1[0]['session_id'] = t2[0]['session_id'] = 'stale'
t1[0]['nested'] = t2[0]['nested'] = L([L(1,2)])
tid = save_sess(t1, stable_uuid('tmpl'), proj)
b = sess_file(tid, proj).read_bytes()
test_eq(save_sess(t2, stable_uuid('tmpl'), proj), tid)
test_eq(sess_file(tid, proj).read_bytes(), b)
test_eq(load_sess(tid, proj)[0].session_id, tid)
test_eq(sess_file(tid, proj).read_jsonl()[0]['nested'], [[1,2]])Appending chains onto the existing tail and leaves the prior bytes untouched:
apsid = save_sess(reid_recs(sample, 'apbase'), stable_uuid('append-base'), proj)
with sess_file(apsid, proj).open('a') as f: f.write(json.dumps(dict(type='last-prompt', lastPrompt='hi'))+'\n')
more = tool_turn('And the humidity?', 'hygro', {}, '41%', 'Humid too.')
more[0]['nested'] = L(1,2)
test_eq(append_sess(more, apsid, proj), apsid)
apl = load_sess(apsid, proj)
test_eq(len(apl), 9)
test_eq(apl[5].parentUuid, apl[3].uuid)
test_eq([r.sessionId for r in apl[-4:]], [apsid]*4)
test_eq(apl[-4].nested, [1,2])
test_fail(lambda: save_sess(sample+sample, cwd=proj), contains='duplicate')
test_fail(lambda: append_sess(more, apsid, proj), contains='duplicate')Session names
Claude Code identifies sessions by UUID, while /rename appends a human-readable title, so we add naming and lookup before using names in forks.
name_sess
def name_sess(
sid, name, cwd:NoneType=None
):Name session sid and return it
A rename consists of custom-title and agent-name records appended to the transcript. The conversation chain is unchanged because these records are session bookkeeping rather than messages.
named_sid = save_sess(reid_recs(sample, 'named'), stable_uuid('named'), proj)
named_name = 'flux-demo'
test_eq(name_sess(named_sid, named_name, proj), named_sid)
named_tail = load_sess(named_sid, proj)[-2:]
test_eq(named_tail.attrgot('type'), ['custom-title','agent-name'])
test_eq([named_tail[0].customTitle,named_tail[1].agentName], [named_name]*2)sess_by_name
def sess_by_name(
name, cwd:str='.'
):Find the first project session transcript with latest custom title name
A transcript may contain several rename records. Searching each file backwards makes the latest custom-title authoritative, matching the name Claude currently presents.
test_eq(sess_by_name(named_name, proj), sess_file(named_sid, proj))
name_sess(named_sid, 'flux-renamed', proj)
test_eq(sess_by_name(named_name, proj), None)
test_eq(sess_by_name('flux-renamed', proj), sess_file(named_sid, proj))fork_sess puts the whole diet on one line: load a session, optionally drop thinking and truncate tool traffic, and write the result under a fresh id, leaving the original untouched. Resume the returned id to compare the munged conversation against the original. With a key, ids re-derive deterministically, so re-running the same munge overwrites its fork instead of accumulating copies.
fork_sess
def fork_sess(
sid:NoneType=None, # Session id to fork; `cur_sess()` if None
cwd:NoneType=None, # Project directory; passed to `sess_file` via `load_sess`
mx:NoneType=None, # If given, truncate tool input/output strings to `mx` characters
think:bool=True, # Keep thinking records?
key:NoneType=None, # If given, record and session ids re-derive deterministically from this salt
name:NoneType=None, # Optional name for the new session
):Write a munged copy of session sid under a fresh id, returning the new id to resume
Forking the sample with a thinking record appended and a tight cap: the fork loses the thinking record and truncates the tool result, the original keeps all five records, and the same call gives the same fork id:
think = mk_rec('assistant', [dict(type='thinking', thinking='private', signature='sig')])
tsid = save_sess(reid_recs([*sample,think], 'munge'), stable_uuid('munge'), proj)
fkid = fork_sess(tsid, proj, mx=10, think=False, key='fork1')
test_eq(len(load_sess(fkid, proj)), 4)
assert rec_txt(load_sess(fkid, proj)[2]).endswith('…[+12 chars]')
test_eq(len(load_sess(tsid, proj)), 5)
test_eq(fork_sess(tsid, proj, mx=10, think=False, key='fork1'), fkid)
z0 = fork_sess(tsid, proj, mx=0, key='fork0')
assert rec_txt(load_sess(z0, proj)[2]).endswith('chars]') # mx=0 truncates to nothing; only None disablesfork_sess can name the new session in the same operation. A deterministic key keeps this example idempotent across notebook runs.
fork_name = 'flux-fork'
fork_named_sid = fork_sess(sid, proj, key='named-fork', name=fork_name)
test_eq(sess_by_name(fork_name, proj), sess_file(fork_named_sid, proj))
fork_named_sid'e9f38eac-89ba-5bed-bc1e-9e4f90caf6c6'From dialogs
An authored dialog converts to a session in two steps: dlg2chat (in hist) recovers the canonical messages with real tool calls, and denorm_msgs from fastllm.anthropic shapes them for Claude. dlg2sess then writes the records. The session id derives from the dialog name, so re-converting an edited dialog overwrites its session rather than accumulating copies.
dlg2sess
def dlg2sess(
dlg, # The dialog to convert
cwd:NoneType=None, # Project directory for the session; the current directory if None
key:str='dlg2sess', # Salt for deterministic record ids
aim_info:NoneType=None, # Model capability dict; images enabled if None
):Write dlg as a Claude Code session for the project at cwd, returning the session id to resume; tagged raw messages re-emit their original records
dlg2msgs
def dlg2msgs(
dlg, # A `Dialog`, ending with a prompt
aim_info:NoneType=None, # Model capability dict for media handling; images enabled if None
):Anthropic-style messages for dlg, with each reply’s tool calls recovered as real blocks
A dialog whose reply used a tool, with a referenced image attachment, exercises the whole path. The tool call comes back as paired tool_use/tool_result blocks, and the attachment becomes a base64 image block, both exactly as real transcripts carry them:
def tool_dtl(func, args, result):
"A tool-call wire block in the reply format `fmt2hist` parses"
d = json.dumps(dict(id='call1', name=func, args=args, result=result))
return f"```json {{.tool}}\n{d}\n```"
png = tiny_png
fdlg = Dialog(name='flux')
fatt = Attachment(png, 'image/png')
fdlg.mk_message(f'The rig: ', msg_type=snote, attachments=[fatt])
freply = f"Let me check.\n\n{tool_dtl('flux_meter', {'unit':'kf'}, 'flux: 41.7 kilofinches')}\n\nThe flux is 41.7 kilofinches."
fdlg.mk_message('Measure the flux please.', msg_type=sprompt, output=freply)
fmsgs = dlg2msgs(fdlg)
[(m['role'], [b['type'] for b in m['content']] if isinstance(m['content'], list) else 'str') for m in fmsgs][('user', ['text', 'image', 'text', 'text']),
('assistant', ['text', 'tool_use']),
('user', ['tool_result']),
('assistant', ['text'])]The converted messages preserve the four-part tool exchange, pair the tool result with its call id, and retain the image as a PNG block.
test_eq([m['role'] for m in fmsgs], ['user','assistant','user','assistant'])
ftu = first(b for b in fmsgs[1]['content'] if b['type']=='tool_use')
ftr = first(b for b in fmsgs[2]['content'] if b['type']=='tool_result')
test_eq(ftr['tool_use_id'], ftu['id'])
fimg = first(b for b in fmsgs[0]['content'] if b['type']=='image')
test_eq(fimg['source']['media_type'], 'image/png')Roundtrip and determinism, into the same scratch project as the sample session:
fsid = dlg2sess(fdlg, proj)
fback = load_sess(fsid, proj)
test_eq(len(fback), 4)
test_eq(sess_thread(fback).attrgot('uuid'), fback.attrgot('uuid'))
test_eq(dlg2sess(fdlg, proj), fsid)And the live proof, resuming the dialog-built session and asking about the planted fact (spends tokens, so out of CI):
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessageopts = ClaudeAgentOptions(resume=fsid, cwd=str(proj), model='haiku')
async for m in query(prompt='What did the flux_meter tool report, exactly?', options=opts):
if isinstance(m, ResultMessage): print(m.result)The flux_meter tool reported exactly:
```
flux: 41.7 kilofinches
```Back to dialogs
The reverse direction turns a recorded session into a dialog, ready for editing before it becomes a template. recs2chat normalizes conversation records to canonical fastllm messages: tool calls and results share one id, tool-result records take the tool role, and injected skill text remains a user message.
Claude Code tool results can contain tool_reference blocks (deferred-tool listings), a block type no provider wire format has. Registering a Part subclass is all it takes for the canonical machinery to carry and render it – norm_tr_parts dispatches unknown tags through the registry, and ctext/formatted render per class:
ToolReference.formatted
def formatted():Call self as a function.
ToolReference
def ToolReference(
tool_name:str=''
)->None:A deferred-tool reference in a Claude Code tool result
recs2chat
def recs2chat(
recs, # Session records, e.g. from `load_sess`
):Canonical fastllm messages for the conversation records in recs
On the captured fixture, removing thinking leaves ten canonical messages. The skill invocation, Bash command, and clikernel execution survive as paired tool calls and results:
source_msgs = recs2chat(source_clean)
test_eq(Counter(m.role for m in source_msgs), Counter(user=2,assistant=5,tool=3))
parts = L(source_msgs).flatmap(lambda m:m.content)
tus = parts.filter(lambda p:isinstance(p, ToolUse))
trs = parts.filter(lambda p:isinstance(p, ToolResult))
test_eq(tus.attrgot('name'), ['Skill','Bash','mcp__clikernel__execute'])
test_eq(tus.attrgot('id'), trs.attrgot('id'))
test_eq(source_msgs[-1].text, 'fixture complete')Claude Code can switch model mid-response (capacity fallback), and it records that switch as a fallback block inside the assistant message’s content. The block carries no conversational content, so recs2chat skips it; genuinely unknown block types still raise, since a new block type may need real handling:
fb = dict(type='fallback', to=dict(model='claude-opus-4-8'), **{'from': dict(model='claude-fable-5')})
frec = dict(type='assistant', message=dict(role='assistant', content=[fb, dict(type='text', text='Continuing after fallback.')]))
fmsgs = recs2chat([frec])
test_eq(len(fmsgs), 1)
with expect_fail(ValueError, 'unsupported assistant block'): recs2chat([dict(type='assistant', message=dict(role='assistant', content=[dict(type='mystery')]))])
fmsgs[0].textEach canonical message is stamped with provenance meta: its source record’s created timestamp and uid. chat2dlg turns that into cell metadata and a stable id, so converting the same records twice gives identical message ids – what lets a regenerated session dialog be cited across regenerations:
mr = msgs2recs([dict(role='user', content='Hi'), dict(role='assistant', content=[dict(type='text', text='Hello')])])
cm = recs2chat(mr)
test_eq(cm[0].meta, dict(created=mr[0]['timestamp'], uid=mr[0]['uuid']))
mdlg1,mdlg2 = chat2dlg(cm, 'm1'), chat2dlg(recs2chat(mr), 'm2')
test_eq(mdlg1.messages[0].id, mdlg2.messages[0].id)
test_eq(mdlg1.messages[0].meta['uid'], mr[0]['uuid'])
cm[0].metaA tool result can carry image blocks (screenshot-returning tools); they render as a <media> placeholder rather than raising, while genuinely unknown block types still do:
irb = dict(content=[dict(type='text', text='Screenshot taken:'), dict(type='image', source=dict(type='base64', media_type='image/png', data='aGk='))])
test_eq(_tr_txt(irb), 'Screenshot taken:\n<media input_image image/png>')
test_eq(_tr_txt(dict(content=[dict(type='tool_reference', tool_name='Bash')])), '<tool_reference tool="Bash"/>')
with expect_fail(KeyError): _tr_txt(dict(content=[dict(type='mystery')]))Re-identification must preserve each tool call/result pairing even though both ids change.
rmsgs = recs2chat(r1)
rparts = L(rmsgs).flatmap(lambda m:m.content)
rtus = rparts.filter(lambda p:isinstance(p, ToolUse))
rtrs = rparts.filter(lambda p:isinstance(p, ToolResult))
test_eq(rtus.attrgot('id'), rtrs.attrgot('id'))chat2dlg (in aidialog.hist) builds one prompt per user turn. Prompt content is the user turn verbatim: text parts are joined by blank lines, and images become attachments with references left in the text. Injected skill contents therefore form a second prompt in this real session.
source_dlg = chat2dlg(source_msgs, 'ant fixture')
test_eq(len(source_dlg.messages), 2)
test('Use the ant-fixture skill.', source_dlg.messages[0].content, in_)
test('Base directory for this skill:', source_dlg.messages[1].content, in_)
replies = '\n'.join(m.ai_res for m in source_dlg.messages)
for name in ('Skill','Bash','mcp__clikernel__execute'): test(name, replies, in_)
test(source_dlg.messages[-1].ai_res, 'fixture complete', str.endswith)Tool rendering is normally capped for readability. Passing mx=None preserves a large result in full.
big = 'z'*9999
bdlg = chat2dlg(recs2chat(tool_turn('Big!', 'probe', {}, big, 'Done.')), 'untruncated', mx=None)
assert big in bdlg.messages[0].ai_resFor a canonical turn with one tool call, rendering it as a dialog reply and parsing that reply recovers exactly the content it started from. Provenance meta is user-side only: assistant and tool messages fold into the reply text, where it has no representation:
round_msgs = recs2chat(tool_turn('Measure it.', 'probe', {}, '41.7', 'Done.'))
round_dlg = chat2dlg(round_msgs, 'roundtrip')
test_eq(fmt2hist(round_dlg.messages[0].ai_output), [Msg(m.role, m.content) for m in round_msgs[1:]])A pasted image survives the full circle: it becomes an attachment on the prompt, and converting the dialog onwards to messages yields the same base64 block the record held.
ib = [dict(type='text', text='The rig:'), dict(type='image', source=dict(type='base64', media_type='image/png', data=base64.b64encode(tiny_png).decode()))]
irecs = msgs2recs([dict(role='user', content=ib), dict(role='assistant', content=[dict(type='text', text='Nice rig.')])])
idlg = chat2dlg(recs2chat(irecs), 'rig')
test_eq(idlg.messages[0].attachments[0].data, tiny_png)
assert f'attachment:{idlg.messages[0].attachments[0].id}' in idlg.messages[0].content
test_eq(first(b for b in dlg2msgs(idlg)[0]['content'] if b['type']=='image')['source']['data'], base64.b64encode(tiny_png).decode())sess2dlg
def sess2dlg(
sid:NoneType=None, # Session id; `cur_sess()` if None
cwd:NoneType=None, # Project directory; passed to `sess_file` via `load_sess`
name:NoneType=None, # Dialog name; the session id if None
mx:int=2000, # Maximum characters per rendered tool input/output string; None disables truncation (see `hist2fmt`)
recs:NoneType=None, # Already-loaded records; if given, `sid` and `cwd` are ignored
):The conversation of a session as a dialog, one prompt per user turn; system records ride along as tagged raws
sess2dlg composes the whole read path: load or accept records, walk the parent chain, drop thinking records, and convert. Reading the captured fixture gives the same two-prompt dialog built from its canonical messages above:
rdlg = sess2dlg(name='ant fixture back', recs=source_recs)
test_eq([m.content for m in rdlg.messages], [m.content for m in source_dlg.messages])
test_eq([m.ai_output for m in rdlg.messages], [m.ai_output for m in source_dlg.messages])Claude Code’s bookkeeping record kinds (injected-context attachment records, per-turn state, file snapshots) are dropped: they’re regenerated live and nothing reads them back. system records are the one kind kept: each becomes a raw message tagged with rec_kind (the record’s type:subtype, which is what its otherwise-blank summary row shows) in its meta and the original record verbatim under rec, so dlg2sess can re-emit it unchanged (envelope re-chained, everything else byte-for-byte) and history projections know to skip it. Each prompt also picks up its turn’s request timestamp and the reply’s usage in meta:
sysrec = dict(type='system', subtype='demo_note', uuid=stable_uuid('sys-demo'), timestamp='2026-01-01T00:00:01.000Z', level='info')
ssid = save_sess(list(reid_recs(sample, 'sysdemo')) + [sysrec], stable_uuid('sysdemo'), proj)
sysdlg = sess2dlg(ssid, proj, 'with system')
sysm = sysdlg.messages[-1]
test_eq((sysm.msg_type, sysm.meta['rec_kind'], sysm.meta['rec']['subtype']), (sraw, 'system:demo_note', 'demo_note'))
test_eq(sysm.preview(), f'{sysm.id}:r:<system:demo_note>') # a contentless raw shows its kind, not a blank row
sback = load_sess(ssid, proj)
test_eq(sysdlg.messages[0].meta['timestamp'], sback[0].timestamp) # the turn's request time
test_eq(sysdlg.messages[0].meta['usage'], obj2dict(sback[3].message.usage)) # the reply's usageBuilding the dialog back into a session re-emits the tagged record verbatim: only the envelope fields the chaining owns (sessionId, parentUuid) change, and the conversation records around it are untouched by its presence, since dlg2hist skips tagged raws:
sysid2 = dlg2sess(sysdlg, proj, key='sysback')
b2 = load_sess(sysid2, proj)
test_eq(b2[-1].type, 'system')
env = ('sessionId','parentUuid')
test_eq({k:v for k,v in obj2dict(b2[-1]).items() if k not in env}, {k:v for k,v in obj2dict(sback[-1]).items() if k not in env})
test_eq([r.type for r in b2[:-1]], [r.type for r in sback[:-1]]) # conversation records unaffectedInside a live session
Claude Code exports CLAUDE_CODE_SESSION_ID to child processes, so code running inside a session can read the transcript it is part of. The cells below use that live transcript when its file exists, and otherwise use the checked-in Agent SDK fixture.
recs = load_sess() if sess_file().exists() else source_recs
len(recs)20Conversation records are user and assistant; the rest is bookkeeping Claude Code adds as it runs. The captured fixture includes injected-context attachments, queue operations, a generated title, and a last-prompt marker:
Counter(recs.attrgot('type'))Counter({'assistant': 8,
'user': 5,
'attachment': 3,
'queue-operation': 2,
'ai-title': 1,
'last-prompt': 1})Here is one user record in full:
first(recs, lambda r: r.type=='user' and isinstance(r.get('message',{}).get('content'), str)){ 'cwd': '/Users/jhoward/aai-ws/llmsurgery/nbs/data/ant/project',
'entrypoint': 'sdk-py',
'gitBranch': 'main',
'isSidechain': False,
'message': { 'content': 'Use the ant-fixture skill. Then use Bash to run '
"`printf 'bash fixture\\n'`. Then use clikernel to "
'evaluate `6*7`. After all tools finish, reply '
'exactly: fixture complete.',
'role': 'user'},
'parentUuid': None,
'permissionMode': 'default',
'promptId': '00806146-782d-4c10-87c4-495081564d72',
'promptSource': 'sdk',
'sessionId': 'e12b4bd3-4f36-4dda-873d-2ff25d1f1044',
'timestamp': '2026-07-17T02:51:28.115Z',
'type': 'user',
'userType': 'external',
'uuid': 'f711b566-2404-41a5-a1b6-36f99d52e391',
'version': '2.1.209'}t = sess_thread(recs)
len(t), len(recs)(16, 20)The difference between the two counts is bookkeeping records without UUIDs plus any abandoned branches. On the captured fixture, four unlinked bookkeeping records are omitted while three linked attachment records remain. Consecutive records on the active chain link up:
assert all(b.parentUuid==a.uuid for a,b in zip(t, t[1:]))The proof that a written session works is still a resume. The sample below resumes the deliberately synthetic flux session because that section tests session writing; it spends tokens, so it is excluded from automated notebook runs.
opts = ClaudeAgentOptions(resume=sid, cwd=str(proj), model='haiku')
async for m in query(prompt='What is the flux reading? Reply with only the value.', options=opts):
if isinstance(m, ResultMessage): print(m.result)41.7 kilofinchesResolving session references
Callers hold whatever reference is nearest to hand - a uuid from the environment, or the title /rename gave the session. resolve_session accepts either, so tools like claudedojo -r need no separate name-handling path.
resolve_session
def resolve_session(
ref:NoneType=None, cwd:str='.'
):Resolve a session id or custom title to (sid,path)
Resolution checks an id first because that lookup is direct, then falls back to the latest custom title. The returned path is the exact transcript later compaction steps will inspect or append to.
named_path = sess_file(named_sid, proj)
test_eq(resolve_session(named_sid, proj), (named_sid,named_path))
test_eq(resolve_session('flux-renamed', proj), (named_sid,named_path))
with expect_fail(FileNotFoundError): resolve_session('missing-session', proj)Recognizing earlier compactions
Repeated synthetic compaction extends the previous bounded conversation rather than compacting it again. Explicit metadata identifies the summary and its /compact display records.
The summary itself carries the previous conversation. The following caveat, command, and stdout records only make /compact visible in Claude’s interface, so they are skipped when finding genuinely new conversation records.
tagged_summary = dict2obj(mk_rec('user', 'Earlier conversation', isCompactSummary=True, llmsurgeryCompact=True))
test_eq(_is_synthetic_compact(tagged_summary), True)The records Claude displays for /compact are wrappers rather than new conversation content.
display_rec = dict2obj(mk_rec('user', '<command-name>/compact</command-name>', llmsurgeryCompact=True))
test_eq(_is_compact_wrapper(display_rec), True)split_compaction
def split_compaction(
recs
):Return the prior compact body and records added after it
split_compaction follows the active parent chain and uses the latest synthetic summary. The small _link helper below creates that chain for tests only; it is deliberately private and not exported.
def _link(prev, r, sid):
"Link one test record after `prev`"
r['sessionId'],r['parentUuid'] = sid,prev.get('uuid') if prev else None
return rWe can now make a minimal active chain: one prior summary, one /compact display record, and one new user/assistant exchange.
split_sid = stable_uuid('split-compaction')
split_body = '§ Earlier request. §\n» Earlier answer. »'
split_summary = _link(None, mk_rec('user', compact_content(split_body, sess_file(sid, proj)), isCompactSummary=True, llmsurgeryCompact=True), split_sid)
split_wrapper = _link(split_summary, mk_rec('user', '<command-name>/compact</command-name>', llmsurgeryCompact=True), split_sid)The new exchange links directly after the display record, as it would after resuming the compacted session.
split_user = _link(split_wrapper, mk_rec('user', 'Continue the work.'), split_sid)
split_asst = _link(split_user, mk_rec('assistant', [dict(type='text', text='Continuing.')]), split_sid)
split_recs = L(dict2obj(r) for r in (split_summary,split_wrapper,split_user,split_asst))Splitting recovers the old DSL body unchanged.
prior,new = split_compaction(split_recs)
test_eq(prior, split_body)The wrapper is discarded, leaving only the newly added conversation records.
test_eq([rec_role(r) for r in new], ['user','assistant'])
test_eq([rec_txt(r) for r in new], ['Continue the work.','Continuing.'])Without an earlier synthetic compaction, the prior body is empty and the active thread is returned intact.
plain_prior,plain_new = split_compaction(back)
test_eq(plain_prior, '')
test_eq(plain_new.attrgot('uuid'), sess_thread(back).attrgot('uuid'))Synthetic compaction
The compact DSL describes conversation content; this section wraps that content in Claude Code records. A compaction starts a new active chain with a system boundary, followed by one synthetic summary and the three records Claude displays for /compact.
sess_meta
def sess_meta(
recs
):Latest available session metadata for synthetic records
Transcript metadata is not present uniformly on every record. sess_meta searches backwards for each field so synthetic records inherit the latest value available in the session.
fork_meta = sess_meta(fork_recs)
fork_meta{'userType': 'external',
'entrypoint': 'sdk-py',
'cwd': '/Users/jhoward/aai-ws/llmsurgery/nbs/data/ant/project',
'sessionId': '5d2b9564-f7f8-48c3-b611-f33fbe39c577',
'version': '2.1.209',
'gitBranch': 'main'}The running example uses the native fork captured above. We re-identify its records into a disposable session, split after the Bash result, and save the fixture bytes so preparation can be shown to be read-only.
The split index is specific to the captured transcript. When refreshing the fixture, inspect fork_thread and choose a boundary immediately after a completed tool result, before the next assistant action. Do not separate a tool_use from its matching tool_result. Both sides should contain enough real activity to make the first compaction and the later incremental compaction informative.
compact_name = 'ant-fork-compact'
compact_first = reid_recs(fork_thread[:10], 'compact-real-first')
compact_rest = reid_recs(fork_thread[10:], 'compact-real-rest')
compact_sid = save_sess(compact_first, stable_uuid('compact-real'), proj)
name_sess(compact_sid, compact_name, proj)
compact_path = sess_file(compact_sid, proj)
compact_before,fork_before = compact_path.read_bytes(),fork_path.read_bytes()
len(compact_first),len(compact_rest),compact_sid,compact_path(10,
6,
'91a8050c-7243-58d4-8802-0dc33e88638e',
Path('/Users/jhoward/.claude/projects/-private-var-folders-51-b2-szf2945n072c0vj2cyty40000gn-T-tmp7r6kftc5/91a8050c-7243-58d4-8802-0dc33e88638e.jsonl'))compact_records
def compact_records(
recs, content, pre_toks, post_toks
):Create a tagged compact boundary, summary, and visible /compact records
All five records carry an explicit llmsurgery tag. The boundary records token accounting, the summary contains the continuation document, and the remaining records reproduce Claude’s visible /compact interaction.
prepare_compaction
def prepare_compaction(
ref, cwd:str='.', policy:dict={'user_toks': 2000, 'asst_toks': 150, 'call_toks': 60, 'result_toks': 35},
enc:NoneType=None
):Prepare an incremental synthetic compaction without writing it
prepare_compaction is read-only. It preserves an earlier llmsurgery segment, compacts only the active suffix, leaves the final five canonical messages untruncated, and returns both the generated document and records for inspection.
Preparation constructs the continuation document and five synthetic records without changing the transcript.
compaction = prepare_compaction(compact_sid, proj)
test_eq(compaction.path.read_bytes(), compact_before)
compaction.pre_toks,compaction.post_toks,len(compaction.records)(90, 293, 5)The prepared records can be inspected before deciding to append them.
show_recs(compaction.records, mx=300, showall=True)--- system:compact_boundary 2026-07-17T05:27:49 ---
--- user 2026-07-17T05:27:49 ---
This session is being continued from an earlier conversation. `§ … §` encloses user text; `» … »` encloses assistant text; fenced `python` is persistent-kernel execution; `!` marks Bash; `▶` marks another tool call; `>` marks its result; `¶` replaces a newline; `***` separates user-led turns; and ` …[+783 chars]
--- user 2026-07-17T05:27:49 ---
<local-command-caveat>Caveat: The messages below were generated by the user while running local commands. DO NOT respond to these messages or otherwise consider them in your response unless the user explicitly asks you to.</local-command-caveat>
--- user 2026-07-17T05:27:49 ---
<command-name>/compact</command-name>
<command-message>compact</command-message>
<command-args></command-args>
--- user 2026-07-17T05:27:49 ---
<local-command-stdout>Compacted (ctrl+o to see full summary)</local-command-stdout>append_compaction
def append_compaction(
compaction
):Append a prepared compaction and restore its session name
Appending establishes the compact boundary as the new active chain and restores the session name.
append_compaction(compaction)
compact_recs = load_sess(compact_sid, proj)
compact_thread = sess_thread(compact_recs)
test_eq(sess_by_name(compact_name, proj), compact_path)
len(compact_recs),len(compact_thread)(19, 5)The five active records are the boundary, summary, caveat, command, and stdout. Their tags make recognition independent of their prose, while the summary body remains available for repeated incremental compaction.
test_eq(len(compact_thread), 5)
test_eq((compact_thread[0].type,compact_thread[0].subtype), ('system','compact_boundary'))
test_eq(compact_thread[1].isCompactSummary, True)
test_eq([r.get('llmsurgeryCompact') for r in compact_thread], [True]*5)
compact_prior,compact_new = split_compaction(compact_recs)
test_eq(compact_new, [])
test_eq(compact_prior, compact_body(rec_txt(compact_thread[1])))
show_recs(compact_thread, mx=300, showall=True)--- system:compact_boundary 2026-07-17T05:27:49 ---
--- user 2026-07-17T05:27:49 ---
This session is being continued from an earlier conversation. `§ … §` encloses user text; `» … »` encloses assistant text; fenced `python` is persistent-kernel execution; `!` marks Bash; `▶` marks another tool call; `>` marks its result; `¶` replaces a newline; `***` separates user-led turns; and ` …[+783 chars]
--- user 2026-07-17T05:27:49 ---
<local-command-caveat>Caveat: The messages below were generated by the user while running local commands. DO NOT respond to these messages or otherwise consider them in your response unless the user explicitly asks you to.</local-command-caveat>
--- user 2026-07-17T05:27:49 ---
<command-name>/compact</command-name>
<command-message>compact</command-message>
<command-args></command-args>
--- user 2026-07-17T05:27:49 ---
<local-command-stdout>Compacted (ctrl+o to see full summary)</local-command-stdout>Repeated compaction preserves the prior compact body and processes only records added after its /compact wrapper. We append the remaining records from the same native fork, which contain the clikernel call and final response.
append_sess(compact_rest, compact_sid, proj)
test_eq(fork_path.read_bytes(), fork_before)
len(load_sess(compact_sid, proj))25Preparing again recovers the previous body exactly and creates a new segment from the remaining real conversation records.
compaction2 = prepare_compaction(compact_sid, proj)
test_eq(compaction2.prior, compaction.chat)
test('6*7', compaction2.new_chat, in_)
test('fixture complete', compaction2.new_chat, in_)
test_eq(compaction2.chat, join_compacts(compaction.chat, compaction2.new_chat))
compaction2.pre_toks,compaction2.post_toks(109, 311)Appending the second preparation again leaves five active synthetic records, now carrying both immutable segments.
append_compaction(compaction2)
compact_recs2 = load_sess(compact_sid, proj)
compact_thread2 = sess_thread(compact_recs2)
test_eq(len(compact_thread2), 5)
test_eq(compact_body(rec_txt(compact_thread2[1])), compaction2.chat)
show_recs(compact_thread2, mx=300, showall=True)--- system:compact_boundary 2026-07-17T05:27:49 ---
--- user 2026-07-17T05:27:49 ---
This session is being continued from an earlier conversation. `§ … §` encloses user text; `» … »` encloses assistant text; fenced `python` is persistent-kernel execution; `!` marks Bash; `▶` marks another tool call; `>` marks its result; `¶` replaces a newline; `***` separates user-led turns; and ` …[+833 chars]
--- user 2026-07-17T05:27:49 ---
<local-command-caveat>Caveat: The messages below were generated by the user while running local commands. DO NOT respond to these messages or otherwise consider them in your response unless the user explicitly asks you to.</local-command-caveat>
--- user 2026-07-17T05:27:49 ---
<command-name>/compact</command-name>
<command-message>compact</command-message>
<command-args></command-args>
--- user 2026-07-17T05:27:49 ---
<local-command-stdout>Compacted (ctrl+o to see full summary)</local-command-stdout>compact_session combines the preparation and append steps into the high-level operation callers normally need.
compact_session
def compact_session(
ref, cwd:str='.', policy:dict={'user_toks': 2000, 'asst_toks': 150, 'call_toks': 60, 'result_toks': 35},
enc:NoneType=None
):Generate and append a synthetic session compaction
The high-level helper is exercised on a fresh disposable copy of the complete native fork. It should prepare and append in one call, preserve the name, leave the captured fixture untouched, and establish the same five-record active chain.
compact_once_name = 'ant-fork-compact-once'
compact_once_sid = save_sess(reid_recs(fork_thread, 'compact-session-real'), stable_uuid('compact-session-real'), proj)
name_sess(compact_once_sid, compact_once_name, proj)
compact_once = compact_session(compact_once_sid, proj)
compact_once_thread = sess_thread(load_sess(compact_once_sid, proj))
test_eq(sess_by_name(compact_once_name, proj), compact_once.path)
test_eq(len(compact_once_thread), 5)
test_eq(compact_body(rec_txt(compact_once_thread[1])), compact_once.chat)
test_eq(fork_path.read_bytes(), fork_before)
compact_once.pre_toks,compact_once.post_toks,len(compact_once_thread)(108, 306, 5)The active chain is the complete result callers receive from compact_session.
show_recs(compact_once_thread, mx=300, showall=True)--- system:compact_boundary 2026-07-17T05:27:49 ---
--- user 2026-07-17T05:27:49 ---
This session is being continued from an earlier conversation. `§ … §` encloses user text; `» … »` encloses assistant text; fenced `python` is persistent-kernel execution; `!` marks Bash; `▶` marks another tool call; `>` marks its result; `¶` replaces a newline; `***` separates user-led turns; and ` …[+827 chars]
--- user 2026-07-17T05:27:49 ---
<local-command-caveat>Caveat: The messages below were generated by the user while running local commands. DO NOT respond to these messages or otherwise consider them in your response unless the user explicitly asks you to.</local-command-caveat>
--- user 2026-07-17T05:27:49 ---
<command-name>/compact</command-name>
<command-message>compact</command-message>
<command-args></command-args>
--- user 2026-07-17T05:27:49 ---
<local-command-stdout>Compacted (ctrl+o to see full summary)</local-command-stdout>print(compact_once.chat)§ Use the ant-fixture skill. Then use Bash to run `printf 'bash fixture\n'`. Then use clikernel to evaluate `6*7`. After all tools finish, reply exactly: fixture complete. §
» I'll execute these steps in sequence. »
▶ Skill(skill='ant-fixture')
> Launching skill: ant-fixture
***
§ …compacted… §
! printf 'bash fixture\n'
> bash fixture
```python
6*7
```
> 42
» fixture complete »Does this all look fairly sound to you? Anything jump out as screwy?
🤖Reply🤖
Overall, it looks well thought through. The narrative, real fixtures, deterministic rewriting, parent-chain tests, and dialog/session round trips are unusually solid. A few things jump out:
CC_VERSIONis already stale relative to the fixture. It is set to2.1.206, while the captured records show2.1.209. Sinceversionis optional, either derive it dynamically, inherit it from source metadata, or omit it. A hard-coded stale version weakens the “realistic envelope” claim.strip_thinkonly removes thinking-only records. A record containing boththinkingandtool_use/textretains its thinking block. That makesfork_sess(..., think=False)and the prose about removing thinking slightly misleading, and could expose or retain substantial thinking content. Either strip thinking blocks from mixed records too, or rename/document the narrower behavior.Prepared compactions can go stale.
prepare_compactionis deliberately separated fromappend_compaction, but the latter does not verify that the transcript is unchanged. If Claude appends a turn between those calls, the new compact boundary makes that turn unreachable. Recording and checking a file hash or byte length before append would make the inspect-then-commit workflow safe.sess_threadassumes the last conversation-like record is the main chain’s tip. A laterisSidechain=Truerecord could make it follow a sidechain instead. If Claude stores sidechains in the same JSONL, this is a real correctness issue. At minimum, explicitly filter or test sidechain behavior.Compaction token accounting looks dimensionally inconsistent.
pre_tokscounts the reconstructed compact DSL, whereaspost_tokscountscompact_content(...), including its wrapper. The examples consequently report90 → 293and108 → 306. Those values may be harmless metadata, but they do not appear to represent actual context before and after compaction.The central compaction path lacks the strongest test. You prove ordinary synthetic sessions resume, but do not resume a synthetically compacted session and verify that Claude recalls facts from both the old summary and the newly compacted suffix. That is the one live, excluded-from-CI test I would most want here.
Two smaller edges:
fork_sessusesif mx:, somx=0silently disables truncation;if mx is not None:better matches the API.append_sesslinks to the last UUID in file order, not necessarily the active chain tip. That is fine for the fixtures but deserves either an explicit invariant or a chain-aware implementation.
Nothing suggests the basic design is wrong. The stale-write risk and mixed-thinking behavior are the two issues I would address first; the end-to-end compact-resume test would then validate the format assumptions.
Locked agents
A recurring automation pattern: an agent must run a fixed sequence of prepared commands, read each result, and judge whether to continue, with no freedom to improvise other commands. LLMs are good at the judgment and bad at the restraint. Told “run exactly these commands”, a capable model that mistrusts its own reading of a large output will reinvent verification as grep pipelines and scratch files, so the restriction must be mechanical, not textual.
Claude Code subagents are the obvious host, but a child of a live session inherits that session’s environment: permission prompts for every command outside the user’s allow rules, the user’s hooks, and MCP server instructions injected beside tool results. The Agent SDK gives the same model and tools without any of that. Each item below cost one failed run to learn:
setting_sources=[]keeps the child bare. Without it the child loads user settings, and a blocking PreToolUse hook beatsallowed_tools(blocking hooks take precedence over allow rules). Headless, nothing surfaces the refusal: the run ends withsubtype='success'and an emptyresult, having done nothing.allowed_toolsaccepts permission rules likeBash(/path/step1.sh *)(make the script executable and name it directly; abash /path/...spelling only widens the prefix). Put the fixed commands in small scripts, allow exactly those prefixes, and the model’s discretion shrinks to judgment over their output, the part it is good at. One caveat, verified empirically: on this path the rules are pure prefix matchers, with none of the compound-command splitting documented for interactive sessions, soBash(echo hi *)also permitsecho hi there | wc -l. The lock stops off-prefix commands; auditing the kept transcript covers pipe suffixes.- Nested
claude -pspawned from a session’s Bash tool fails subscription auth (“Not logged in”). The SDK child authenticates with the same login fine. - Keep the whole message stream, not just
ResultMessage. A run killed by a blocked tool call looks successful from itsResultMessagealone; only the transcript shows the tool call that never received a result. - Give the protocol a version string the report must echo, so a run on a stale prompt is detectable from its report rather than by inference.
- End each script’s output with a sentinel line and require the agent to see it, since a model otherwise cannot distinguish truncated output from complete output.
asyncio.wait_for bounds each run, and cancelling it kills the child. asyncio.gather over several calls runs a batch in parallel.
run_locked_agent
async def run_locked_agent(
sysp:str, # Protocol for the child agent, as its system prompt
prompt:str, # The task instance, e.g. a repo path
allowed:list, # Permission rules, e.g. [r'Bash(/path/step1.sh *)']
model:str='sonnet', # Model alias or full id
max_turns:int=15, # Turn budget for the child
timeout:int=600, # Seconds before the child is cancelled and killed
):Run a bare headless agent locked to allowed commands, returning (report, message stream)
A minimal live run locks the child to one read-only git command. It spends tokens, so it stays out of automated runs:
locked_sysp = 'Run exactly one command, once: `git log --oneline -3`, via the Bash tool. Report only the newest commit subject line.'
rpt,ms = await run_locked_agent(locked_sysp, 'Go.', [r'Bash(git log *)'], model='haiku', max_turns=3, timeout=120)
print(rpt)Claude as a chat API
A saved session plus resume turns Claude Code into something close to a stateless chat API: the caller keeps the conversation, files it with msgs2sess on each request, and sends the newest user message as the prompt of the resumed query. What remains is tools and streaming: the caller’s tools must be requested by Claude but executed by the caller, and the reply must read back as ordinary Anthropic events. Claude Code’s hook system covers the tool half natively: a PreToolUse hook returning permissionDecision: "defer" stops the run at the tool call and records the deferral in the transcript, and resuming a session with a deferred call makes Claude Code re-invoke it, so its result can be supplied through the normal MCP channel. The claude_code backend in fastllm-claude-code is this pattern end to end.
defer_tools builds the three ClaudeAgentOptions pieces that hand a tool list over to this protocol: an MCP server hosting the tools, the matching permission rules, and the routing hook. The hook applies one rule: a call whose name and input have a held result is allowed—and the MCP handler answers it from the store—while everything else defers, ending the run so the caller can execute the requested calls. A fresh request holds no results, so every call defers; a continuation holds exactly the results being returned. Results are held under the qualified tool name (mcp__<server>__<name>), which is how both the hook and the transcript record the call.
defer_tools
def defer_tools(
name, # MCP server name to register the tools under
tools, # The caller's tools, as `(name, description, schema)` triples
held:NoneType=None, # Results queued by `hold_result`; calls without one defer
):mcp_servers, allowed_tools, and hooks options deferring tools to the caller
hold_result
def hold_result(
held, name, input, content, is_error:bool=False
):Queue a result for defer_tools to return when Claude re-invokes name with input; returns held
tool_reply
def tool_reply(
content, is_error:bool=False
):An MCP tool return value carrying an Anthropic tool_result’s content
held = hold_result({}, 'mcp__probe__flux_meter', {}, 'flux: 41.7 gauss')
test_eq(first(held.values()), [dict(content=[dict(type='text', text='flux: 41.7 gauss')], isError=False)])
ib = dict(type='image', source=dict(type='base64', media_type='image/png', data='AAAA'))
test_eq(tool_reply([ib])['content'], [dict(type='image', data='AAAA', mimeType='image/png')])
servers,allowed,hooks = defer_tools('probe', [('flux_meter', 'Read the flux', {})], held)
test_eq(allowed, ['mcp__probe__flux_meter'])
test_eq(defer_tools('probe', []), ({},[],{}))
serversOn the wire the hosted tools are mcp__<server>__<name>, and a resumed history containing earlier calls must use those same qualified names or Claude won’t connect them to the tools now on offer. prefix_tools qualifies past tool_use blocks, leaving alone names that are already qualified and Claude Code’s own tools (e.g. WebSearch), which run under their bare names.
prefix_tools
def prefix_tools(
msgs, # Anthropic-style messages
prefix, # Stub name prefix, e.g. 'mcp__probe__'
skip:tuple=(), # Native tool names to leave unqualified, e.g. 'WebSearch'
):Copy of msgs with client tool_use names qualified by prefix
tmsgs = [dict(role='assistant', content=[dict(type='text', text='Checking.'), dict(type='tool_use', id='t1', name='flux_meter', input={})]),
dict(role='assistant', content=[dict(type='tool_use', id='t2', name='WebSearch', input=dict(query='flux'))])]
pref = prefix_tools(tmsgs, 'mcp__probe__', skip=['WebSearch'])
test_eq([b['name'] for m in pref for b in m['content'] if b['type']=='tool_use'], ['mcp__probe__flux_meter','WebSearch'])
test_eq(tmsgs[0]['content'][1]['name'], 'flux_meter')A deferral is recorded in the transcript as a hook_deferred_tool attachment naming the call, and that record is all resume needs—so the state can be forged: file a conversation ending at an assistant tool_use, append mk_deferred for it, and resume with no_prompt(), adding no user turn at all. Claude Code re-invokes the deferred call, the held result answers it, and the CLI inserts its own continuation sentinel (“Continue from where you left off.”) before the model proceeds. Claude Code re-invokes only the last deferred call, so a caller returning several results files all but the last as ordinary tool_result records and defers exactly one.
no_prompt
def no_prompt():An empty streaming-input prompt: resume a session without adding a user turn
mk_deferred
def mk_deferred(
tu, # The pending `tool_use` block dict
cwd:str='.', # Project directory recorded in the envelope
uid:NoneType=None, # Record uuid; random if None
ts:str='2026-01-01T00:00:00.000Z', # Timestamp recorded in the envelope
):A hook_deferred_tool attachment record marking tu as deferred, ready for save_sess
dtu = dict(type='tool_use', id='toolu_01', name='mcp__probe__flux_meter', input={})
datt = mk_deferred(dtu, uid=stable_uuid('datt'))
test_eq(datt['attachment'], dict(type='hook_deferred_tool', toolUseID='toolu_01', toolName='mcp__probe__flux_meter',
toolInput={}, hookName='settings', hookEvent='PreToolUse', permissionMode='default'))
test_eq(mk_deferred(dtu, uid=stable_uuid('datt')), datt)The reply arrives as SDK message objects, and with include_partial_messages=True the raw Anthropic stream events ride along in StreamEvents. Two mismatches remain between that stream and a plain API response. First, an error ends the stream as a ResultMessage rather than raising; QueryError converts it, preferring the CLI’s explicit api_error_status and falling back to sniffing a status code out of the message text.
QueryError
def QueryError(
result
):An error ResultMessage from a query stream
def _err_rm(**kw): return ResultMessage(subtype='success', duration_ms=0, duration_api_ms=0, is_error=True, num_turns=1, session_id='s', **kw)
err = QueryError(_err_rm(result='API Error: 429 rate limited'))
test_eq((err.status, str(err)), (429, 'API Error: 429 rate limited'))
test_eq(QueryError(_err_rm(result='overloaded', api_error_status=529)).status, 529)Second, one logical response can span several messages (thinking rounds, server-side tool use), and each message restarts its content-block index at 0—while a consumer of one response expects a single global sequence. _reidx rebases each message’s block indices past the highest index the previous messages reached.
evs = [dict(type='message_start'), dict(type='content_block_start', index=0), dict(type='content_block_stop', index=0), dict(type='message_stop'),
dict(type='message_start'), dict(type='content_block_start', index=0), dict(type='content_block_start', index=1), dict(type='message_stop')]
f = _reidx()
test_eq([f(e).get('index') for e in evs], [None,0,0,None,None,1,2,None])aquery_events runs one query and yields only the raw events, re-indexed. Deferral needs no special handling here: once every requested call has deferred, the run ends by itself with a complete stream, message_stop included, and a stop_reason of tool_deferred on the result. The one adjustment is cosmetic: with a prefix, hosted tool_use events are presented under their bare names, since the caller knows its tool as flux_meter, not mcp__probe__flux_meter.
aquery_events
def aquery_events(
prompt, # The new user turn: a string, or a streaming-input iterable, e.g. `no_prompt()`
opts, # `ClaudeAgentOptions`, with `include_partial_messages=True`
prefix:NoneType=None, # MCP name prefix to strip from `tool_use` events, matching `defer_tools` naming
):Anthropic stream events from a query, re-indexed as one global sequence; raises QueryError on an error result
A live request with everything wired: Claude is told it has a flux_meter, asks to use it, the call defers, and the run stops with the bare tool name restored and a complete stream (it spends tokens, so it stays out of automated runs). defer_tools is called without held results here—a fresh request defers everything.
qserv,qallow,qhooks = defer_tools('probe', [('flux_meter', 'Read the flux', {})])
qopts = ClaudeAgentOptions(model='haiku', cwd=str(proj), include_partial_messages=True, mcp_servers=qserv, allowed_tools=qallow,
hooks=qhooks, strict_mcp_config=True, tools=[], setting_sources=[], system_prompt='You have a flux_meter tool. Use it when asked.')
levs = [ev async for ev in aquery_events('Measure the flux please.', qopts, prefix='mcp__probe__')]
ltu = first(ev['content_block'] for ev in levs if ev.get('content_block',{}).get('type')=='tool_use')
test_eq(ltu['name'], 'flux_meter')
test_eq(levs[-1]['type'], 'message_stop')
ltuAnd the continuation, forged from nothing: the conversation is rebuilt with mk_deferred marking the call, the result is held, and the resume runs with no_prompt(). Claude Code re-invokes the call, pairs the held result into the transcript, and inserts its sentinel—no synthetic user text anywhere. What the model says next is its own business (free of any larger task it often just measures again), so the checks pin the protocol instead: the held result is consumed, and lands in the transcript as an ordinary tool record.
cheld = hold_result({}, 'mcp__probe__flux_meter', {}, 'flux: 41.7 gauss')
cserv,callow,chooks = defer_tools('probe', [('flux_meter', 'Read the flux', {})], cheld)
ctu = dict(type='tool_use', id='toolu_c01', name='mcp__probe__flux_meter', input={})
crecs = [mk_rec('user', 'Measure the flux please.'), mk_rec('assistant', [ctu])]
ccsid = save_sess(crecs+[mk_deferred(ctu, cwd=proj)], cwd=proj)
ccopts = ClaudeAgentOptions(model='haiku', cwd=str(proj), resume=ccsid, include_partial_messages=True, mcp_servers=cserv,
allowed_tools=callow, hooks=chooks, strict_mcp_config=True, tools=[], setting_sources=[])
cevs = [ev async for ev in aquery_events(no_prompt(), ccopts, prefix='mcp__probe__')]
back = sess_thread(load_sess(ccsid, proj))
test_eq(cheld, {k: [] for k in cheld})
assert any(rec_txt(r)=='flux: 41.7 gauss' for r in conv_recs(back))
print(show_recs(back[:4], mx=100))Cleanup
Remove the sample from ~/.claude/projects, along with the scratch project.
shutil.rmtree(sess_dir(proj))
shutil.rmtree(proj)