mirror

An ipynb mirror of every session transcript, so the dialog tools work on whole conversation histories

Finding an old discussion means searching conversations, but transcripts are JSONL full of envelope noise, and each host stores them differently. The mirror solves this by convergence: every Claude session and Codex thread is kept as an ordinary dialog ipynb under one root, so nbrg searches all history at once, and find_msgs, summary_dlg, and view_msg read any hit – transcript handling becomes exactly as convenient as dialog handling, on both hosts, and inherits every future improvement to those tools.

The division of labor is: filename is identity (the mirror keeps the transcript’s relative path and stem), mtime is sync (each mirror carries its source’s mtime, so staleness is a pure stat comparison), and meta is truth (nb-level meta records host, source path, and the conversation’s real time span; each message’s meta carries its source record’s created time and uid, and ids are deterministic, so hits stay citable across regenerations). Mirrors live under XDG state (not cache: index keeps mirrors whose transcript was garbage-collected, so the mirror doubles as an archive, and cache directories are fair game for cleanup tools).

from fastcore.test import *
from importlib.resources import files
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

The tests use the same checked-in fixtures as the sess notebook, in throwaway stores, with a throwaway mirror root:

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'),(osrc,'oai')):
    dst.parent.mkdir(parents=True)
    shutil.copy(Path(files('llmsurgery')/'data'/src/'source.jsonl'), 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

A mirror keeps its transcript’s relative path and stem (filename is identity, so the pairing is a name join in either direction) and its source’s exact mtime (mtime is sync). The write is atomic – a temp file replaced into place, timestamped after the rename – so a crashed sync can never leave a half-written mirror that looks fresh:

---

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 brings the whole mirror in sync: a mirror is fresh exactly when its mtime equals its source’s (!= not <, so a transcript restored from backup still triggers a rebuild). Regeneration is idempotent and cheap, so spurious mtime bumps – accidentally opening an old session, say – cost one rebuild and heal themselves. Orphan mirrors, whose transcript was garbage-collected, are deliberately kept: the mirror doubles as an archive. One broken transcript must not abort the run, so failures are collected and reported per file:

---

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

A full pass builds the codex mirror (the ant one is already fresh from above), a second pass rebuilds nothing, and a touched source rebuilds exactly once – with identical message ids, which is what keeps a hit citable across regenerations:

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

Deleting a transcript orphans its mirror rather than removing it, and a truncated or unparseable transcript lands in failed without stopping the run:

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

The result carries the mirror root, so the everyday search is one expression, always searching a current mirror: 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)