mirror

Convert Claude Code and Codex transcripts to dialog notebooks for searching, reading, and archiving conversations

Claude Code and Codex store conversations in different JSONL formats. index converts their transcripts to dialog notebooks under a shared root. Search those notebooks with nbrg. Read matches with find_msgs, summary_dlg, and view_msg.

The mirror also serves as an archive. It retains notebooks after their source transcripts disappear. Its default location is $XDG_STATE_HOME/llmsurgery/mirror, outside the cache directories that cleanup tools can remove.

Notebook metadata records the host, source path, and conversation time span. Message metadata retains the source record’s created time and uid. Unchanged messages keep their IDs across rebuilds, preserving references to search results.

from fastcore.test import *
from importlib.resources import files
from fastclaude.session import ant_data
from aidialog.ipynb import read_ipynb
import shutil, tempfile, time

Finding every transcript

---

source

transcripts

def transcripts(
    ant_root:NoneType=None, # Claude sessions store; `ant.SESSIONS` if None
    codex_home:NoneType=None, # Codex home; `oai.CODEX_HOME` if None
):

Every transcript in both hosts’ stores, as (host, store, path) rows

Create temporary transcript stores from the checked-in fixtures:

aroot,home,mroot = Path(tempfile.mkdtemp()),Path(tempfile.mkdtemp()),Path(tempfile.mkdtemp())
asrc = aroot/'-Users-me-proj'/'ab12cd34-0000-4000-8000-000000000000.jsonl'
osrc = home/'sessions'/'2026'/'01'/'24'/'rollout-2026-01-24T02-44-30-ef56ab78.jsonl'
for dst,src in ((asrc,ant_data/'source.jsonl'),(osrc,Path(files('llmsurgery')/'data'/'oai'/'source.jsonl'))):
    dst.parent.mkdir(parents=True)
    shutil.copy(src, dst)
ts = transcripts(aroot, home)
test_eq([(h,p.name) for h,s,p in ts], [('ant',asrc.name),('oai',osrc.name)])
ts

Syncing one transcript

Each notebook keeps its transcript’s relative path under an ant or oai directory. It replaces the .jsonl extension with .ipynb.

mirror_sess writes to a temporary file before atomically replacing the destination. It then copies the source’s mtime to the notebook. An interrupted write cannot leave a partially written notebook at the destination.

---

source

mirror_sess

def mirror_sess(
    host, # `'ant'` or `'oai'`
    store, # The store root `path` was found under
    path, # Transcript path
    root:NoneType=None, # Mirror root; `MIRROR` if None
):

Write the dialog mirror of the transcript at path, stamping the source’s mtime; returns the mirror’s path

---

source

mirror_path

def mirror_path(
    host, # `'ant'` or `'oai'`
    store, # The store root `path` was found under
    path, # Transcript path
    root:NoneType=None, # Mirror root; `MIRROR` if None
):

Where the transcript at path mirrors to

mp = mirror_sess(*ts[0], root=mroot)
test_eq(mp, mroot/'ant'/'-Users-me-proj'/'ab12cd34-0000-4000-8000-000000000000.ipynb')
test_eq(mp.stat().st_mtime_ns, asrc.stat().st_mtime_ns)
md = read_ipynb(mp)
test_eq(md.meta['llmsurgery']['host'], 'ant')
assert md.meta['llmsurgery']['created'] <= md.meta['llmsurgery']['last']
md.summary()

Reindexing

index rebuilds a notebook whenever its mtime differs from the transcript’s. The comparison uses !=, not <, to detect older transcripts restored from backups. An mtime change without a content change costs one rebuild. Use force=True to rebuild everything after changing the conversion code.

A failed conversion does not stop the remaining conversions. The result lists rebuilt notebooks in built and errors by source file in failed. It also includes the count of unchanged notebooks in fresh.

---

source

sess_index

def sess_index(
    force:bool=False, # Rebuild every mirror, e.g. after the conversion logic changes?
    verbose:bool=False, # Print each rebuilt mirror?
):

Sync the session mirror at MIRROR

---

source

index

def index(
    root:NoneType=None, # Mirror root; `MIRROR` if None
    ant_root:NoneType=None, # Claude sessions store; `ant.SESSIONS` if None
    codex_home:NoneType=None, # Codex home; `oai.CODEX_HOME` if None
    force:bool=False, # Rebuild every mirror, e.g. after the conversion logic changes?
    verbose:bool=False, # Print each rebuilt mirror?
):

Sync the mirror with both hosts’ stores: rebuild stale, keep orphans, collect failures; the result carries the mirror root

Build the remaining notebook, then check that another pass rebuilds nothing:

r1 = index(mroot, aroot, home)
test_eq((len(r1.built), r1.fresh, len(r1.failed)), (1, 1, 0))
r2 = index(mroot, aroot, home)
test_eq((len(r2.built), r2.fresh), (0, 2))
r2

Check that a rebuild preserves message IDs:

ids1 = [m.id for m in read_ipynb(mp).messages]
os.utime(asrc)
r3 = index(mroot, aroot, home)
test_eq(len(r3.built), 1)
test_eq([m.id for m in read_ipynb(mp).messages], ids1)
test_eq(len(index(mroot, aroot, home, force=True).built), 2)
r3

Check that indexing retains deleted transcripts’ notebooks and reports malformed JSON:

asrc.unlink()
test_eq(index(mroot, aroot, home).fresh, 1)
assert mp.exists()
bad = asrc.parent/'bbbbbbbb-0000-4000-8000-000000000000.jsonl'
bad.write_text('not json\n')
rf = index(mroot, aroot, home)
test_eq(len(rf.failed), 1)
test_eq(rf.failed[0][0], bad)
assert not mirror_path('ant', aroot, bad, mroot).exists()
rf.failed

Search after indexing with nbrg(pat, index().root).

test_eq(index(mroot, ant_root=aroot, codex_home=home).root, mroot)

Cleanup

for p in (aroot, home, mroot): shutil.rmtree(p)