ant

Read, write, and build Claude Code session transcripts

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.

from fastcore.test import *
from fastllm.chat import tool_dtls_tag, fmt2hist
from llmsurgery.dialog import *
from collections import Counter
import tempfile, shutil

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.

---

source

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).

---

source

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'))

---

source

sess_file

def sess_file(
    sid:NoneType=None, # Session id; `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')

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.

---

source

stable_uuid

def stable_uuid(
    s
):

A uuid deterministically derived from string s

---

source

canon

def canon(
    o
):

Canonical compact JSON for o, key-sorted, for stable hashing

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')

---

source

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': '1d377c05-0278-4605-afa7-29c0d9beb546',
 'parentUuid': None,
 'sessionId': None,
 'timestamp': '2026-07-14T04:12:07.703Z',
 '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()))

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.

---

source

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

---

source

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.

---

source

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

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))

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.

---

source

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

---

source

mk_tr

def mk_tr(
    tu, # The `tool_use` block being answered
    content, # The tool's output
):

The tool_result content block answering tu

---

source

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

'4d5671f8-b730-43a2-b35e-6977f164f0e3'

Reading a session

A transcript is one JSON object per line. load_sess wraps each in dict2obj so fields read as attributes.

---

source

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.

The parent chain

Resume does not replay the file top to bottom. Reconstruction starts at the last record and walks parentUuid links backwards, so a record nothing links to is dropped (the CLI prints a warning). Rewinding a conversation is what creates such records: the abandoned turns stay in the file, off the final chain. sess_thread performs the same walk.

---

source

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

On the sample, every record is on the chain:

test_eq(sess_thread(back).attrgot('uuid'), back.attrgot('uuid'))

Break a link and the walk stops early, mirroring what resume does with unchained records:

broken = load_sess(sid, proj)
broken[2].parentUuid = None
test_eq(len(sess_thread(broken)), 2)

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.

---

source

rec_txt

def rec_txt(
    r, # A session record
):

Every readable string in r’s message content, joined, for finding records by text

Content blocks nest (a tool_result can itself hold a list of text blocks), so rec_txt walks the whole message and joins every readable string it finds, skipping bookkeeping fields — block types, tool ids, and thinking signatures — that would otherwise pollute matches. On the sample, the planted fact appears in the tool result and in the reply that quotes it:

test_eq([r['type'] for r in sample if 'kilofinches' in rec_txt(r)], ['user','assistant'])

---

source

rec_role

def rec_role(
    r, # A session record
):

The conversational role of r: a user record carrying tool results counts as tool

---

source

conv_recs

def conv_recs(
    recs, # Session records, e.g. from `load_sess`
):

Just the records carrying conversation messages, dropping Claude Code’s bookkeeping

conv_recs is the searchable subset of a transcript, and rec_role labels each record the way recs2chat classifies them: the user type covers both real prompts and the tool results riding back, and telling them apart is the first thing every reader of a transcript needs. On the sample session:

test_eq(len(conv_recs(sample)), 4)
test_eq([rec_role(r) for r in sample], ['user','assistant','tool','assistant'])

---

source

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
):

Search every conversation record of a session, returning SessHits with the records on .recs

---

source

SessHits

def SessHits(
    *args, **kwargs
):

Search hits with a match-centered preview per line

The result displays one line per hit — index, role, timestamp, and the text around the first match — and keeps the searched records on .recs, so a hit’s index leads straight to its neighbors. Searching the whole file rather than the thread is deliberate: pre-compaction history is exactly what archaeology is usually after. On the sample session the planted fact appears in the tool result and the reply that reports it:

hits = sess_search('kilofinches', sid, proj)
test_eq([h.role for h in hits], ['tool','assistant'])
hits

    2 tool      2026-07-14T04:12 flux: 41.7 kilofinches
    3 assistant 2026-07-14T04:12 The flux reading is 41.7 kilofinches.

---

source

show_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 an interesting hit, and read 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:], mx=60)
assert s.count('---\n')==4 and 'flux: 41.7' in s
s
sysr = dict(type='system', subtype='compact_boundary', timestamp='2026-01-01T00:00:00.000Z')
sa = show_recs([sample[0], sysr], showall=True)
assert sa.count('---\n')==2 and 'system:compact_boundary' in sa

--- user 2026-07-14T04:12:07 ---
Measure the flux please.
--- assistant 2026-07-14T04:12:07 ---
flux_meter
--- tool 2026-07-14T04:12:07 ---
flux: 41.7 kilofinches
--- assistant 2026-07-14T04:12:07 ---
The flux reading is 41.7 kilofinches.

Curating a captured session

Synthetic construction suits a short demo. For anything longer, capture beats writing: run a real session, then keep the span worth replaying. The search tools above find the span; curation then needs two operations: dropping what resume does not need, and re-deriving ids so one capture gives one file.

Thinking blocks are usually the bulk of a captured assistant turn, and resume rebuilds a conversation without them. Live transcripts write one content block per assistant record, so stripping thinking means dropping whole records; save_sess re-chains whatever survives.

---

source

strip_think

def strip_think(
    recs, # Session records
):

Drop records whose message content is only thinking blocks; resume does not need them

think = mk_rec('assistant', [dict(type='thinking', thinking='Quiet planning.', signature='')])
test_eq(strip_think([*sample, think]), sample)

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.

---

source

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:

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 in every line. reid_recs re-derives them from position and a salt instead: record uuids, tool_use/tool_result pairs, assistant API metadata, and any envelope field that referenced a renamed uuid (such as sourceToolAssistantUUID). Records come back as fresh dicts; the originals are untouched.

---

source

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-identified records keep the tool pairing, differ under a different salt, and leave the originals alone:

r1,r2 = reid_recs(sample, 'tmpl'),reid_recs(sample, 'tmpl')
test_eq(canon(list(r1)), canon(list(r2)))
test_eq(r1[2]['message']['content'][0]['tool_use_id'], r1[1]['message']['content'][0]['id'])
test_ne(r1[0]['uuid'], reid_recs(sample, 'other')[0]['uuid'])
test_ne(r1[1]['message']['content'][0]['id'], sample[1]['message']['content'][0]['id'])

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(json.loads(sess_file(tid, proj).read_text().splitlines()[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])

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.

---

source

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
):

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:

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)

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.

---

source

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

---

source

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 details block in the reply format `fmt2hist` parses"
    d = json.dumps(dict(id='call1', server=False, call=dict(function=func, arguments=args), result=result))
    return f"{tool_dtls_tag}\n<summary><code>{func}(...)</code></summary>\n\n```json\n{d}\n```\n\n</details>"

png = tiny_png
fdlg = Dialog('flux')
fatt = Attachment(png, 'image/png')
fdlg.mk_message(f'The rig: ![](attachment:{fatt.id})', 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'])]

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, ResultMessage
opts = 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: one Msg per record, with tool calls and results carrying the same part data that fmt2hist produces, and tool_result records taking the tool role. Thinking blocks are kept as thinking parts (run strip_think first to drop them), and any block type we cannot faithfully convert raises rather than silently degrading.

---

source

recs2chat

def recs2chat(
    recs, # Session records, e.g. from `load_sess`
):

Canonical fastllm messages for the conversation records in recs

On the sample session, the four records become user, assistant, tool, and assistant messages, with the tool call and its result joined by one id:

smsgs = recs2chat(sample)
test_eq([m.role for m in smsgs], ['user','assistant','tool','assistant'])
test_eq(smsgs[0].text, 'Measure the flux please.')
test_eq(smsgs[1].content[0].data['id'], smsgs[2].content[0].data['id'])
test_eq(_tr_txt(dict(content=[dict(type='tool_reference', tool_name='probe')])), '<tool_reference tool="probe"/>')
test_eq(recs2chat([mk_rec('assistant', 'hi')])[0].text, 'hi')

---

source

chat2dlg

def chat2dlg(
    msgs, # Canonical messages, e.g. from `recs2chat`
    name, # Dialog name
    cls:type=Dialog, # Dialog class to create
    mx:int=2000, # Maximum characters per rendered tool input/output string; None disables truncation (see `hist2fmt`)
):

A dialog for msgs: one prompt per user turn, replies rendered in the format fmt2hist parses

Prompt content is the user turn verbatim: text parts joined by blank lines, and images pulled out as attachments with a reference left in the text. A session that was itself built from a dialog therefore converts back with its rendered XML wrapping (<prompt>, <message type="note">) still in place, not the original notes; the reverse direction is for sessions recorded live, where user content is plain.

mdlg = chat2dlg(smsgs, 'flux back')
test_eq(len(mdlg.messages), 1)
test_eq(mdlg.messages[0].content, 'Measure the flux please.')

big = 'z'*9999
bdlg = chat2dlg(recs2chat(tool_turn('Big!', 'probe', {}, big, 'Done.')), 'untruncated', mx=None)
assert big in bdlg.messages[0].ai_res

The reply renders as text plus a tool-call details block, and fmt2hist parses it back to exactly the canonical messages it came from. This is the law that makes the conversion safe to edit: the dialog holds nothing the session format cannot round-trip.

test_eq(fmt2hist(mdlg.messages[0].ai_output), smsgs[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())

---

source

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`)
):

The conversation of session sid as a dialog, one prompt per user turn; system records ride along as tagged raws

sess2dlg composes the whole read path: load the transcript, walk the parent chain, drop thinking records, and convert. Reading the sample session back gives the same dialog we built from its messages by hand:

rdlg = sess2dlg(sid, proj, 'flux back')
test_eq(rdlg.messages[0].content, mdlg.messages[0].content)
test_eq(rdlg.messages[0].ai_output, mdlg.messages[0].ai_output)

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 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'))
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 usage

Building 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 unaffected

Inside a live session

Since Claude Code exports CLAUDE_CODE_SESSION_ID to child processes, code running inside a session can read the very transcript it is part of. The cells below do that when a live transcript exists, and fall back to the sample session otherwise (a plain shell, CI, or a resumed session whose advertised id names no file).

recs = load_sess() if sess_file().exists() else load_sess(sid, proj)
len(recs)

4

The conversation itself is the user and assistant records. The rest is bookkeeping Claude Code adds as it runs: attachment for injected context such as skills and file contents, system for hook output, and assorted prompt, mode, and snapshot markers. Compaction adds a user record flagged isCompactSummary, carrying a summary of everything before it.

Counter(recs.attrgot('type'))

Counter({'user': 2, 'assistant': 2})

Here is one user record in full:

first(recs, lambda r: r.type=='user' and isinstance(r.get('message',{}).get('content'), str))

{ 'cwd': '/private/var/folders/51/b2_szf2945n072c0vj2cyty40000gn/T/tmpw6hbvcbu',
  'gitBranch': 'HEAD',
  'isSidechain': False,
  'message': { 'content': 'Measure the flux please.',
               'role': 'user',
               'type': 'message'},
  'parentUuid': None,
  'permissionMode': 'default',
  'sessionId': '4d5671f8-b730-43a2-b35e-6977f164f0e3',
  'timestamp': '2026-07-14T04:12:07.848Z',
  'type': 'user',
  'userType': 'external',
  'uuid': 'd1b64443-322f-4a42-873f-4f2084145e5a',
  'version': '2.1.206'}

t = sess_thread(recs)
len(t), len(recs)

(4, 4)

In a live transcript the gap between the two counts is bookkeeping records that carry no uuid, plus any abandoned branches; the sample has neither. Consecutive records on the chain link up:

assert all(b.parentUuid==a.uuid for a,b in zip(t, t[1:]))

The proof that the sample is a working session: resume it and ask about the planted fact. The Claude Agent SDK drives the same CLI, so resume there reads the same files. This spends tokens, so it is excluded from automated tests.

from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage
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 kilofinches

Cleanup

Remove the sample from ~/.claude/projects, along with the scratch project.

shutil.rmtree(sess_dir(proj))
shutil.rmtree(proj)