ant

Read, search, and edit Claude Code sessions. Compact conversations and recover prompt history.

Use sess2dlg to read a Claude Code session as a dialog. It combines each user turn and its replies into one prompt message. d.summary() lists messages with their sizes. Search with d.find_msgs(pat) and read matches with view_msg or view_msgs. See doc(aidialog.dlgskill) for these tools.

sess2dlg follows the active chain. After compaction this includes the summary and subsequent conversation. To find earlier discussion still present in the transcript, search all records with sess_search. It excludes base64 data, signatures, and bookkeeping from the searchable text.

Save a dialog with write_ipynb to search it alongside other notebooks with rgapi.nbrg. The saved prompt sources include their replies.

Working with records

Use the record APIs to edit a session or inspect its metadata:

• Find the transcript with sess_file, cur_sess, or resolve_session. The latter accepts a session ID or its /rename title.
• Load it with load_sess, or use load_recs for an arbitrary path.
• Select the active chain with sess_thread. Use conv_recs to keep conversation records and rec_role to distinguish tool results from user messages.
• Search with sess_search. Each hit has its record on .rec. Read surrounding records with show_recs.

Claude Code stores transcripts under ~/.claude/projects. fastclaude.session documents the format and provides the underlying reading and writing functions.

Reading functions and prepare_compaction do not modify transcripts. save_sess replaces a whole session file. append_sess, fork_curated, append_compaction, and compact_session also write session files. Inspect their documentation and target records before calling them.

Prompt history

prompt_hist reads ~/.claude/history.jsonl, Claude Code’s global record of typed prompts. It includes each prompt’s project directory and timestamp. Use it to recover the user’s side of a conversation after transcript deletion or rewriting.

from fastcore.test import *
from aidialog.msg_parts import fmt2hist
from collections import Counter
import json, tempfile, shutil

Fixtures

fastclaude provides captured Agent SDK transcripts in ant_data, including a session and its native fork. Its session documentation explains how to refresh them.

The examples also create a synthetic “flux” session in a temporary project. Later cells modify this session without changing the captured transcripts.

source_path,fork_path = ant_data/'source.jsonl',ant_data/'fork.jsonl'
source_recs,fork_recs = load_recs(source_path),load_recs(fork_path)
source_thread,fork_thread = sess_thread(source_recs),sess_thread(fork_recs)
len(source_thread),len(fork_thread)

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)
back = load_sess(sid, proj)
sid

Searching a session

sess_thread stops at the last compaction boundary. sess_search searches all conversation records, including earlier discussion still in the file. Read the records around a hit to understand its context.

---

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

rec_txt extracts text from nested content blocks. Here the clikernel result is a user record, which rec_role classifies as tool:

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

Count the conversation records by role:

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 second user message contains injected skill content. The user role does not necessarily mean a human wrote the message.

sess_search returns SessHits, an L of matches. Each row shows the record index, role, timestamp, and text around the first match. The result retains all searched records on .recs.

---

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=180, # 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

---

source

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

Find the clikernel result:

hits = sess_search(r'\b42\b', recs=source_recs)
test_eq([h.role for h in hits], ['tool'])
hits

   10 tool      2026-07-17T02:51 42

Selections retain the display format and .recs:

test_eq(hits.attrgot('role'), ['tool'])
test_eq((type(hits[:1]).__name__, hits[:1].recs), ('SessHits', hits.recs))

---

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

Use show_recs to read a slice around a search hit. It truncates long records and can show incomplete exchanges that recs2chat cannot convert.

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 complete

Use showall=True to include bookkeeping records:

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

---

source

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

---

source

PromptHist

def PromptHist(
    *args, **kwargs
):

Prompt-history rows in local time, one line per prompt

prompt_hist displays one prompt per line, oldest first. It converts the history file’s millisecond timestamps to the machine’s local time.

In since, a timestamp without a timezone means local time. prompt_hist converts timestamps with Z or a timezone offset to local time before filtering. You can copy since from a displayed row or a transcript record.

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)

Filter by project, time, or text pattern:

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

Claude Code can resume a session without its thinking-only records. strip_think removes these records. It keeps records containing other content alongside thinking.

---

source

strip_think

def strip_think(
    recs, # Session records
):

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

---

source

is_think_rec

def is_think_rec(
    r, # Session record
):

Whether r is an assistant record containing only thinking blocks

Remove thinking-only records from the captured session:

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

Tool calls can contain large strings, such as file contents in edit inputs or read results. trunc_tools caps strings inside tool_use inputs and tool_result content. It preserves the surrounding structure and adds a count of omitted characters to each truncated string. It returns copies without modifying the originals.

---

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

Use a small cap to see the truncation on this fixture:

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

Check the omitted character count on a larger result:

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)

Use reid_recs to make IDs reproducible. It derives record UUIDs, tool-call IDs, and API IDs from their positions and a key. Matching tool calls and results still share an ID. The original records do not change.

---

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

The same records and key give the same IDs:

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)

Fix the timestamps too for reproducible file contents. When records contain both session_id and sessionId, save_sess updates both:

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

Writing and appending reject duplicate UUIDs:

test_fail(lambda: save_sess(sample+sample, cwd=proj), contains='duplicate')
test_fail(lambda: append_sess(more, apsid, proj), contains='duplicate')

Session names

Use name_sess to give a session a title, as with Claude Code’s /rename command.

---

source

name_sess

def name_sess(
    sid, name, cwd:NoneType=None
):

Name session sid and return it

Renaming appends custom-title and agent-name records without changing the conversation chain.

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)

---

source

sess_by_name

def sess_by_name(
    name, cwd:str='.'
):

Find the first project session transcript with latest custom title name

sess_by_name uses each transcript’s latest custom-title:

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_curated writes a copy of the active conversation with optional thinking removal and tool-content truncation. Resume the returned session ID to compare the edited conversation with the original.

A key makes the fork’s session and record IDs deterministic. Repeating a call with the same key overwrites that fork. The original session stays unchanged.

---

source

fork_curated

def fork_curated(
    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

Fork the sample without thinking and with shorter tool results:

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_curated(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_curated(tsid, proj, mx=10, think=False, key='fork1'), fkid)

mx=0 removes the text but retains the truncation marker. mx=None disables truncation:

z0 = fork_curated(tsid, proj, mx=0, key='fork0')
assert rec_txt(load_sess(z0, proj)[2]).endswith('chars]')

Give the fork a name:

fork_name = 'flux-fork'
fork_named_sid = fork_curated(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

dlg2sess writes a dialog as a Claude Code session that you can resume. It converts replies to tool calls and results in Anthropic’s message format. Prompt content excludes the Solveit serving envelope.

The session ID derives from the dialog name and key. Converting again with the same name and key overwrites that session. dlg2msgs provides the converted messages without writing a session.

---

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: each reply’s tool calls recovered as real blocks, prompts as bare content (no serving envelope)

Represent a tool call and its result in a dialog reply with a fenced json {.tool} block:

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```"

[('user', ['text', 'image', 'text', 'text']),
 ('assistant', ['text', 'tool_use']),
 ('user', ['tool_result']),
 ('assistant', ['text'])]

Convert a dialog containing an image and a tool call:

png = tiny_png
fdlg = Dialog(name='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]

Check the tool-call pairing and PNG attachment:

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

Write the session and read it back:

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)

Resume the session and ask Claude about the tool result. This example spends tokens and does not run in 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

sess2dlg converts a session into a dialog for reading or editing. It first calls recs2chat to convert records into canonical fastllm messages. Tool results use the tool role and share IDs with their calls. Injected skill text remains a user message.

Claude Code uses tool_reference blocks to list deferred tools in tool results. ToolReference registers this block type with Part. norm_tr_parts uses the registry to parse it, and formatted defines its text representation.

---

source

ToolReference.formatted

def formatted():

---

source

ToolReference

def ToolReference(
    tool_name:str=''
)->None:

A deferred-tool reference in a Claude Code tool result

---

source

recs2chat

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

Canonical fastllm messages for the conversation records in recs

Check the captured Skill, Bash, and clikernel calls after conversion:

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

A fallback block records a model switch during a response. recs2chat skips this block because it contains no conversation text. Unsupported assistant block types raise ValueError.

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].text

recs2chat copies each record’s timestamp and UUID into message metadata as created and uid. chat2dlg uses these to create cell metadata and stable message IDs. Converting the same records again preserves references to the resulting messages.

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].meta

Images in tool results render as <media> placeholders. Unknown block types raise KeyError:

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

Changing IDs preserves tool-call pairings:

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 creates one prompt per user turn. It joins text parts with blank lines and converts images to attachments referenced in the prompt. Injected skill content becomes a separate prompt in this example.

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)

Use mx=None to keep tool results in full:

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

A dialog reply retains assistant and tool message content, including tool-call pairings. It does not retain their provenance metadata. Here parsing the reply recovers the original content:

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

Check that conversion to a dialog and back preserves a pasted image:

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`)
    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 selects the active chain and removes thinking-only records before conversion. Here it gives the same dialog as the separate steps 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])

sess2dlg omits bookkeeping such as injected-context attachments, per-turn state, and file snapshots. It preserves system records as raw messages. Their metadata contains the original record under rec and its type:subtype under rec_kind. The latter appears in the summary row for an otherwise empty message.

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

Prompt metadata includes the request’s timestamp and the reply’s usage:

sback = load_sess(ssid, proj)
test_eq(sysdlg.messages[0].meta['timestamp'], sback[0].timestamp)
test_eq(sysdlg.messages[0].meta['usage'], obj2dict(sback[3].message.usage))

dlg2sess restores the saved system records, updating sessionId and parentUuid to join the new chain. It preserves their other fields. History conversion skips tagged raw messages rather than adding them to the conversation:

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

Inside a live session

Claude Code sets CLAUDE_CODE_SESSION_ID in child processes. load_sess() uses it to read the current transcript. This example uses the captured fixture when no live transcript exists.

recs = load_sess() if sess_file().exists() else source_recs
len(recs)

20

Count conversation and bookkeeping records in the transcript:

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)

sess_thread omits records without UUIDs and records outside the active chain. It retains linked bookkeeping records, including this fixture’s attachments. Each record in the active chain links to its predecessor:

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

Resolving session references

resolve_session accepts a UUID or /rename title. It returns the session ID and transcript path.

---

source

resolve_session

def resolve_session(
    ref:NoneType=None, cwd:str='.'
):

Resolve a session id or custom title to (sid,path)

Resolution tries the ID first, then each transcript’s latest custom title. A missing session raises FileNotFoundError.

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 compaction preserves the previous summary and adds a summary of new records. Metadata identifies llmsurgery’s summaries and their /compact display records.

Recognize a synthetic summary by its metadata:

tagged_summary = dict2obj(mk_rec('user', 'Earlier conversation', isCompactSummary=True, llmsurgeryCompact=True))
test_eq(_is_synthetic_compact(tagged_summary), True)

The caveat, command, and stdout records display /compact in Claude Code. They contain no new conversation content.

display_rec = dict2obj(mk_rec('user', '<command-name>/compact</command-name>', llmsurgeryCompact=True))
test_eq(_is_compact_wrapper(display_rec), True)

---

source

split_compaction

def split_compaction(
    recs
):

Return the prior compaction’s text and records added after it

split_compaction finds the latest compact summary on the active chain. It returns that summary’s text and the records after its display records. For an llmsurgery summary it extracts the compact DSL body. For a native Claude Code summary it keeps the whole text.

def _link(prev, r, sid):
    "Link one test record after `prev`"
    r['sessionId'],r['parentUuid'] = sid,prev.get('uuid') if prev else None
    return r

Create a summary and its display record:

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)

Add a new exchange after the display record:

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 new records exclude the /compact display:

test_eq([rec_role(r) for r in new], ['user','assistant'])
test_eq([rec_txt(r) for r in new], ['Continue the work.','Continuing.'])

A native /compact summary also ends the earlier conversation. Subsequent compaction keeps that summary and processes new records after it. It must not include the history the native compaction discarded.

nat_summary = _link(split_asst, mk_rec('user', 'Native summary of everything so far.', isCompactSummary=True), split_sid)
nat_user = _link(nat_summary, mk_rec('user', 'Keep going.'), split_sid)
nat_recs = split_recs + L(dict2obj(r) for r in (nat_summary,nat_user))
nat_prior,nat_new = split_compaction(nat_recs)
test_eq(nat_prior, 'Native summary of everything so far.')
test_eq(nat_new.attrgot('uuid'), [nat_user['uuid']])
nat_prior

'Native summary of everything so far.'

With no earlier compaction, split_compaction returns an empty summary and the full active chain.

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. compact_records creates the session records around it: a system boundary, a summary record and three /compact display records. Preparing these records does not change the transcript. Appending them starts a new active chain.

---

source

sess_meta

def sess_meta(
    recs
):

Latest available session metadata for synthetic records

Records can omit session metadata. sess_meta finds the latest available value of each field for use in synthetic records.

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

Copy the captured fork to a temporary session. Split it after the Bash result to leave a second exchange for incremental compaction.

When refreshing the fixture, inspect fork_thread before choosing the split index. Split immediately after a completed tool result, before the next assistant action. Never separate a tool_use from its matching tool_result.

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

---

source

compact_records

def compact_records(
    recs, content, pre_toks, post_toks
):

Create a tagged compact boundary, summary, and visible /compact records

All five records have an llmsurgeryCompact tag. The boundary includes token counts. The summary holds the continuation document.

---

source

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 returns a continuation document and synthetic records without writing them. It preserves the latest summary, whether native or synthetic, and compacts the records after it. The last five user prompts and their final replies remain untruncated.

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)

Inspect the records before appending 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>

---

source

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)

Check that the active chain contains the five tagged records:

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>

Append the fork’s remaining exchange, containing 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))

25

Prepare another compaction. It preserves the earlier summary and adds the new exchange:

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)

Append it and check that the new summary contains both exchanges:

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>

Use compact_session to prepare and append in one call.

---

source

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

Compact a separate copy of the full fork:

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)

Read the resulting active chain:

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 »

Locked agents

run_locked_agent is for tasks where an agent runs prepared commands and judges their results. For example, it can decide whether a release check passed without inventing new checks. Instructions alone did not prevent our agents from adding grep pipelines and scratch files.

run_locked_agent passes the rules in allowed to a headless fastclaude.astream child. It does not load the parent session’s user settings, hooks, or MCP instructions. These rules restrict command prefixes, not complete shell expressions. Keep the event stream to check what actually ran.

Each item below cost one failed run to learn:

• Use setting_sources=() to exclude user settings. A blocking PreToolUse hook takes precedence over allowed rules. In our headless run, the blocked child returned subtype='success' with an empty result without doing the work.
• Put prepared commands in executable scripts and allow prefixes such as Bash(/path/step1.sh *). Invoke the script directly. Allowing bash /path/... widens the prefix.
• In our headless tests, Bash(echo hi *) also permitted echo hi there | wc -l. We did not observe the compound-command splitting documented for interactive sessions. Review the transcript for pipe suffixes as well as off-prefix commands.
• A nested claude -p launched from a session’s Bash tool failed subscription authentication with “Not logged in”. The astream child authenticated with the same login.
• Keep all events. A final success event does not establish that tools ran. Look for calls without matching results.
• Give the protocol a version string and require the report to echo it. This detects runs using stale prompts.
• End each script’s output with a sentinel line. Require the agent to confirm it saw the sentinel before interpreting the output as complete.

asyncio.wait_for limits the run duration. Cancellation kills the child. Use asyncio.gather to run a batch in parallel.

---

source

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
    cwd:NoneType=None, # Directory the child works in; fastclaude's isolated work dir if None
):

Run a bare headless agent locked to allowed commands, returning (report, raw events)

This live example permits one read-only git command. It spends tokens and does not run automatically.

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, cwd='..')
print(rpt)

The newest commit subject line is: **bump**

Cleanup

Remove the flux sample’s sessions from ~/.claude/projects, along with the scratch project.

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