dlgskill

Read, search, and edit dialogs and notebooks through the llmsurgery Dialog/Message model

Use this whenever the question or edit concerns a notebook’s content: messages, sources, outputs, prompts and their replies.

The hierarchy

Notebook work happens at three levels, and picking the right level is most of using this module well:

• Content (this module): what the messages say and how they change. summary_dlg, find_msgs, view_dlg, the message editing operations, and reply2dlg/dlg2reply for a prompt’s reply.
• Representation (fastcore.nbio): which keys exist, whether a file is schema-valid, whether bytes changed. Start with validate_nb/validate_cell, which name the offending cell; use read_nb directly when the question is about the dict itself.
• Raw text: only when the file will not parse at all.

Dropping a level is correct exactly when the question is about the representation rather than the content (“why does Jupyter reject this file?”, “did that write change any bytes?”). Treat each drop as a signal, though: needing nbio or raw JSON to answer a content question means a higher-level tool was missing. Re-read this module’s docs to check it truly is missing, and then propose adding it rather than repeating the workaround.

Core APIs

Every function takes dlg as a Dialog, an ipynb path, or None meaning the current dialog file (set_dlg/cur_dlg); changes persist automatically only for file-backed dialogs.

• summary_dlg(dlg): one preview line per message.
• find_msgs(pattern, dlg, ...): search by regex, type, error state, heading, or ids; context defaults to 1 (the neighbouring message usually explains the match). Returns live Message objects in a Msgs list (.show(maxlen) for display control), so results are edited directly rather than re-addressed.
• view_dlg(dlg_or_msgs) / view_msg(id) / view_msgs(*ids) / msg2xml(m): full views in the shared item2xml grammar (a prompt’s reply is its <out> section); view_dlg also renders a find_msgs result.
• Structure: add_msg, del_msgs, move_msgs, split_msg, merge_msgs, copy_msgs/cut_msgs/paste_msgs, create_dlg; the %%add_msg magic takes verbatim bodies.
• Text edits: msg_str_replace, msg_strs_replace, msg_insert_line, msg_replace_lines, msg_del_lines (all with re_filter/line-range powers; out=True edits a prompt’s reply or a code message’s outputs literal); python_msgs/ast_msgs for structural rewrites; lnhashview_msg/exhash_msg for hash-verified line edits (needs the exhash package).
• reply2dlg(pmsg)/dlg2reply(sub): explode a reply into note/code messages and back; byte-exact for fmt2hist-clean replies.

Workflow: read_ipynb (or a live Dialog), summary_dlg to orient, find_msgs to locate, view then edit, dlg.validate() to catch model-level damage early, dlg.save() to write back. Editing operations return diffs; reading the diff is the immediate verification.

from fastcore.test import *
from tempfile import mkdtemp
from exhash import lnhash

The current dialog, and finding messages

---

source

cur_dlg

def cur_dlg():

Call self as a function.

---

source

set_dlg

def set_dlg(
    fname, # ipynb path that later calls will default to
):

Set the current dialog file, used by these functions when dlg is None

---

source

Dialog.msg

def msg(
    id, # A `Message` (matched by its id), or an id: exact, or unique prefix
):

The matching Message in this dialog

On-disk cell ids are message ids without the leading underscore, so msg accepts either form, and unique prefixes work like everywhere else in this ecosystem. The fixture below is written to disk too, since path-based calls re-read the file per call:

tdir = Path(mkdtemp())
d = Dialog('demo')
setup = d.mk_message('# Setup', msg_type=snote)
calc = d.mk_message('x = 21\nx*2', msg_type=scode, output=code_output('42'))
ask = d.mk_message('Double it.', msg_type=sprompt, output='It is **42**.')
bad = d.mk_message('1/0', msg_type=scode, output=[dict(output_type='error', ename='E', evalue='boom', traceback=['tb'])])
write_ipynb(d, tdir/'demo.ipynb')
test_eq(d.msg(calc.id), calc)
test_eq(d.msg(calc.id[:5]), calc)  # unique prefixes work too
with expect_fail(KeyError, 'no message'): d.msg('zzzz')

Views

---

source

summary_dlg

def summary_dlg(
    dlg:NoneType=None, # A `Dialog`, ipynb path, or iterable of `Message`s (e.g. a `find_msgs` result); current dialog file if None
    maxlen:int=120, # Maximum characters per line
):

One preview line per message

s = str(summary_dlg(d))
test_eq(len(s.splitlines()), 4)
assert 'reply(13)' in s and '# Setup' in s

---

source

view_dlg

def view_dlg(
    dlg:NoneType=None, # A `Dialog`, ipynb path, or iterable of `Message`s (e.g. a `find_msgs` result); current dialog file if None
    incl_out:bool=False, # Include code outputs?
    only_errors:bool=False, # Show only code messages with error outputs (implies `incl_out`)?
    trunc_out:bool=True, # Truncate each output to ~512 chars?
):

Dialog (or a selection of its messages) as concise XML

---

source

msg2xml

def msg2xml(
    m, incl_out:bool=False, trunc_out:bool=True, ids:bool=True
):

One message as concise XML: content bare inside its type tag, with an <out> section for a code output or a prompt’s reply

v = str(view_dlg(d, incl_out=True))
assert '<prompt id="'+ask.id+'">Double it.<out>It is **42**.</out></prompt>' in v  # a prompt's reply is its out
assert '>x = 21\nx*2<out>42</out></code>' in v
verr = str(view_dlg(d, only_errors=True))
assert '1/0' in verr and 'Setup' not in verr

---

source

view_msgs

def view_msgs(
    *ms, # `Message`s or ids
    dlg:NoneType=None, # `Dialog` or path for id lookup; the current dialog file if None
    nums:bool=True, # Show line numbers?
):

Show several messages, each preceded by a # msg <id> header

---

source

view_msg

def view_msg(
    m, # A `Message`, or an id looked up in `dlg`
    dlg:NoneType=None, # `Dialog` or path when `m` is an id; the current dialog file if None
    nums:bool=True, # Show line numbers?
    view_range:list=None, # Optional 1-based (start, end) range; end=-1 for last line
):

Show message content with optional line numbers

test_eq(str(view_msg(calc)), '1: x = 21\n2: x*2')
test_eq(str(view_msg(calc, view_range=(2,-1))), '2: x*2')
assert str(view_msgs(setup, calc, nums=False)).startswith(f"# msg {setup.id}")

Searching

---

source

find_msgs

def find_msgs(
    re_pattern:str='', # Regex over content (a prompt's reply included), DOTALL+MULTILINE; an invalid regex matches literally
    dlg:NoneType=None, # A `Dialog`, or ipynb path; the current dialog file if None
    msg_type:str=None, # Optional limit by type ('code', 'note', 'prompt', or 'raw')
    only_err:bool=False, # Only code messages with error outputs?
    only_exp:bool=False, # Only messages with an nbdev `#| export` directive?
    ids:str='', # Optionally filter by ids (comma-separated str, or list); results are always in dialog order, whatever order the ids are given
    before:int=0, # Also include n messages before each match
    after:int=0, # Also include n messages after each match
    context:int=None, # Messages of context around matches (default 1, or 0 when `headers_only`)
    limit:int=None, # Max matched messages
    use_case:bool=False, # Case-sensitive matching?
    use_regex:bool=True, # Regex matching (else plain substring)?
    headers_only:bool=False, # Only heading notes (an outline view)?
    header_section:str=None, # Return the section starting with this heading, plus its children
    pred:callable=None, # Extra match criterion, e.g. host-specific flags
)->Msgs: # Live messages, so results can be edited directly

Find messages in dlg (a Dialog, path, or iterable of messages) matching all the given criteria

find_msgs returns the messages themselves, so a search result can be edited in place. A prompt matches on its request or its reply; an invalid regex matches literally (search text arrives verbatim from humans); pred= takes any extra criterion, the channel hosts pass their own flag logic through; header_section follows note-message headings; headers_only gives an outline. context defaults to 1, since the neighbouring message is usually the explanation of whatever matched; pass context=0 for exact hits only:

test_eq(find_msgs('42', d, context=0), [ask])  # matches the reply text
test_eq(find_msgs(r'x\*2', d, msg_type=scode, context=0), [calc])
test_eq(find_msgs(dlg=d, only_err=True, context=0), [bad])
test_eq(len(find_msgs(r'x\*2', d)), 3)  # context defaults to 1: the match plus both neighbours
test_eq(find_msgs(dlg=d, headers_only=True), [setup])
test_eq(find_msgs(header_section='Setup', dlg=d, context=0), list(d.messages))  # everything under the heading
test_eq(find_msgs('42', d.messages[:3], context=0), [ask])  # any message iterable works too
find_msgs('42', d, context=0)[0].content = 'Double it, please.'
test_eq(ask.content, 'Double it, please.')  # live objects: the edit landed on the dialog
test_eq(find_msgs('*42*', d, context=0), [ask])  # an invalid regex falls back to a literal match
test_eq(find_msgs(dlg=d, pred=lambda m: m.msg_type==sprompt, context=0), [ask])  # host-specific extra criterion
xd = Dialog('exports')
exp1 = xd.mk_message('#| export\ndef f(): pass', msg_type=scode)
exp2 = xd.mk_message('def g(): pass', msg_type=scode)
exp2.exported = True
deep = xd.mk_message('## Deep\nbody under the heading', msg_type=snote)
test_eq(find_msgs(dlg=xd, only_exp=True, context=0), [exp1, exp2])
test_eq(find_msgs(header_section='Deep', dlg=xd, context=0), [deep])  # matched on the first line only
test_eq(find_msgs(header_section='## Deep', dlg=xd, context=0), [deep])  # a '#'-prefixed argument pins the level
test_eq(find_msgs(header_section='# Deep', dlg=xd, context=0), [])

Structural operations

---

source

move_msgs

def move_msgs(
    ids, # Message(s) or id(s) to move
    before:NoneType=None, # Move before this message or id
    after:NoneType=None, # Move after this message or id
    dlg:NoneType=None, # A `Dialog`, or ipynb path; the current dialog file if None
):

Move messages, keeping their relative order; returns them

---

source

merge_msgs

def merge_msgs(
    *ids, # Adjacent messages or ids to merge
    dlg:NoneType=None, # A `Dialog`, or ipynb path; the current dialog file if None
):

Merge into the first message (returned, keeping its id): same-type merges keep the type, mixed become a note via merge_content; outputs clear, attachments combine

---

source

split_msg

def split_msg(
    id, # Message or id to split
    *linenos:int, # Split before each of these 1-based lines
    dlg:NoneType=None, # A `Dialog`, or ipynb path; the current dialog file if None
):

Split a message into pieces (returned as Msgs; the first keeps its id): one '\n\n' is absorbed at each cut (so merge_msgs restores blank-line-separated content byte-exactly), meta_attrs fields and a leading #| export copy to every piece, attachments follow their references, and unreferenced ones stay on the first piece

sd = Dialog('s')
sm = sd.mk_message('#| export\ndef f(): pass\ndef g(): pass', msg_type=scode)
i1,i2 = split_msg(sm, 3, dlg=sd)
test_eq([m.content for m in sd.messages], ['#| export\ndef f(): pass', '#| export\ndef g(): pass'])
merge_msgs(i1, i2, dlg=sd)
test_eq(sd.messages[0].content, '#| export\ndef f(): pass\n\ndef g(): pass')
hd = Dialog('hoist')
h1, h2 = hd.mk_message('def a(): pass'), hd.mk_message('#| export\ndef b(): pass')
test_eq(merge_msgs(h1, h2, dlg=hd).content, '#| export\ndef a(): pass\n\ndef b(): pass')  # any exported piece exports the merge
sd.mk_message('tail', msg_type=snote)
move_msgs(i1, after=sd.messages[-1], dlg=sd)
test_eq([m.content[:4] for m in sd.messages], ['tail', '#| e'])

ad = Dialog('atts')
at1, at2 = Attachment(b'x', 'text/plain'), Attachment(b'y', 'text/plain')
am = ad.mk_message(f'see attachment:{at1.id}\nplain tail', msg_type=snote, attachments=[at1, at2])
p1, p2 = split_msg(am, 2, dlg=ad)
test_eq(len(p1.attachments), 2)  # the referenced one, plus the unreferenced one kept on the first piece
test_eq(p2.attachments, [])
jd = Dialog('junction')
j = jd.mk_message('a\n\n\nb', msg_type=snote)
j1, j2 = split_msg(j, 4, dlg=jd)
test_eq((j1.content, j2.content), ('a\n', 'b'))  # one '\n\n' absorbed at the cut: exactly what merge re-inserts
test_eq(merge_msgs(j1, j2, dlg=jd).content, 'a\n\n\nb')  # ...so blank-line-separated content round trips byte-exactly
k1, k2 = split_msg(jd.mk_message('a\n\nb', msg_type=snote), 3, dlg=jd)
test_eq((k1.content, k2.content), ('a', 'b'))
w1, w2 = split_msg(jd.mk_message('a\n\n\nb', msg_type=snote), 3, dlg=jd)  # cut on the blank line: residue stays on the right
test_eq((w1.content, w2.content), ('a', '\nb'))
with expect_fail(ValueError, 'Exactly one'): move_msgs(w2, dlg=jd)

---

source

paste_msgs

def paste_msgs(
    before:NoneType=None, # Insert before this message or id
    after:NoneType=None, # Insert after this message or id
    dlg:NoneType=None, # A `Dialog`, or ipynb path; the current dialog file if None
):

Insert copies of the buffered messages (fresh ids) before/after a message or id; returns the new messages

---

source

cut_msgs

def cut_msgs(
    *ids, # Messages or ids to cut
    dlg:NoneType=None, # A `Dialog`, or ipynb path; the current dialog file if None
):

Copy messages into the paste buffer, then remove them from the dialog

---

source

copy_msgs

def copy_msgs(
    *ids, # Messages or ids to copy
    dlg:NoneType=None, # A `Dialog`, or ipynb path; the current dialog file if None
):

Copy messages into the paste buffer (replacing its contents), for later paste_msgs

copy_msgs(i1, dlg=sd)
pasted = paste_msgs(after=i1, dlg=sd)
test_eq(len(sd.messages), 3)
assert pasted[0].id != sd.msg(i1).id and pasted[0].content == sd.msg(i1).content
cut_msgs(pasted[0].id, dlg=sd)
test_eq(len(sd.messages), 2)

Editing message text

The editing operations run fastcore.tools’ text primitives over a message’s content, or, with out=True, over a prompt’s reply markdown (assignment re-wraps it through prompt_output). Only prompts have text-editable output: for a code message’s outputs, assign msg.output directly. The old behavior of editing a prompt cell’s joined request+reply source is retired. In bulk mode (a list of ids, or ‘all’), no-ops are dropped from the result but errors are reported.

set_dlg(tdir/'demo.ipynb')
diff = msg_str_replace(calc.id, 'x = 21', 'x = 40')
assert '-x = 21' in str(diff) and '+x = 40' in str(diff)
test_eq(read_ipynb(tdir/'demo.ipynb').msg(calc.id).content, 'x = 40\nx*2')

diff = msg_str_replace(ask.id, '**42**', '**80**', out=True)
assert '+It is **80**.' in str(diff)
test_eq(read_ipynb(tdir/'demo.ipynb').msg(ask.id).ai_res, 'It is **80**.')
assert str(msg_str_replace(calc.id, "'42'", "'80'", out=True)).startswith('error:')
assert str(msg_del_lines(setup.id, 1, 1, out=True)).startswith('error:')
bulk = msg_str_replace([setup.id, bad.id], '1/0', 'pass')
test_eq(len(bulk), 2)  # bulk mode reports errors alongside diffs
assert 'error' in bulk[0][1] and '+pass' in bulk[1][1]

---

source

ast_msgs

def ast_msgs(
    repls:list, # (pattern, replacement) ast-grep rules
    *ids, # Messages or ids; all code messages if none given
    dlg:NoneType=None, # A `Dialog`, or ipynb path; the current dialog file if None
):

Apply ast-grep repls to message sources (requires the optional remold package)

---

source

python_msgs

def python_msgs(
    func, # `str -> str` applied to each message's content
    *ids, # Messages or ids; all code messages if none given
    dlg:NoneType=None, # A `Dialog`, or ipynb path; the current dialog file if None
):

Rewrite message sources with func, returning (id, diff) pairs for changed messages

changed = python_msgs(lambda t: t.replace('x', 'y'), dlg=d)
test_eq(len(changed), 1)  # only `calc` contains an x
assert d.msg(calc.id).content.startswith('y = ')

Hash-verified edits

lnhashview_msg and exhash_msg bring exhash’s stale-context-proof addressing to an in-memory message. They import exhash lazily so llmsurgery carries no hard dependency on it:

---

source

exhash_msg

def exhash_msg(
    m, *cmds:tuple, # exhash command tuples, addresses from `lnhashview_msg`
    sw:int=4, # Shift width for indent commands
):

Apply exhash commands to m.content, returning the diff

---

source

lnhashview_msg

def lnhashview_msg(
    m
):

Hash-addressed view of m.content, for exhash_msg (requires the exhash package)

diff = exhash_msg(calc, (lnhash(1, 'y = 21'), 's', '21', '12'))
test_eq(calc.content, 'y = 12\ny*2')
with expect_fail(Exception, 'stale'): exhash_msg(calc, (lnhash(1, 'wrong text'), 'd'))

Working with dialog files

Every function above takes dlg as a Dialog, a path, or None meaning the current dialog file (set_dlg). Operations persist automatically only when the dialog is file-backed, so views never dirty a file, edits on a loaded file save themselves, and edits on an in-memory Dialog are the caller’s to save. write_ipynb’s byte-stability keeps the resulting diffs minimal. A few functions below exist only for the file world:

set_dlg(tdir/'demo.ipynb')
assert '<prompt' in str(view_dlg())
assert str(summary_dlg()).count('\n')==3
assert str(summary_dlg(find_msgs(r'x\*2', msg_type=scode, context=0))).count('\n')==0  # renders a find result too
f = view_dlg(find_msgs(r'x\*2', msg_type=scode, context=0), incl_out=True)
assert '<code' in str(f) and 'Setup' not in str(f)
xd.mk_message('', msg_type=sraw, meta=dict(rec_kind='system', rec={}))
assert 'kind="system"' in str(view_dlg(xd))  # tagged raws are badged, never shown as ordinary content

---

source

create_dlg

def create_dlg(
    fname:str, # path for the new ipynb; must not already exist
    source:str='', # source for its first message
    msg_type:str='code', # 'code', 'note', 'prompt', or 'raw'
):

Create a new dialog file containing one message, returning the Dialog (with path_ stamped)

---

source

del_msgs

def del_msgs(
    *ids, # Messages or ids to delete
    dlg:NoneType=None, # A `Dialog`, or ipynb path; the current dialog file if None
):

Delete messages by id, returning the removed messages

---

source

add_msg

def add_msg(
    source:str, # source for the new message
    msg_type:str='code', # 'code', 'note', 'prompt', or 'raw'
    before:str=None, # message or id to insert before
    after:str=None, # message or id to insert after
    dlg:NoneType=None, # A `Dialog`, or ipynb path; the current dialog file if None
):

Add a new message before/after an existing one (pass exactly one), returning it

nm = add_msg('print(9)', after=calc.id)
test_eq(read_ipynb(tdir/'demo.ipynb').msg(nm).content, 'print(9)')
test_eq(del_msgs(nm)[0].content, 'print(9)')
with expect_fail(KeyError, 'no message'): read_ipynb(tdir/'demo.ipynb').msg(nm)
nd = create_dlg(tdir/'new.ipynb', '# hi', 'note')
test_eq(nd.messages[0].msg_type, snote)
test_eq(read_ipynb(tdir/'new.ipynb').msg(nd.messages[0]).content, '# hi')

s1, s2 = split_msg(calc.id, 2)
test_eq(read_ipynb(tdir/'demo.ipynb').msg(s2).content, 'x*2')
merge_msgs(s1, s2)
test_eq(read_ipynb(tdir/'demo.ipynb').msg(s1).content, 'x = 40\n\nx*2')
copy_msgs(s1)
pids = paste_msgs(after=s1)
test_eq(read_ipynb(tdir/'demo.ipynb').msg(pids[0].id).content, 'x = 40\n\nx*2')
del_msgs(pids[0].id)
python_msgs(lambda t: t.replace('40', '21'), s1)
test_eq(read_ipynb(tdir/'demo.ipynb').msg(s1).content, 'x = 21\n\nx*2')

Under IPython, the %%add_msg magic takes the message body verbatim (no Python quoting): %%add_msg [<path>] before=<id>|after=<id> [code|note|prompt|raw].

---

source

load_ipython_extension

def load_ipython_extension(
    ipython
):

Call self as a function.

---

source

add_msg_magic

def add_msg_magic(
    line, cell
):

Add a new message with the magic body as its source, taken verbatim.