oai

Read, write, and build Codex session rollouts

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

Rollout layout

Codex stores conversations below CODEX_HOME/sessions, partitioned by date. The thread UUID appears in the rollout filename. cur_thread reads the running thread id, rollout_file finds its JSONL file without assuming a date, and project_thread finds the latest thread for a project when no live id is available. resolve_thread accepts a UUID or a thread name.

A rollout is an event log. Each line is an envelope with a timestamp, record type, and payload. Responses API input and output appear as response_item records. The history sent to the next model turn is smaller than the full event log: response_items extracts those items, and active_items applies the latest compaction replacement. load_rollout preserves every record.

Codex has no separate machine-wide prompt-history file. User messages remain in their thread rollouts, so locating the correct thread is part of prompt recovery.

Reading workflow

To find something that was said - an earlier discussion, a decision, work lost to compaction - start with thread2dlg, not with the records. It is near-instant, and turns a long event log into a dialog of a few dozen messages: d.summary() is the map (one sized row per message, a prompt’s reply on its own line), d.find_msgs(pat) the search, view_msg/view_msgs the read. doc(aidialog.dlgskill) covers that layer, and a dialog written with write_ipynb is an ordinary ipynb whose prompt sources carry their replies, so rgapi’s nbrg searches saved threads across files, replies included.

Work at the record level for surgery, or when an item’s envelope is itself the question: locate with cur_thread, project_thread, or resolve_thread; load with load_rollout; extract the current model history with active_items; search readable content with item_search, whose hits each carry their item on .item; inspect a slice with show_items. Prefer item_search to grepping JSONL, since raw rollouts contain protocol events and encoded data.

The reading functions do not modify rollouts. App-server owns ordinary thread creation, forking, naming, and native compaction. The synthetic compaction append functions write rollout JSONL directly and require Codex to be closed. Read their docs and inspect the prepared records before appending them.

Codex keeps every conversation as a thread: a rollout file under CODEX_HOME/sessions, and an app-server that owns thread creation, history injection, resume, naming, and forks. Rollouts are append-only logs rather than ready-made conversations. They include model-visible Responses API items alongside turn context, UI events, world state, and native compaction records.

A synthetic history injected through app-server resumes like lived history. That makes threads useful as editable templates and worked examples, while leaving Codex responsible for its own storage.

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

Where sessions live

Codex partitions rollouts by date below CODEX_HOME/sessions. The thread UUID is the final component of the filename. Codex passes CODEX_THREAD_ID to command processes it starts, so code in those processes can resolve its own rollout directly. A configured MCP server is long-lived and does not inherit the id for each thread.

source

cur_thread

def cur_thread():

The current Codex thread id, when running inside Codex

rollout_file resolves a UUID without assuming which date partition contains it, and takes any unique id prefix as well as a full UUID. An ambiguous prefix raises rather than picking one of the matches.

source

rollout_file

def rollout_file(
    thread_id, # Thread UUID or unique id prefix
    codex_home: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

load_recs is the raw JSONL boundary. It preserves every rollout record, including events and superseded response history.

source

load_recs

def load_recs(
    path
):

Rollout records read directly from JSONL path

load_rollout combines current-thread resolution with the raw reader.

source

load_rollout

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

The records of a persisted Codex thread

A command run outside Codex has no CODEX_THREAD_ID. project_thread supplies the matching fallback used by codexdojo -r: the newest rollout whose session metadata names the project directory. Code running in the MCP server can use the same fallback when choosing the newest project thread is safe.

source

project_thread

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

The newest (thread_id, rollout_path) for cwd

with tempfile.TemporaryDirectory() as td:
    home = Path(td)
    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')
    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))

Creating dummy data

The examples use rollouts made by the installed app-server rather than hand-written envelopes. A source thread was created in a temporary Codex home, injected with ordinary and custom tool calls, closed, resumed by another app-server, and forked. The checked-in files preserve the exact records Codex wrote without containing a real user conversation.

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

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

The active history

A rollout is an event log, not the history sent to the next model turn. Before compaction, its response_item payloads are the history. A native compacted record replaces everything before it with replacement_history; later response items extend that replacement. The latest compacted record wins.

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

Every Responses API item recorded in recs, including superseded history; each item carries its envelope timestamp

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

A small real-format compaction fixture makes the replacement rule explicit. The old flux exchange remains in the file, while the replacement and its suffix are the only active items.

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

Codex forks through app-server rather than by copying JSONL. The fork gets a fresh thread ID and normal rollout metadata, while its injected history remains model-visible.

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

Writing a session

Responses history is a flat item list. Message items carry text, and each tool interaction is a call item followed by an output with the same call_id. Codex currently records host orchestration as a custom exec call whose input is JavaScript. The useful inner operation is normally tools.exec_command or an MCP call 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 recognizes the narrow wrappers Codex uses for one nested operation: stdout forwarding for exec_command, and content forwarding for MCP results. The forwarding boilerplate has drifted across Codex versions, so the tail is compared variable- and whitespace-normalized against the known display-only variants in _MCP_TAILS; arbitrary JavaScript and the earlier stream wrapper remain ordinary raw exec calls. exec_input emits the newest wrapper for editable nested calls.

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

source

parse_exec

def parse_exec(
    src
):

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

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

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"'})

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

Captured IDs are random. reid_items derives item and call IDs from position and a salt, recursively updating outputs and metadata that refer to them.

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

The app-server owns thread files, and the openai_codex SDK is its typed client: AsyncCodex starts, resumes, reads, names, and forks threads. codex_client pins it to the PATH codex binary so the SDK and CLI share one rollout format, and redirects CODEX_HOME for tests. The one operation llmsurgery needs that lacks a first-class SDK surface is thread/inject_items; the SDK’s typed methods are all one-liners over a generic request, so a @patch adds injection — and the create/append conveniences built on it — in the SDK’s own shape.

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

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

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]

A sample session

The checked-in source rollout contains both examples. App-server created the file, then a second server resumed it and injected the custom call. No model ran.

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

load_rollout keeps every record. active_items is the reading path for model history; bookkeeping remains available for format investigation.

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

Searching a session

source

show_items

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

Readable model history for items

source

item_search

def item_search(
    pat, # Regex to find
    items, # Responses items
    maxlen:int=160, # 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

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

source

item_txt

def item_txt(
    item
):

Every readable string in a Responses item, joined

hits = item_search(r'41\.7 kilofinches', source_items)
test_eq([h.role for h in hits], ['tool','assistant'])
test_eq(hits.attrgot('role'), ['tool','assistant'])  # `ItemHits` is an `L`, so selections keep the row display and `.hist`
test_eq((type(hits[:1]).__name__, hits[:1].hist), ('ItemHits', hits.hist))
hits

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

Prompt history

Codex has no separate prompt-history JSONL. User messages remain in per-thread rollouts. prompt_hist scans them and returns the thread, project, timestamp, and text.

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

Reasoning items are large and replay does not require them. Tool inputs and outputs can also dominate a capture. The curation helpers return fresh dictionaries and keep custom calls executable after truncating their logical arguments.

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

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

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 live in Codex’s thread index rather than as ad-hoc rollout records. App-server’s thread/name/set operation updates that index. Thread IDs remain the portable resume reference.

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

From dialogs

An authored dialog converts through the same canonical history used by Claude. Normal tool calls become function_call items. A supported tool name beginning with tools. becomes a custom Codex exec item, so a dialog can show tools.mcp__clikernel__execute(code=...) while the resumed thread receives the current native JavaScript wrapper.

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 familiar flux dialog still emits a paired ordinary function call and 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'])

Changing the tool name to tools.mcp__clikernel__execute exercises the native custom mapping used by current Codex sessions.

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

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 normalizes messages and both tool-call families. A server-side web_search_call has no output item of its own, so it gets a synthetic server tool completed result: a call without one is invalid in most wire formats, and UIs show the same placeholder. Developer messages, reasoning, native compaction, ghost snapshots, and tool-search lookups are rollout context or client plumbing rather than editable turns; items2dlg preserves developer and compaction items as tagged raw messages so dlg2items can re-emit them.

source

thread2dlg

def thread2dlg(
    thread_id:NoneType=None, # Thread UUID; current thread if None
    codex_home: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

source

items2chat

def items2chat(
    items, # Responses API items
):

Canonical fastllm messages for editable conversation items

Each returned item carries its envelope timestamp, and items2chat turns it into per-message provenance meta (created), which chat2dlg stores as cell metadata. Items from a compaction’s replacement_history have no envelope, so they carry none – meta records only what the rollout actually knows:

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

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]

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

Inside a live session

CODEX_THREAD_ID is available to command processes started by Codex, but not to the long-lived MCP server. These cells inspect the running thread when the environment provides an id and otherwise use the checked-in fixture.

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

Resolving session references

Codex names are index metadata, while rollout filenames are keyed by UUID. File-level tools therefore resolve the current or explicit UUID. Name lookup belongs to app-server clients rather than filesystem heuristics.

source

resolve_thread

def resolve_thread(
    ref:NoneType=None, # Thread UUID; current thread if None
    codex_home: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 does not infer a boundary from prose. It reads the latest native compacted record, preserving its encrypted compaction item and replacement messages exactly, then returns only response items appended later.

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

Codex performs compaction itself through thread/compact/start. The result is an encrypted replacement history, which llmsurgery preserves exactly rather than trying to decode. The operation calls a model and is intentionally a live example; the next section writes the same record shape with a readable compact-DSL document instead.

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

Synthetic compaction

The compact DSL describes conversation content; this section writes it in Codex’s own record format. A synthetic compaction is one compacted record appended to the rollout while Codex is closed: its replacement_history carries the context items the DSL cannot express, ending with one user message holding the compact document. Codex rebuilds history from the latest compacted record, so a resumed thread starts from the document instead of the full transcript.

Real native records ground the mirrored shape. Alongside message (blank in practice) and replacement_history, current payloads carry window_number (how many compactions so far) and first/previous/window_id context-window UUIDs; older rollouts have fewer fields, and Codex still reads them. The pinning rule is visible in the replacement roles: every prior user and developer message is kept verbatim, in order, then one encrypted compaction item stands in for everything else. On a machine with compacted Codex threads, the newest one shows both:

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]

Where Codex pins every user and developer message, the synthetic replacement pins only what the DSL cannot render — developer and system messages, and prior compaction items — since the document itself keeps user text at high fidelity (user_toks=1000). _split_synthetic is the incremental hook: a trailing synthetic summary is recognized by its document body, extracted, and re-joined with newly-compacted turns, while a native (encrypted) prior is carried forward untouched.

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:PosixPath=Path('/home/runner/.codex'),
    policy:dict={'user_toks': 2000, 'asst_toks': 150, 'call_toks': 60, 'result_toks': 35}, enc:NoneType=None
):

Prepare an incremental synthetic compaction without writing it

prepare_compaction is read-only and mirrors llmsurgery.ant: the prior synthetic body (if any) joins the newly-compacted items, the final five messages stay untruncated, and reasoning items vanish just as native compaction drops them. The window fields follow the current native semantics — window_number counts compactions, previous_window_id is the window being replaced (the thread’s original from session_meta when this is the first), and a fresh window_id starts the next one. Preparing against a copy of the source fixture pins its three developer messages (permissions and agent instructions) ahead of 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)
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:PosixPath=Path('/home/runner/.codex'),
    policy:dict={'user_toks': 2000, 'asst_toks': 150, 'call_toks': 60, 'result_toks': 35}, enc: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

Appending happens while Codex is closed, like ant.append_compaction; the rollout is append-only, so one record establishes the new active history, and thread names live in the app-state index rather than the rollout, so nothing needs restoring. After the append, the whole active history is the replacement, and the document body round-trips through _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))

Repeated compaction is incremental, as in ant: the prior document body is preserved verbatim while only records added after the last compaction are newly compacted, and the window fields chain — previous_window_id picks up 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 acceptance check is live: create a thread through app-server, synthetically compact it offline, and resume — Codex parses the appended record and rebuilds the active history from the pinned context plus the document.

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

Temporary app-server examples use their own Codex home and disappear with the temporary directory. Checked-in fixtures are package data and remain unchanged.

Headless runs with `codex exec`

Everything above works on rollout files; codex exec is how new ones get made without a TUI: one prompt in, an agent run out. The prompt comes from argv or stdin, -C sets the working root (a git repo, unless --skip-git-repo-check), and -m picks the model. Any config.toml key can be overridden per run with -c key=value, the value parsed as TOML; model_reasoning_effort and approval_policy are the useful ones here. --ephemeral skips writing a session rollout, and --ignore-user-config skips ~/.codex/config.toml, though auth, skills, and .rules files still load. A ChatGPT subscription login works from headless runs.

For scripting, --json turns stdout into a JSONL event stream: thread.started, turn.started, item.started/item.completed (item types include agent_message, command_execution with the exact command and exit code, file_change, and mcp_tool_call), then turn.completed with token usage. -o path also writes the final agent message to a file, and --output-schema makes it conform to a JSON Schema. A piped stdin is appended to the prompt as a <stdin> block, so automation should redirect stdin from /dev/null.

!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

The permission model is the inverse of Claude Code’s. There is no per-command allowlist inside the sandbox. Instead -s read-only, -s workspace-write, or -s danger-full-access selects an OS-level sandbox (Seatbelt on macOS) bounding what any command may touch: workspace-write permits writes only under the working root and temp directories, with network off by default, and within those bounds the agent runs whatever commands it likes.

Two mechanisms control crossing that boundary. First, the model can request that a command run outside the sandbox, which normally prompts the user; -c approval_policy=never denies every such request, making the sandbox a hard boundary (verified: under workspace-write a requested write outside the workspace was refused, and the run reported it could not comply). Second, execpolicy rules, Starlark files like this:

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

Rules match argv prefixes. decision is allow, prompt, or forbidden, and the most restrictive match wins. An allow match runs the command outside the sandbox with no prompt, but only when every command in the invocation matches an allow rule: a bash -lc wrapper whose script is a plain linear chain (&&, ||, ;, |, with no redirects, expansions, or assignments) is split with tree-sitter and each piece checked separately, while anything fancier is checked as one opaque invocation that matches nothing. Verified live under -s read-only -c approval_policy=never: the allowed script’s writes succeeded even though the model wrapped it in bash -lc, and step1.sh && echo x >> file.txt ran entirely inside the read-only sandbox, where both writes were denied.

That combination is a locked-subagent recipe: read-only sandbox, escalation denied, allow rules naming exactly the scripts the child may run for real. The child reads and reasons freely, and the only state changes it can make are the prepared commands.

Getting allow rules into a run has traps, each of which cost a debugging round:

No flag takes a rules file. Allow rules load only from a rules/ folder of an active config layer: ~/.codex/rules/*.rules (applies to every run) or <repo>/.codex/rules/*.rules (loads only when the project is trusted). The config key rules.prefix_rules cannot substitute: it accepts only prompt and forbidden.
Trust can be granted per run with -c projects.<path>.trust_level=trusted, but only for paths containing no dots: the -c parser splits the key on every . and keeps quote characters literally, so the TOML-style projects."/path" spelling silently stores a wrong key and the project rules never load.
Project detection starts from the process cwd, and the default root marker is a .git directory; its existence is enough, and project_root_markers makes it configurable.
codex execpolicy check --rules file.rules cmd args... evaluates a command against a rules file offline and prints the decision as JSON, e.g. {"decision":"allow","matchedRules":[...]}. Test rules this way before relying on them, since a rules file that fails to load is a silent no-op.

Two costs to know about. Every run reloads the user’s skills and AGENTS.md, even under --ignore-user-config: tens of thousands of input tokens per call, though mostly cache hits. And there is no Python SDK; the TypeScript @openai/codex-sdk and an MCP server mode exist, but from Python you spawn the CLI per task and parse the JSONL.

!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

codex_custom_output

codex_custom_call

exec_input

parse_exec

reid_items

AsyncCodex.append_thread

AsyncCodex.create_thread

AsyncThread.inject_items

codex_client

Synthetic tool calls

custom_turn

tool_turn

A sample session

Reading a session

Searching a session

show_items

item_search

ItemHits

conv_items

item_role

item_txt

Prompt history

prompt_hist

PromptHist

Curating a captured session

curate_items

trunc_tools

strip_reasoning

Session names

From dialogs

dlg2items

Back to dialogs

thread2dlg

items2dlg

items2chat

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`