oai

Build, inspect, and fork synthetic Codex sessions

Codex keeps each conversation as a thread: a rollout file of Responses API items under CODEX_HOME/sessions, plus an app-server that can create threads, append raw items to their model-visible history, and fork them. Injected items land in the normal rollout, so a synthetic history is indistinguishable from a lived one, and codex resume <thread-id> picks it up like any other session.

This module builds the smallest useful Python interface to those operations, and converts authored dialogs into threads through the same pipeline llmsurgery.ant uses for Claude: dlg2chat recovers canonical messages with real tool calls, and fastllm’s Responses denormalizer shapes them as items. Talking to the local codex executable happens over app-server’s JSONL stdio protocol; creating and editing a session does not itself call a model.

Responses API history

Codex stores the model-visible conversation in the same item format used by the Responses API. Ordinary messages contain input_text for a user and output_text for an assistant. A tool interaction is a function_call followed by a function_call_output carrying the same call id.

source

tool_turn

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

A complete synthetic tool-use turn

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

tool_turn is enough to plant a worked example: the tool never needs to exist or to have run. The matching call ids are what tie the recorded request and result together.

sample = tool_turn('Measure the flux.', 'flux_meter', {}, 'flux: 41.7 kilofinches', 'The flux is 41.7 kilofinches.')
sample

Rollout files

Threads are not purely server-side. Codex records each one under CODEX_HOME/sessions, partitioned by date, in a JSONL rollout. The thread id is the final component of its filename.

source

load_rollout

def load_rollout(
    thread_id, codex_home:PosixPath=Path('/home/runner/.codex')
):

Read the JSONL rollout for thread_id

source

rollout_file

def rollout_file(
    thread_id, codex_home:PosixPath=Path('/home/runner/.codex')
):

The persisted rollout for thread_id, if it exists

A rollout includes configuration and event records as well as the Responses API items. load_rollout deliberately preserves all of them: they are useful when investigating what Codex actually reconstructed on resume.

Curating a captured rollout

Short examples are easiest to write synthetically. A dojo is long enough that capturing one successful run is safer, after which we can select the useful span, discard reasoning, and make its ids reproducible.

source

item_txt

def item_txt(
    item
):

Every string in a Responses API item, joined, for finding captured items by text

source

response_items

def response_items(
    recs
):

Extract the model-visible Responses API items from rollout records

Rollout files contain session metadata, turn contexts, UI events, and model history. Only response_item records are sent back as prior conversation. item_txt walks nested content and tool outputs, making the boundaries of a useful captured span easy to find.

source

curate_items

def curate_items(
    recs, key:str=''
):

Extract, strip reasoning from, and re-identify captured rollout records

source

reid_items

def reid_items(
    items, key:str=''
):

Deterministically re-derive response item and paired call ids, returning fresh dicts

source

strip_reasoning

def strip_reasoning(
    items
):

Drop Responses API reasoning items; replay does not need them

A reasoning item can contain a large summary and encrypted payload, but is not required for the worked interaction. reid_items changes both item ids and call ids, recursively updating outputs that refer to their calls. The same capture and key therefore produce identical injectable history without mutating the capture.

Talking to app-server

codex app-server speaks JSON-RPC over newline-delimited stdin and stdout. A connection starts with one initialize request and an initialized notification. Responses can be interleaved with event notifications, so the client keeps reading until it sees the id of its own request and retains everything else in events.

source

CodexAppServer

def CodexAppServer(
    cmd:str='codex', codex_home:NoneType=None
):

A small async client for codex app-server

The high-level operations correspond directly to app-server methods. start_thread creates the rollout, inject_items appends prebuilt history without starting a model turn, and fork_thread copies stored history to a new thread id. create_thread composes the first two, which is the usual path from curated items to a ready-to-resume fake dojo session.

from fastcore.test import *
import tempfile
from fastllm.chat import tool_dtls_tag
from llmsurgery.dialog import *

Before wiring curation to app-server, check it on a miniature captured rollout. Reasoning should disappear, ids should be repeatable, tool calls must remain paired with their outputs, and the source capture must remain untouched.

captured = [dict(type='response_item', payload=o) for o in [
    sample[0], dict(type='reasoning', id='rs_old', encrypted_content='secret'), *sample[1:]]]
ci = strip_reasoning(response_items(captured))
r1,r2 = reid_items(ci, 'dojo'),reid_items(ci, 'dojo')
test_eq(r1, r2)
test_eq(len(r1), 4)
test_eq(r1[2]['call_id'], r1[1]['call_id'])
test_ne(r1[1]['call_id'], ci[1]['call_id'])
test_eq(ci[1]['call_id'], sample[1]['call_id'])
test_eq([o['type'] for o in ci if 'kilofinches' in item_txt(o)], ['function_call_output', 'message'])

The end-to-end check uses a temporary Codex home. It starts a thread, plants a tool-use example, forks the result, and checks the persisted rollouts. Since no turn is started, this neither contacts a model nor spends tokens.

with tempfile.TemporaryDirectory() as d:
    home = Path(d)
    proj = home/'project'; proj.mkdir()  # chkstyle: ignore
    items = tool_turn('Measure the flux.', 'flux_meter', {}, 'flux: 41.7 kilofinches', 'The flux is 41.7 kilofinches.')

    async with CodexAppServer(codex_home=home) as app:
        thread = await app.create_thread(reid_items(items, 'live'), cwd=proj)
        fork = await app.fork_thread(thread.id)

    test_eq(fork.forkedFromId, thread.id)
    for tid in (thread.id, fork.id): assert any('41.7 kilofinches' in repr(r) for r in load_rollout(tid, home))

From dialogs

The same authored dialog that llmsurgery.ant.dlg2sess writes as a Claude session converts to a Codex thread here. dlg2items recovers the canonical messages and denormalizes them as Responses items; dlg2thread injects them into a fresh persisted thread, ready for codex resume. The thread id is minted by codex, so re-converting a dialog makes a new thread rather than overwriting (codex owns its session store; unlike Claude transcripts, we never write rollout files ourselves).

source

dlg2thread

async def dlg2thread(
    dlg, # The dialog to convert
    cwd:NoneType=None, # Project directory for the thread; the current directory if None
    codex_home:NoneType=None, # Codex home; `~/.codex` if None
    **kwargs
):

Create a persisted Codex thread whose history is dlg, returning the thread id for codex resume

source

dlg2items

def dlg2items(
    dlg, # A `Dialog`, ending with a prompt
    aim_info:NoneType=None, # Model capability dict for media handling; images enabled if None
):

Responses API items for dlg, with each reply’s tool calls recovered as real items

The flux dialog again: a reply that used a tool, plus a referenced image attachment. The tool call comes back as a paired function_call/function_call_output, and the image as an input_image:

def tool_dtl(func, args, result):
    "A tool-call details block in the reply format `fmt2hist` parses"
    d = json.dumps(dict(id='call1', server=False, call=dict(function=func, arguments=args), result=result))
    return f"{tool_dtls_tag}\n<summary><code>{func}(...)</code></summary>\n\n```json\n{d}\n```\n\n</details>"

png = tiny_png
fdlg = Dialog('flux')
fatt = Attachment(png, 'image/png')
fdlg.mk_message(f'The rig: ![](attachment:{fatt.id})', msg_type=snote, attachments=[fatt])
freply = f"Let me check.\n\n{tool_dtl('flux_meter', {'unit':'kf'}, 'flux: 41.7 kilofinches')}\n\nThe flux is 41.7 kilofinches."
fdlg.mk_message('Measure the flux please.', msg_type=sprompt, output=freply)
fitems = dlg2items(fdlg)
[i['type'] for i in fitems]
test_eq([i['type'] for i in fitems], ['message', 'message', 'function_call', 'function_call_output', 'message'])
test_eq([i.get('role') for i in fitems][:2], ['user', 'assistant'])
fc = first(i for i in fitems if i['type']=='function_call')
fo = first(i for i in fitems if i['type']=='function_call_output')
test_eq(fo['call_id'], fc['call_id'])
test_eq(fc['name'], 'flux_meter')
assert any(c['type']=='input_image' for c in fitems[0]['content'])

dlg2thread needs the codex binary (but no key and no network), so like the other app-server cells it stays out of CI. Resuming the thread and asking about the planted fact proves the conversion end to end:

tid = await dlg2thread(fdlg, proj)
!codex exec -m gpt-5.6-sol -c model_reasoning_effort=medium resume {tid} "What did the flux_meter tool report, exactly?"
tid = await dlg2thread(fdlg, tempfile.mkdtemp())
print(tid, rollout_file(tid))

Export