oai

Read, write, and build Codex session rollouts

Find, read, search, and edit Codex session rollouts on this machine.

Rollout layout

Codex stores rollouts under CODEX_HOME/sessions in date partitions. The filename includes the thread UUID. Use cur_thread for a running thread’s id, rollout_file to locate its JSONL file, or project_thread for the newest thread in a project. resolve_thread accepts an id or unique prefix. Thread-name lookup requires app-server.

A rollout records events in append order. Each line has a timestamp, record type, and payload. response_item records contain Responses API items. response_items extracts every such item. active_items reconstructs the history after the latest compaction replacement. load_rollout retains the complete event log.

prompt_hist reads user messages from thread rollouts. Finding the right thread is part of recovering earlier prompts.

Reading workflow

For an earlier discussion or decision, start with thread2dlg. It converts the active history into an aidialog dialog. Use d.summary() for a map with message sizes and a separate line for each prompt’s reply. Search with d.find_msgs(pat), then read with view_msg or view_msgs. Read aidialog.dlgskill for these tools. Save with write_ipynb to search across notebooks using rgapi.nbrg. The saved prompt sources include their replies.

Use records directly when you need their envelopes or want to edit a rollout. Locate it with cur_thread, project_thread, or resolve_thread. Read it with load_rollout. Select the current history with active_items, search with item_search, and inspect slices with show_items. Search hits retain the original item in .item. These tools avoid the encoded data and protocol noise in raw JSONL.

Reading functions do not modify rollouts. App-server handles ordinary thread creation, forking, naming, and native compaction. Synthetic compaction functions append JSONL directly. Close Codex before using them. Read their contracts and inspect the prepared record before appending it.

Codex stores each conversation as a thread with a rollout file. App-server handles creation, history injection, resume, naming, and forks. The append-only rollout contains Responses API items alongside turn context, UI events, world state, and compaction records.

You can inject a synthetic history through app-server and resume it as a conversation. This supports editable templates and worked examples without constructing Codex’s storage metadata yourself.

from collections import Counter
from fastcore.test import *
from aidialog.msg_parts import fmt2hist
from importlib.resources import files
import tempfile

Where sessions live

Rollout filenames end with the thread UUID inside date-partitioned folders under CODEX_HOME/sessions. Commands that Codex starts inherit CODEX_THREAD_ID. A long-lived configured MCP server does not receive a new environment for each thread.

source

cur_thread

def cur_thread():

The current Codex thread id, when running inside Codex

rollout_file searches all date partitions for a UUID or unique prefix. It raises on ambiguous matches rather than selecting one arbitrarily.

source

rollout_file

def rollout_file(
    thread_id, # Thread UUID or unique id prefix
    codex_home:pathlib.PosixPath=Path('/home/runner/.codex'), # Codex home containing sessions
):

The persisted rollout for thread_id, if it exists

test_eq(rollout_file('missing-thread', tempfile.mkdtemp()), None)

with tempfile.TemporaryDirectory() as td:
    ses = Path(td)/'sessions'/'2026'/'08'
    ses.mkdir(parents=True)
    for n in ['rollout-2026-08-05-aaaa1111.jsonl','rollout-2026-08-05-aaaa2222.jsonl','rollout-2026-08-05-bbbb.jsonl']: (ses/n).touch()
    test_eq(rollout_file('bbbb', td), ses/'rollout-2026-08-05-bbbb.jsonl')
    with expect_fail(contains='matches 2 sessions'): rollout_file('aaaa', td)
    res = rollout_file('aaaa1', td)
res.name

'rollout-2026-08-05-aaaa1111.jsonl'

load_recs reads every JSONL record, including records other than response_item.

source

load_recs

def load_recs(
    path
):

Rollout records read directly from JSONL path

load_rollout finds the requested or current thread and reads its complete rollout.

source

load_rollout

def load_rollout(
    thread_id:NoneType=None, # Thread UUID; `cur_thread()` if None
    codex_home:pathlib.PosixPath=Path('/home/runner/.codex'), # Codex home containing sessions
):

The records of a persisted Codex thread

Outside a Codex command process, use project_thread when no thread id is available. It finds the most recently modified rollout whose session metadata matches the project directory. codexdojo -r uses this lookup. An MCP server can use it too when choosing the newest project thread is appropriate.

source

project_thread

def project_thread(
    cwd:str='.', # Project directory
    codex_home:pathlib.PosixPath=Path('/home/runner/.codex'), # Codex home containing sessions
):

The newest (thread_id, rollout_path) for cwd

home = Path(tempfile.mkdtemp())
sessions = home/'sessions'
sessions.mkdir()
proj = home/'project'
proj.mkdir()
path = sessions/'rollout-thread-x.jsonl'
path.write_text(json.dumps(dict(type='session_meta',payload=dict(id='thread-x',cwd=str(proj))))+'\n')

A newer file without session metadata does not qualify. This lookup returns the older file that identifies the project:

newer = sessions/'rollout-newer.jsonl'
newer.write_text('{}\n')
mt = path.stat().st_mtime+1
os.utime(newer, (mt,mt))
test_eq(project_thread(proj,home),('thread-x',path))
shutil.rmtree(home)

Creating dummy data

The checked-in fixtures contain app-server output from synthetic conversations, not user sessions. To create them, we injected ordinary and custom calls into a temporary Codex home. Another app-server then resumed and forked the source thread. The fixtures retain the records Codex wrote.

oai_data = Path(files('llmsurgery')/'data'/'oai')
source_path,fork_path = oai_data/'source.jsonl',oai_data/'fork.jsonl'
source_recs,fork_recs = load_recs(source_path),load_recs(fork_path)
Counter(source_recs.attrgot('type')),Counter(fork_recs.attrgot('type'))

(Counter({'response_item': 12,
          'session_meta': 1,
          'world_state': 1,
          'turn_context': 1}),
 Counter({'response_item': 13,
          'session_meta': 2,
          'world_state': 1,
          'turn_context': 1,
          'event_msg': 1}))

Read the owning thread id from the first session metadata record:

source

thread_id

def thread_id(
    recs, # Rollout records
):

The owning thread id from the first session metadata record

source_tid,fork_tid = thread_id(source_recs),thread_id(fork_recs)
test_ne(source_tid, fork_tid)
source_tid,fork_tid

('019f78da-e13e-72a2-9c10-aff27ecc7dfd',
 '019f78e2-3278-71f1-83db-beb79144d642')

The active history

response_item payloads form the initial conversation history. A compacted record replaces the earlier history with its replacement_history. Later response items extend that replacement. If a rollout contains multiple compactions, the latest one determines the active history.

source

active_items

def active_items(
    recs, # Rollout records
):

The Responses API history Codex reconstructs from recs

source

split_compaction

def split_compaction(
    recs, # Rollout records
):

The latest native replacement history and response items recorded after it

source

response_items

def response_items(
    recs, # Rollout records
):

All recorded Responses API items, including superseded history, with envelope timestamps when present

test_eq(active_items(source_recs), response_items(source_recs))

In this fixture, the old request remains in the file. Only the replacement history and subsequent items belong to the active conversation.

old = [dict(type='response_item', payload=dict(type='message', role='user',
    content=[dict(type='input_text', text='Old request')]))]
replacement = [dict(type='message', role='user', content=[dict(type='input_text', text='Earlier conversation')]),
    dict(type='compaction', id='cmp_fixture', encrypted_content='encrypted fixture')]
compacted = dict(type='compacted', payload=dict(message='', replacement_history=replacement,
    window_number=1, first_window_id='w1', previous_window_id='w1', window_id='w2'))
suffix = [dict(type='response_item', payload=dict(type='message', role='user',
    content=[dict(type='input_text', text='Continue the work.')]))]
compact_recs = L(dict2obj(x) for x in [*old,compacted,*suffix])
len(compact_recs)

split_compaction returns the replacement and subsequent items separately. active_items joins them:

prior,new = split_compaction(compact_recs)
test_eq([x['type'] for x in prior], ['message','compaction'])
test_eq([x['type'] for x in new], ['message'])
test_eq(active_items(compact_recs), prior+new)

Forking

Use app-server to fork a thread. It creates a new thread id and rollout metadata while retaining the injected history. Copying JSONL alone does not perform that operation.

source_types = [o['type'] for o in response_items(source_recs)]
fork_types = [o['type'] for o in response_items(fork_recs)]
test('custom_tool_call', source_types, in_)
test('custom_tool_call', fork_types, in_)
for typ in ('function_call','function_call_output','custom_tool_call','custom_tool_call_output'):
    test_eq(source_types.count(typ),fork_types.count(typ))
test_eq(fork_types.count('message'),source_types.count('message')+1)
source_types

['message',
 'message',
 'message',
 'message',
 'message',
 'function_call',
 'function_call_output',
 'message',
 'message',
 'message',
 'custom_tool_call',
 'custom_tool_call_output']

Writing a session

Responses history uses a flat list of items. Message items contain text. Tool calls pair with output items through call_id.

The captured Codex records use custom exec calls with JavaScript input for host operations. The inner call is commonly tools.exec_command or an MCP function such as tools.mcp__clikernel__execute.

source

codex_output

def codex_output(
    call_id, output
):

The result of a Responses API function call

source

codex_call

def codex_call(
    name, arguments, call_id:NoneType=None
):

A Responses API function call item

source

codex_msg

def codex_msg(
    role, text
):

A Responses API message item

parse_exec extracts one nested operation from a recognized JavaScript wrapper. It supports exec_command output forwarding and known MCP content-forwarding forms. _MCP_TAILS lists the accepted display-only MCP forms. The parser normalizes the result variable and whitespace before matching them.

Unrecognized JavaScript, including the older stream wrapper, remains a raw exec call. exec_input generates the wrapper used here for editable nested calls.

source

parse_exec

def parse_exec(
    src
):

The logical nested tool call in a simple Codex exec input, or None

exec_input constructs stable JavaScript for one logical call. codex_custom_call and codex_custom_output put it in paired Responses items:

source

codex_custom_output

def codex_custom_output(
    call_id, output
):

The result of a Codex custom tool call

source

codex_custom_call

def codex_custom_call(
    name, arguments, call_id:NoneType=None, item_id:NoneType=None
):

A Codex custom exec call wrapping one logical tools.* operation

source

exec_input

def exec_input(
    name, arguments
):

Stable exec JavaScript for one logical tools.* call

cc = codex_custom_call('tools.mcp__clikernel__execute', dict(code='6*7'))
test_eq(parse_exec(cc['input']).name, 'tools.mcp__clikernel__execute')
test_eq(parse_exec(cc['input']).arguments, {'code':'6*7'})
test_eq(codex_custom_output(cc['call_id'], '42')['call_id'], cc['call_id'])
exc = codex_custom_call('tools.exec_command', dict(cmd='pwd'))
test_eq(parse_exec(exc['input']).name, 'tools.exec_command')
test_eq(parse_exec(exc['input']).arguments, {'cmd':'pwd'})

The captured wrapper contains the nested tool’s name and arguments. The parser also accepts supported template literals, including String.raw:

captured = r'''const r = await tools.mcp__clikernel__execute({code:"dojo_start()"});
for (const c of (r.content || [])) c.type === "image" ? image(c) : c.text && text(c.text);
'''
test_eq(parse_exec(captured).name, 'tools.mcp__clikernel__execute')
test_eq(parse_exec(captured).arguments, {'code':'dojo_start()'})
templ = captured.replace('{code:"dojo_start()"}', r'{code:`x = r"\\b"`}')
test_eq(parse_exec(templ).arguments, {'code':r'x = r"\b"'})
raw = captured.replace('{code:"dojo_start()"}', r'{code:String.raw`x = "\n"`}')
test_eq(parse_exec(raw).arguments, {'code':r'x = "\n"'})

The stream wrapper is not a supported nested call. The compact MCP spelling matches and is the form exec_input emits:

stream = r'''const r = await tools.write_stdin({session_id:17,chars:"6*7\n"});text(r.output)'''
test_eq(parse_exec(stream), None)
cur = r'''const r=await tools.mcp__clikernel__execute({code:"dojo_start()"}); for (const c of (r.content||[])) c.type==="text"?text(c.text):c.type==="image"?image(c):null;'''
test_eq(parse_exec(cur).name, 'tools.mcp__clikernel__execute')
test_eq(parse_exec(cur).arguments, {'code':'dojo_start()'})
assert 'c.type==="text"' in exec_input('tools.mcp__clikernel__execute', dict(code='6*7'))

reid_items replaces captured random ids with ids derived from item position and a salt. It recursively updates matching references, including paired output call_id fields and metadata.

source

reid_items

def reid_items(
    items, # Responses API items
    key:str='', # Salt for deterministic IDs
):

Copies of items with response item and paired call IDs re-derived

openai_codex.AsyncCodex is the Python client used here for app-server. It starts, resumes, reads, names, and forks threads. codex_client selects the codex executable on PATH to keep SDK and CLI storage formats aligned. Tests can pass a separate Codex home.

This code adds AsyncThread.inject_items for thread/inject_items. It uses the SDK’s generic request method and typed response model. The create and append helpers below build on that method.

source

AsyncThread.inject_items

async def inject_items(
    items
):

Append raw Responses API items to this thread’s model history

source

codex_client

def codex_client(
    codex_home:NoneType=None
):

An AsyncCodex on the PATH codex binary, optionally with a redirected CODEX_HOME

create_thread starts a thread with prefilled history. append_thread resumes an existing thread and injects more items:

source

AsyncCodex.append_thread

async def append_thread(
    thread_id, items, **kwargs
):

Resume thread_id and append raw Responses items

source

AsyncCodex.create_thread

async def create_thread(
    items, cwd:NoneType=None, **kwargs
):

Start a persisted thread whose history is pre-filled with items

Synthetic tool calls

source

custom_turn

def custom_turn(
    prompt, name, arguments, output, answer
):

A complete synthetic Codex custom-tool turn

source

tool_turn

def tool_turn(
    prompt, name, arguments, output, answer
):

A complete synthetic function-call turn

sample = tool_turn('Measure the flux.', 'flux_meter', {}, 'flux: 41.7 kilofinches', 'The flux is 41.7 kilofinches.')
kernel_sample = custom_turn('Evaluate the expression.', 'tools.mcp__clikernel__execute', dict(code='6*7'), '42', 'The result is 42.')
[(o['type'],o.get('name')) for o in kernel_sample]

[('message', None),
 ('custom_tool_call', 'exec'),
 ('custom_tool_call_output', None),
 ('message', None)]

A sample session

The source fixture includes both tool examples. App-server created it, then a second server resumed it and injected the custom call. No model ran. Tests read the checked-in records. The disabled live example shows how to create and resume a fixture through app-server:

source_items = response_items(source_recs)
test('flux: 41.7 kilofinches', repr(source_items), in_)
test('custom_tool_call', source_items.attrgot('type'), in_)
len(source_items)

with tempfile.TemporaryDirectory() as d:
    home = Path(d)
    proj = home/'project'
    proj.mkdir()
    items = reid_items([*sample,*kernel_sample,codex_msg('user','x'*(70*1024))], 'live')
    async with codex_client(home) as codex: thread = await codex.create_thread(items, cwd=proj)
    async with codex_client(home) as codex: resumed = await codex.thread_resume(thread.id)
    test_eq(resumed.id, thread.id)
    print(thread.id, rollout_file(thread.id, home))

Reading a session

For a complete record log, use load_rollout. For the history the model would see next, use active_items.

test_eq(thread_id(source_recs), source_tid)
test_eq(len(active_items(source_recs)), len(response_items(source_recs)))
Counter(source_recs.attrgot('type'))

Counter({'response_item': 12,
         'session_meta': 1,
         'world_state': 1,
         'turn_context': 1})

Searching a session

source

item_txt

def item_txt(
    item
):

Every readable string in a Responses item, joined

item_role reports the conversation role of an item. conv_items selects messages and ordinary or custom tool calls and results:

source

conv_items

def conv_items(
    items
):

Conversation-bearing items, excluding reasoning and compaction

source

item_role

def item_role(
    item
):

Conversation role of a Responses item

item_search provides the Responses equivalent of sess_search. Each ItemHits row shows text around the match and retains its item in .item. The result’s .hist contains the searched conversation items:

source

item_search

def item_search(
    pat, # Regex to find
    items, # Responses items
    maxlen:int=180, # Preview characters around the match
):

Search model-visible Responses items: ItemHits rows, each carrying its own item on .item, with every searched item on .hist

source

ItemHits

def ItemHits(
    items:NoneType=None, *rest, use_list:bool=False, match:NoneType=None
):

Search hits with one match-centered preview per line, each carrying its item on .item

Use show_items to display one readable block per conversation item:

source

show_items

def show_items(
    items, # Responses items
    mx:int=500, # Text characters per item
):

Readable model history for items

hits = item_search(r'41\.7 kilofinches', source_items)
test_eq([h.role for h in hits], ['tool','assistant'])
hits

    6 tool       flux: 41.7 kilofinches
    7 assistant  The flux is 41.7 kilofinches.

ItemHits is an L, so you can select a few hits without losing their display or the .hist reference. Use a hit’s index to read the surrounding conversation:

test_eq(hits.attrgot('role'), ['tool','assistant'])
test_eq((type(hits[:1]).__name__, hits[:1].hist), ('ItemHits', hits.hist))

show_items(hits.hist[max(0,hits[0].i-1):hits[0].i+2], mx=120)

--- assistant:function_call ---
flux_meter
{}
--- tool:function_call_output ---
flux: 41.7 kilofinches
--- assistant:message ---
The flux is 41.7 kilofinches.

Prompt history

prompt_hist recovers user prompts from per-thread rollouts. Each row identifies the thread, project, timestamp, and text. The function does not depend on a separate prompt-history file.

source

prompt_hist

def prompt_hist(
    project:NoneType=None, # Only rollouts whose session cwd is this path
    since:NoneType=None, # Only record timestamps at or after this ISO string
    pat:NoneType=None, # Only prompt text matching this regex
    codex_home:pathlib.PosixPath=Path('/home/runner/.codex'), # Codex home containing sessions
):

User prompts retained in Codex rollouts, oldest first

source

PromptHist

def PromptHist(
    *args, **kwargs
):

Codex prompt-history rows, one line per user message

Curating a captured session

Curation removes large reasoning items and can shorten tool arguments and outputs. The helpers return new dictionaries. For recognized custom calls, truncation rebuilds a valid JavaScript wrapper around the shorter arguments. Truncated arguments are not necessarily safe or meaningful to execute.

source

trunc_tools

def trunc_tools(
    items, # Responses items
    mx:int=2000, # Maximum characters per string in tool arguments and results
):

Copies of items with tool arguments and outputs truncated

source

strip_reasoning

def strip_reasoning(
    items
):

Drop Responses API reasoning items

curate_items selects active history, removes reasoning, optionally truncates tool strings, and derives new ids:

source

curate_items

def curate_items(
    recs, # Captured rollout records
    key:str='', # Salt for deterministic IDs
    mx:NoneType=None, # Optional tool string cap
):

Extract active history, drop reasoning, optionally truncate tools, and re-identify

curated = curate_items(source_recs, 'fixture')
test_eq(curated, curate_items(source_recs, 'fixture'))
calls = [o for o in curated if o['type'] in ('function_call','custom_tool_call')]
outs = [o for o in curated if o['type'] in ('function_call_output','custom_tool_call_output')]
test_eq([o['call_id'] for o in calls], [o['call_id'] for o in outs])
test_ne(calls[0]['call_id'], first(o for o in source_items if o['type'] in ('function_call','custom_tool_call'))['call_id'])

Session names

Thread names belong to Codex’s index. Set them through app-server’s thread/name/set, not by appending a rollout record. Use the thread id to resume it elsewhere.

async with codex_client() as codex: await (await codex.thread_resume(cur_thread())).set_name('oai fixture')

From dialogs

dlg2items converts an authored dialog through canonical history, as the Claude converter does. Ordinary tools become function_call items. Supported tools. names become custom exec items with JavaScript wrappers. This lets a dialog expose tools.mcp__clikernel__execute(code=...) as an editable tool call.

source

dlg2items

def dlg2items(
    dlg, # Dialog ending with a prompt
    aim_info:NoneType=None, # Model capability dict for media handling
):

Responses API items for dlg, including native custom exec calls

The flux dialog produces an ordinary function call and a matching output:

def tool_dtl(func,args,result):
    d = json.dumps(dict(id='call1', name=func, args=args, result=result))
    return f"```json {{.tool}}\n{d}\n```"

fdlg = Dialog(name='flux')
fdlg.mk_message('Measure the flux please.',msg_type=sprompt,
    output=f"Let me check.\n\n{tool_dtl('flux_meter',{},'flux: 41.7 kilofinches')}\n\nThe flux is 41.7 kilofinches.")
fitems = dlg2items(fdlg)
test_eq([o['type'] for o in fitems], ['message','message','function_call','function_call_output','message'])
test_eq(fitems[2]['call_id'],fitems[3]['call_id'])

Use tools.mcp__clikernel__execute to produce a custom exec call instead:

kdlg = Dialog(name='kernel')
kdlg.mk_message('Evaluate it.',msg_type=sprompt,
    output=f"{tool_dtl('tools.mcp__clikernel__execute',{'code':'6*7'},'42')}\n\nThe result is 42.")
kitems = dlg2items(kdlg)
test_eq([o['type'] for o in kitems], ['message','custom_tool_call','custom_tool_call_output','message'])
test_eq(parse_exec(kitems[1]['input']).name, 'tools.mcp__clikernel__execute')

A bare MCP name also works. Conversion adds the tools. prefix:

ndlg = Dialog(name='neutral')
ndlg.mk_message('Evaluate it.',msg_type=sprompt,
    output=f"{tool_dtl('mcp__clikernel__execute',{'code':'6*7'},'42')}\n\nThe result is 42.")
nitems = dlg2items(ndlg)
test_eq([o['type'] for o in nitems], ['message','custom_tool_call','custom_tool_call_output','message'])
test_eq(parse_exec(nitems[1]['input']).name, 'tools.mcp__clikernel__execute')

Back to dialogs

items2chat converts messages, ordinary calls, and custom calls to canonical messages. A server-side web_search_call has no separate output item. Conversion supplies a result such as server tool completed to keep the call paired, as the UI does.

Developer messages, reasoning, compaction records, ghost snapshots, and tool-search records are not editable conversation turns. items2chat omits them. items2dlg separately retains non-user/assistant messages and compaction items as tagged raw cells for dlg2items to restore.

source

items2chat

def items2chat(
    items, # Responses API items
):

Canonical fastllm messages for editable conversation items

items2dlg returns an editable dialog. Tagged raw cells retain developer messages and compaction items before the next prompt. thread2dlg loads a thread’s active history before conversion, or accepts records you already loaded:

source

thread2dlg

def thread2dlg(
    thread_id:NoneType=None, # Thread UUID; current thread if None
    codex_home:pathlib.PosixPath=Path('/home/runner/.codex'), # Codex home
    name:NoneType=None, # Dialog name
    mx:int=2000, # Maximum rendered tool string length
    recs:NoneType=None, # Already-loaded rollout records
):

The active model history of a Codex thread as a dialog

source

items2dlg

def items2dlg(
    items, # Responses API items
    name:str='thread', # Dialog name
    mx:int=2000, # Maximum rendered tool string length
):

Editable dialog for items, preserving developer and compaction items as tagged raws

response_items copies an envelope timestamp onto its item when present. items2chat records that as created metadata, which chat2dlg carries into cells. Items from replacement_history have no enclosing response-item timestamp. Conversion does not invent one:

test_eq(source_items[0]['timestamp'], first(r for r in source_recs if r['type']=='response_item')['timestamp'])
tsmsgs = items2chat(conv_items(source_items))
assert all(m.meta and m.meta['created'] for m in tsmsgs)
tsmsgs[0].meta

{'created': '2026-07-19T05:30:38.286Z'}

round_dlg = items2dlg(kitems,'kernel back',mx=None)
back = dlg2items(round_dlg)
test_eq([o['type'] for o in back], [o['type'] for o in kitems])
test_eq(parse_exec(back[1]['input']).arguments['code'], '6*7')
test_eq(back[2]['call_id'],back[1]['call_id'])

srch = dict(type='web_search_call', status='completed', action=dict(type='search', query='MathML browser support'))
plumbing = [dict(type='ghost_snapshot', ghost_commit=dict(id='9138')), dict(type='tool_search_call', call_id='c1', arguments='{}')]
pair = items2chat([srch, *plumbing])
test_eq([m.role for m in pair], ['assistant','tool'])
test_eq(pair[1].content[0].text, 'server tool completed')
test_eq(pair[0].content[0].id, pair[1].content[0].id)
with expect_fail(ValueError, contains='unsupported response item: mcp_call'): items2chat([dict(type='mcp_call')])
pair[0].content[0]

🔧 web_search({'type': 'search', 'query': 'MathML browser support'})

id: ws_0
server: False
raw: None

The fixture’s conversation remains searchable after conversion to a dialog:

source_dlg = thread2dlg(name='source fixture',recs=source_recs)
test('41.7 kilofinches','\n'.join(m.content+m.ai_res for m in source_dlg.messages),in_)
source_dlg

source fixture

/private/var/folders/51/b2_szf2945n072c0vj2cyty40000gn/T/tmpgltt9j…
Measure the flux. ⇒ [{‘output_type’: ‘display_data’, ‘metadata’: {‘is_ai_res’: …
And the humidity? ⇒ [{‘output_type’: ‘display_data’, ‘metadata’: {‘is_ai_res’: …

Inside a live session

Commands started by Codex can use CODEX_THREAD_ID. A configured MCP server needs the id from its caller because it does not inherit each thread’s environment. This example uses the fixture when no live rollout is available:

live_tid = cur_thread()
live_recs = load_rollout(live_tid) if live_tid and rollout_file(live_tid) else source_recs
live_tid or source_tid,Counter(live_recs.attrgot('type'))

('019f78da-e13e-72a2-9c10-aff27ecc7dfd',
 Counter({'response_item': 12,
          'session_meta': 1,
          'world_state': 1,
          'turn_context': 1}))

Resolving session references

Rollout files identify threads by UUID, not by their index names. resolve_thread accepts the current id or an explicit id or unique prefix. Use app-server for name lookup.

source

resolve_thread

def resolve_thread(
    ref:NoneType=None, # Thread UUID; current thread if None
    codex_home:pathlib.PosixPath=Path('/home/runner/.codex'), # Codex home
):

Resolve a thread UUID to (thread_id, rollout_path)

with expect_fail(FileNotFoundError): resolve_thread('missing-thread',tempfile.mkdtemp())

Recognizing earlier compactions

split_compaction locates the latest compacted record by type. It preserves the replacement messages and encrypted compaction item without interpreting their contents. It returns response items appended after that record separately.

prior,new = split_compaction(compact_recs)
test_eq(prior[-1]['type'],'compaction')
test_eq(item_txt(new[0]),'Continue the work.')
test_eq(response_items(compact_recs)[0]['content'][0]['text'],'Old request')

Native compaction

thread/compact/start asks Codex to compact a thread. The replacement includes encrypted compaction content that llmsurgery preserves without decoding. This operation calls a model, so its live example does not run in the notebook tests. The next section constructs a readable synthetic replacement instead.

async with codex_client() as codex:
    thread = await codex.thread_resume(cur_thread())
    await thread.compact()

Synthetic compaction

A synthetic compaction appends one compacted record to the rollout. Its replacement_history retains context the DSL cannot represent, followed by a user message containing the compact document. On resume, Codex reconstructs history from this replacement rather than the earlier transcript.

Close Codex before appending the record.

The synthetic record follows the native records inspected during development. Their payloads contain message, normally empty, and replacement_history. Newer records also include window_number, first_window_id, previous_window_id, and window_id. Older files contain fewer window fields.

In the inspected native replacements, user and developer messages remain in order with their original content. An encrypted compaction item represents the remaining history. The disabled example below inspects the newest locally compacted thread for these fields and roles:

paths = sorted((CODEX_HOME/'sessions').rglob('*.jsonl'), reverse=True)
c = first(r for p in paths for r in load_recs(p) if r.get('type')=='compacted')
list(c.payload), [(o.get('type'),o.get('role')) for o in c.payload.replacement_history]

Synthetic replacement retains developer and system messages and earlier compaction items. The DSL cannot represent these. User prose goes into the document with the standard 2,000-token allowance.

_split_synthetic recognizes a trailing document by its conversation-body marker and extracts that body. Repeated compaction combines it with new compacted turns. Earlier native encrypted compaction items remain unchanged.

keep,body = _split_synthetic(prior)
test_eq((keep,body), (prior,''))
doc_msg = codex_msg('user', compact_content('§ Hi. §', '/tmp/t.jsonl'))
keep,body = _split_synthetic(prior+[doc_msg])
test_eq((list(keep),body), (list(prior),'§ Hi. §'))
assert _ctx_item(dict(type='message',role='developer')) and not _ctx_item(doc_msg)

source

prepare_compaction

def prepare_compaction(
    ref:NoneType=None, codex_home:pathlib.PosixPath=Path('/home/runner/.codex'),
    policy:dict={'user_toks': 2000, 'asst_toks': 150, 'call_toks': 60, 'result_toks': 35}, enc:NoneType=None,
    strip:NoneType=None
):

Prepare an incremental synthetic compaction without writing it

prepare_compaction reads the rollout without writing it. It combines the previous synthetic body with newly compacted items, following llmsurgery.ant. It protects the last five human prompts and their final assistant prose. Tool calls, results, and intermediate narration retain their budgets. Reasoning items do not appear in the DSL.

Pass strip to filter new items before compaction. llmdojo uses it to exclude practice rounds.

window_number counts compactions. previous_window_id names the window being replaced, initially the original window from session_meta. first_window_id retains that original id. A new UUID becomes window_id.

This example prepares a copy of the fixture. Its three developer messages, containing permissions and agent instructions, precede the document.

home = Path(tempfile.mkdtemp())
day = home/'sessions'/'2026'/'07'/'19'
day.mkdir(parents=True)
(day/f'rollout-2026-07-19T05-30-38-{source_tid}.jsonl').write_bytes(source_path.read_bytes())
compaction = prepare_compaction(source_tid, home)
compaction.rec['type']

'compacted'

The replacement contains three developer messages followed by one user message holding the compact document. The first record’s previous_window_id refers to the source window. There is no earlier synthetic document body:

pl = compaction.rec['payload']
test_eq([o['role'] for o in pl['replacement_history']], ['developer','developer','developer','user'])
test_eq((pl['window_number'],pl['previous_window_id']), (1,'019f78da-e13e-72a2-9c10-b003c05ac899'))
test('41.7 kilofinches', compaction.chat, in_)
test('## Conversation', compaction.content, in_)
test_eq(compaction.prior, '')

source

compact_session

def compact_session(
    ref:NoneType=None, codex_home:pathlib.PosixPath=Path('/home/runner/.codex'),
    policy:dict={'user_toks': 2000, 'asst_toks': 150, 'call_toks': 60, 'result_toks': 35}, enc:NoneType=None,
    strip:NoneType=None
):

Generate and append a synthetic thread compaction

source

append_compaction

def append_compaction(
    compaction
):

Append a prepared compaction record to its thread’s rollout

append_compaction writes the prepared record. Close Codex first, as for ant.append_compaction. That record supplies the new active history. Names remain in the thread index and need no restoration. Here the copied fixture’s document body survives extraction with _split_synthetic:

append_compaction(compaction)
prior2,new2 = split_compaction(load_rollout(source_tid, home))
test_eq((len(new2),_split_synthetic(prior2)[1]), (0,compaction.chat))

A later compaction keeps the prior document body unchanged and processes only new records. Its previous_window_id uses the last compaction’s window_id:

more = [dict(timestamp=_ts(), type='response_item', payload=p)
    for p in (codex_msg('user','Also measure the spam.'), codex_msg('assistant','Spam is at 7 decibels.'))]
with compaction.path.open('a') as f: f.writelines(json.dumps(o)+'\n' for o in more)
c2 = compact_session(source_tid, home)
test_eq((c2.prior,c2.rec['payload']['window_number']), (compaction.chat,2))
test_eq(c2.rec['payload']['previous_window_id'], pl['window_id'])
test('Spam is at 7 decibels', c2.new_chat, in_)
test_eq(split_compaction(load_rollout(source_tid, home))[1], L())

The disabled acceptance example creates a temporary app-server thread, appends synthetic compaction, and resumes it through a second app-server. A successful resume tests whether Codex accepts the record format.

with tempfile.TemporaryDirectory() as d:
    lhome,proj = Path(d),Path(d)/'project'
    proj.mkdir()
    items = reid_items([*sample,*kernel_sample], 'synthetic')
    async with codex_client(lhome) as codex: thread = await codex.create_thread(items, cwd=proj)
    compact_session(thread.id, lhome)
    async with codex_client(lhome) as codex: resumed = await codex.thread_resume(thread.id)
    test_eq(resumed.id, thread.id)
    print(show_items(active_items(load_rollout(thread.id, lhome)), mx=80))

Cleanup

The live app-server examples use temporary Codex homes. Their context managers remove those homes afterward. Checked-in fixtures remain read-only.

Headless runs with `codex exec`

We’ve been reading and editing recorded conversations. To have an agent do new work without opening the TUI, use codex exec. Pass the prompt as an argument or on stdin. -C sets its working directory and -m selects the model. The directory must be a Git repository unless you pass --skip-git-repo-check.

Override configuration with -c key=value. Values use TOML syntax. model_reasoning_effort and approval_policy are useful for these runs. --ephemeral avoids persisting a rollout. --ignore-user-config skips the user configuration file but retains authentication. It does not disable skills or rules. Saved ChatGPT login credentials can authenticate headless runs.

For automation, --json emits JSONL events on stdout. These include thread.started, turn.started, item.started, item.completed, and turn.completed. Command-execution items include the command and exit code. Other items include agent messages, file changes, and MCP calls. Completion reports token usage. Use -o path for the final message and --output-schema for a JSON Schema response. See non-interactive mode.

When you pass a prompt argument and pipe stdin, Codex appends the input in a <stdin> block. Redirect stdin from /dev/null when the run should not consume inherited input.

!codex exec --skip-git-repo-check -C /tmp/sandbox -m gpt-5.6-luna \
    -c model_reasoning_effort=low --json 'Create hello.txt containing: hi' > events.jsonl

Codex limits shell operations primarily through an OS sandbox rather than an in-sandbox command allowlist. -s read-only and -s workspace-write choose restricted modes. -s danger-full-access removes those restrictions. On macOS, the sandbox uses Seatbelt. The standard workspace-write mode permits workspace and temporary-directory writes, with networking disabled by default.

Set approval_policy=never to reject requests for user approval. This does not cancel existing allow rules. In the recorded workspace-write test, an unapproved write outside the workspace failed.

Execpolicy rules grant exceptions for command prefixes:

prefix_rule(
    pattern = ["/path/to/step1.sh"],
    decision = "allow",
    justification = "prepared command for automation",
)

The decisions are allow, prompt, and forbidden. The most restrictive matching decision wins. An allow rule permits execution outside the sandbox without a prompt.

For a plain bash -lc chain, Codex checks each command separately. The supported separators include &&, ||, ;, and |. The entire invocation requires permission for every command. Redirections, expansions, assignments, and other complex syntax prevent this split. Rules then apply to the complete shell invocation, not the inner commands. See execpolicy rules.

Earlier live tests used read-only mode with approval_policy=never. An allowed script wrote successfully through a plain bash -lc wrapper. Adding && echo x >> file.txt prevented the inner-script match. The whole invocation stayed in the sandbox and both writes failed.

For a restricted child agent, combine read-only mode, no approval requests, and narrow allow rules for prepared scripts. Audit those scripts and the complete active ruleset. The rules are exceptions to the sandbox, not restrictions on what an allowed script can do.

These setup details each cost a debugging round:

codex exec has no option that takes a rules-file path. Rules load from rules/ beside an active configuration layer. User rules apply to every run. Project rules under <repo>/.codex/rules/ require project trust. The earlier tests found that rules.prefix_rules accepts prompt and forbidden, not allow rules.
The tested -c parser split configuration keys on dots and retained quote characters. A path without dots worked in projects.<path>.trust_level=trusted. A quoted projects."/path" key did not load the intended project rules. Treat this as a version-specific CLI caveat rather than TOML syntax guidance.
Project discovery used the process working directory and a .git root marker. An empty .git directory sufficed in those tests. project_root_markers configures the markers.
Test rules offline with codex execpolicy check --rules file.rules -- cmd args.... It reports the decision and matching rules as JSON. This checks rule matching, not whether a run will load the intended configuration layer. The rules guide describes loading and offline checks.

--ignore-user-config does not skip skills and AGENTS.md. The measured runs loaded tens of thousands of input tokens, mostly cache hits. That cost depends on the installed guidance.

From Python, you can launch codex exec and parse JSONL. This notebook also uses openai_codex for app-server operations. The TypeScript @openai/codex-sdk and MCP server mode provide other integration options.

!codex exec --skip-git-repo-check -C /tmp/sandbox -s read-only -c approval_policy=never \
    -c projects./tmp/sandbox.trust_level=trusted --json \
    'Run exactly this command: /tmp/sandbox/step1.sh then review its output' > run.jsonl

Rollout layout

Reading workflow

Where sessions live

cur_thread

rollout_file

load_recs

load_rollout

project_thread

Creating dummy data

thread_id

The active history

active_items

split_compaction

response_items

Forking

Writing a session

codex_output

codex_call

codex_msg

parse_exec

codex_custom_output

codex_custom_call

exec_input

reid_items

AsyncThread.inject_items

codex_client

AsyncCodex.append_thread

AsyncCodex.create_thread

Synthetic tool calls

custom_turn

tool_turn

A sample session

Reading a session

Searching a session

item_txt

conv_items

item_role

item_search

ItemHits

show_items

Prompt history

prompt_hist

PromptHist

Curating a captured session

trunc_tools

strip_reasoning

curate_items

Session names

From dialogs

dlg2items

Back to dialogs

items2chat

thread2dlg

items2dlg

Inside a live session

Resolving session references

resolve_thread

Recognizing earlier compactions

Native compaction

Synthetic compaction

prepare_compaction

compact_session

append_compaction

Cleanup

Headless runs with codex exec

Headless runs with `codex exec`