# mirror


<!-- WARNING: THIS FILE WAS AUTOGENERATED! DO NOT EDIT! -->

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`](https://AnswerDotAI.github.io/llmsurgery/mirror.html#index)
keeps mirrors whose transcript was garbage-collected, so the mirror
doubles as an archive, and cache directories are fair game for cleanup
tools).

``` python
from fastcore.test import *
from importlib.resources import files
from aidialog.ipynb import read_ipynb
import shutil, tempfile, time
```

## Finding every transcript

------------------------------------------------------------------------

<a
href="https://github.com/AnswerDotAI/llmsurgery/blob/main/llmsurgery/mirror.py#L26"
target="_blank" style="float:right; font-size:smaller">source</a>

### transcripts

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

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

------------------------------------------------------------------------

<a
href="https://github.com/AnswerDotAI/llmsurgery/blob/main/llmsurgery/mirror.py#L45"
target="_blank" style="float:right; font-size:smaller">source</a>

### mirror_sess

``` python
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*

------------------------------------------------------------------------

<a
href="https://github.com/AnswerDotAI/llmsurgery/blob/main/llmsurgery/mirror.py#L36"
target="_blank" style="float:right; font-size:smaller">source</a>

### mirror_path

``` python
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*

``` python
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`](https://AnswerDotAI.github.io/llmsurgery/mirror.html#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:

------------------------------------------------------------------------

<a
href="https://github.com/AnswerDotAI/llmsurgery/blob/main/llmsurgery/mirror.py#L83"
target="_blank" style="float:right; font-size:smaller">source</a>

### sess_index

``` python
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`*

------------------------------------------------------------------------

<a
href="https://github.com/AnswerDotAI/llmsurgery/blob/main/llmsurgery/mirror.py#L62"
target="_blank" style="float:right; font-size:smaller">source</a>

### index

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

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

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

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

## Cleanup

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