@@ -0,0 +1,881 @@
""" Codex CLI session provider — archives local agent transcripts.
Reads JSONL rollout files from ``~/.codex/sessions/YYYY/MM/DD/rollout-*.jsonl``
(override with ``CODEX_DIR``). Like Claude Code: no tokens, no rate limits, no
ToS risk — the data is local and Codex may prune it.
Local-only, deliberately
------------------------
Codex Cloud tasks (``codex cloud``) live server-side at
``https://chatgpt.com/backend-api/api/codex/tasks { ,/list}`` — the same host and
``/backend-api`` root the ChatGPT provider already speaks. **CLI sessions are
never uploaded there**, so the cloud API is not an alternative source for these
transcripts and this provider does not talk to the network. Verified 2026-08-18
against Codex 0.147.0. If ``codex cloud exec`` ever enters regular use, those
transcripts *would* be cloud-only and would need a separate provider.
Two representations, one file (measured 2026-08-18 over 7 sessions / 0.147.0)
----------------------------------------------------------------------------
Every rollout line is `` { timestamp, ordinal, type, payload}``. Dialogue appears
twice, in two different shapes, and we parse the **typed** one:
* ``response_item`` — the model-facing wire format (mirrors the OpenAI Responses
API). Tool calls arrive as *JavaScript source* because Codex ' s ``exec`` tool is
code-mode::
const r = await tools.exec_command( { " cmd " : " git status " , " workdir " : " /x " , …});
text(r.output);
Exactly one ``tools.*`` call per invocation; three functions observed:
``exec_command`` (180), ``web__run`` (15), ``apply_patch`` (13).
* ``event_msg`` / ``item_completed`` — Codex ' s own typed items, already decoded:
``UserMessage``, ``AgentMessage``, ``Reasoning``, ``CommandExecution``,
``FileChange``, ``Extension``, ``ContextCompaction``.
The typed layer wins on every axis that matters here. It is 1:1 with the raw
layer for prose (91 ``AgentMessage`` ↔ 91 assistant messages, same ids; 345
``Reasoning`` ↔ 345), it hands us structured command/exit-code/output fields
instead of JS we would have to regex, and it pre-filters harness plumbing for
free: all 51 ``developer``-role messages (skills manifests, ``<multi_agent_mode>``,
" Approved command prefix saved " ) plus the 7 ``# AGENTS.md instructions…``
injections and 1 ``<environment_context>`` have no typed item. That is the same
noise ``claude_code._HARNESS_TAG_RE`` strips by hand.
Its one weakness: it records what *ran*, not what was *attempted*. 26 of 180
``exec_command`` calls produced no ``CommandExecution`` item — 14 sandbox launch
failures (``bwrap: loopback: Failed RTM_NEWADDR``), 6 user aborts ( " aborted by
user after 504.6s " ), ~5 still running at turn end, 1 " Script failed " . So we read
the raw layer *only* to count attempts, and the collapsed placeholder reports the
shortfall ( " 3 did not complete " ) rather than silently under-reporting. ``wait``
function calls (50) are process polls, not attempts, and are not counted.
Reasoning is unrecoverable
--------------------------
All 345 reasoning items carry ``encrypted_content``; ``summary`` is ``[]`` in the
raw layer and ``summary_text``/``raw_content`` are empty in the typed layer, in
every session. Codex does not persist readable reasoning locally. Thinking is
therefore always dropped and counted — the same end state as the Claude Code
policy (decision 2026-06-12), but by necessity rather than by choice, so even
``full`` cannot surface it.
Why the sidecar SQLite is not read
----------------------------------
``~/.codex/state_5.sqlite`` carries a ``threads`` table (title, cwd, model,
tokens_used, rollout_path) and ``thread_history_1.sqlite`` a projection of the
items — but ``thread_history_projection_state`` tracks a byte offset *into the
rollout file*, i.e. the JSONL is canonical and SQLite is derived. Its ``title``
is just the first user message truncated (identical to ``first_user_message`` and
``preview``), so it offers nothing the JSONL lacks, and its filename carries a
schema version that will churn. We read the files.
Rendering follows the EXPORTER_HIDDEN_CONTENT policy: prose-only by default
(dialogue kept, tool traffic collapsed to one grouped placeholder per activity
run, reasoning dropped); ``full`` keeps the decoded tool calls and their output.
Subagents: Codex 0.147.0 ' s ``thread_spawn_edges`` table exists but is empty and no
sub-transcripts were observed, so there is no subagent folding here (contrast
``claude_code._load_subagents``). If spawned agents start appearing they will
arrive as new item types and land in the loss report as unknowns.
"""
import json
import logging
import os
import re
from collections import Counter
from datetime import datetime , timezone
from pathlib import Path
from src . blocks import (
COLLAPSED_KIND_HIDDEN_CONTEXT ,
COLLAPSED_KIND_TOOL_DUMP ,
UNKNOWN_REASON_UNKNOWN_TYPE ,
make_collapsed_block ,
make_text_block ,
make_tool_result_block ,
make_tool_use_block ,
make_unknown_block ,
)
from src . loss_report import LossReport
from src . providers . base import (
BaseProvider ,
HIDDEN_CONTENT_FULL ,
ProviderError ,
VALID_HIDDEN_CONTENT_POLICIES ,
resolve_hidden_content_policy ,
)
from src . utils import git_root_name
logger = logging . getLogger ( __name__ )
DEFAULT_SESSIONS_DIR = " ~/.codex/sessions "
# rollout-<ISO-ish timestamp>-<uuid>.jsonl — the trailing UUID is the thread id.
_ROLLOUT_RE = re . compile (
r " ^rollout- \ d {4} - \ d {2} - \ d {2} T[ \ d-]+- "
r " ([0-9a-fA-F] {8} -[0-9a-fA-F] {4} -[0-9a-fA-F] {4} -[0-9a-fA-F] {4} -[0-9a-fA-F] {12} )$ "
)
# The single tools.* call inside an `exec` custom_tool_call's JavaScript body.
_TOOLS_CALL_RE = re . compile ( r " tools \ .([A-Za-z_][ \ w.]*) \ s* \ ( " )
# Typed items that represent tool traffic, mapped to the label used in the
# collapsed placeholder. These labels must match the ones derived from the raw
# layer in _attempt_label, or the attempted-vs-completed delta is meaningless.
_TOOL_ITEM_LABELS = {
" CommandExecution " : " exec_command " ,
" FileChange " : " apply_patch " ,
# Extension is labelled by its `kind` (e.g. "web.search"); see _tool_label.
}
# function_call names that poll an already-running process rather than starting
# new work. Counting them as attempts would inflate the shortfall.
_POLLING_FUNCTIONS = { " wait " }
# How many distinct tool names to list in a collapsed-activity placeholder.
_TOOL_NAMES_SHOWN = 4
# Harness-injected user text. The typed layer already omits these (they have no
# UserMessage item), so this is a belt-and-braces guard for other Codex versions.
_HARNESS_USER_RE = re . compile (
r " ^ \ s*(?:# \ s*AGENTS \ .md instructions for \ b|<environment_context>) " ,
)
def resolve_roots ( sessions_dir = None ) - > list [ Path ] :
""" Resolve the ordered list of Codex ``sessions/`` roots to scan.
Precedence:
1. Explicit ``sessions_dir`` (a single path or a list) — used by tests.
2. ``CODEX_DIR`` env, split on ``os.pathsep`` (``:``) so multiple roots can
be given; a single path (no separator) stays backward compatible. Falls
back to the default root when unset.
3. Additionally, if ``CODEX_HOME`` is set (Codex ' s own name for its state
directory), its ``sessions/`` subdir is appended.
Roots are expanded and de-duplicated (order preserved); existence is checked
by the caller / ``_scan``.
"""
raw : list [ str ]
if sessions_dir is not None :
raw = (
[ str ( p ) for p in sessions_dir ]
if isinstance ( sessions_dir , ( list , tuple ) )
else [ str ( sessions_dir ) ]
)
else :
env = os . getenv ( " CODEX_DIR " )
raw = env . split ( os . pathsep ) if env else [ DEFAULT_SESSIONS_DIR ]
codex_home = os . getenv ( " CODEX_HOME " )
if codex_home :
raw . append ( str ( Path ( codex_home ) / " sessions " ) )
roots : list [ Path ] = [ ]
for p in raw :
p = p . strip ( )
if not p :
continue
path = Path ( p ) . expanduser ( )
if path not in roots :
roots . append ( path )
return roots
class CodexProvider ( BaseProvider ) :
""" Local-file provider over Codex CLI rollout transcripts. """
provider_name = " codex "
def __init__ (
self ,
sessions_dir : str | Path | None = None ,
hidden_content : str | None = None ,
) - > None :
super ( ) . __init__ ( )
self . _sessions_dirs = resolve_roots ( sessions_dir )
self . _hidden_content = (
hidden_content
if hidden_content in VALID_HIDDEN_CONTENT_POLICIES
else resolve_hidden_content_policy ( )
)
# conv_id → rollout file path, populated by _scan()
self . _path_map : dict [ str , Path ] = { }
# ------------------------------------------------------------------
# BaseProvider interface
# ------------------------------------------------------------------
def list_conversations ( self , offset : int = 0 , limit : int = 100 ) - > list [ dict ] :
full = self . _scan ( )
return full [ offset : offset + limit ]
def fetch_all_conversations ( self , since : datetime | None = None ) - > list [ dict ] :
convs = self . _scan ( )
if since is not None :
since_aware = since if since . tzinfo else since . replace ( tzinfo = timezone . utc )
convs = [
c for c in convs
if datetime . fromisoformat ( c [ " updated_at " ] ) > = since_aware
]
logger . info (
" [codex] Found %d session(s) under %s " ,
len ( convs ) ,
" , " . join ( str ( d ) for d in self . _sessions_dirs ) ,
)
return convs
def get_conversation ( self , conv_id : str ) - > dict :
path = self . _path_map . get ( conv_id )
if path is None :
# Direct call without a prior listing (e.g. tests) — scan first.
self . _scan ( )
path = self . _path_map . get ( conv_id )
if path is None or not path . exists ( ) :
raise ProviderError (
self . provider_name ,
f " get_conversation( { conv_id [ : 8 ] } ) " ,
FileNotFoundError ( f " No rollout file for id { conv_id } " ) ,
)
records = _parse_jsonl ( path )
mtime = datetime . fromtimestamp ( path . stat ( ) . st_mtime , tz = timezone . utc )
return {
" id " : conv_id ,
" _path " : str ( path ) ,
" _records " : records ,
# Listing and normalized updated_at must match, or the cache
# staleness comparison would re-export every session every run.
" _mtime_iso " : mtime . isoformat ( ) ,
}
def normalize_conversation ( self , raw : dict , loss_report : LossReport | None = None ) - > dict :
report = loss_report if loss_report is not None else LossReport ( )
policy = getattr ( self , " _hidden_content " , None ) or resolve_hidden_content_policy ( )
conv_id = raw . get ( " id " ) or " "
records : list [ dict ] = raw . get ( " _records " ) or [ ]
title = _extract_title ( records )
# Append the repos this session touched, e.g. "… [repo-a, repo-b]".
# Codex sessions are commonly all launched from one workspace root, so
# without this every session lands in the same notebook with no way to
# tell them apart.
launch_cwd = _extract_launch_cwd ( records )
if launch_cwd :
repos = _repos_touched ( records , launch_cwd )
if repos :
title = f " { title } [ { ' , ' . join ( repos ) } ] "
project = _extract_project ( records )
created_at = _extract_created_at ( records )
updated_at = raw . get ( " _mtime_iso " ) or next (
( r . get ( " timestamp " ) for r in reversed ( records ) if r . get ( " timestamp " ) ) , " "
)
messages = _extract_messages ( records , conv_id , report , policy )
for _ in messages :
report . record_message ( )
report . record_conversation ( )
return {
" id " : conv_id ,
" title " : title ,
" provider " : self . provider_name ,
" project " : project ,
" created_at " : created_at or " " ,
" updated_at " : updated_at or " " ,
" message_count " : len ( messages ) ,
" messages " : messages ,
}
# ------------------------------------------------------------------
# Scanning
# ------------------------------------------------------------------
def _scan ( self ) - > list [ dict ] :
existing = [ d for d in self . _sessions_dirs if d . is_dir ( ) ]
if not existing :
logger . warning (
" [codex] No sessions directory exists: %s " ,
" , " . join ( str ( d ) for d in self . _sessions_dirs ) ,
)
return [ ]
# conv_id → (mtime, conv dict). If the same thread id appears under two
# roots (e.g. a live dir and a backup), the newer-mtime copy wins.
by_id : dict [ str , tuple [ float , dict ] ] = { }
for root in existing :
# Rollouts are filed under YYYY/MM/DD; rglob keeps us agnostic to
# that layout in case Codex reorganises it.
for session_file in sorted ( root . rglob ( " rollout-*.jsonl " ) ) :
match = _ROLLOUT_RE . match ( session_file . stem )
if not match :
logger . debug ( " [codex] Skipping unrecognised filename %s " , session_file . name )
continue
try :
stat = session_file . stat ( )
except OSError :
continue
if stat . st_size == 0 :
continue
conv_id = match . group ( 1 )
prev = by_id . get ( conv_id )
if prev is not None and prev [ 0 ] > = stat . st_mtime :
logger . debug (
" [codex] Duplicate session %s in %s ; keeping newer copy " ,
conv_id [ : 8 ] , root ,
)
continue
self . _path_map [ conv_id ] = session_file
title , project , created = _read_session_meta ( session_file )
by_id [ conv_id ] = (
stat . st_mtime ,
{
" id " : conv_id ,
" title " : title ,
" project " : project ,
# The --project filter and dry-run table read the
# listing dict, not the normalized conversation.
" _project_name " : project ,
" created_at " : created ,
" updated_at " : datetime . fromtimestamp (
stat . st_mtime , tz = timezone . utc
) . isoformat ( ) ,
" _path " : str ( session_file ) ,
} ,
)
# Deterministic order: by creation date bucket, then thread id.
return sorted (
( conv for _ , conv in by_id . values ( ) ) ,
key = lambda c : ( c [ " created_at " ] , c [ " id " ] ) ,
)
# ---------------------------------------------------------------------------
# Internal helpers
# ---------------------------------------------------------------------------
def _parse_jsonl ( path : Path ) - > list [ dict ] :
""" Read a JSONL file into records, tolerating (and logging) bad lines. """
records : list [ dict ] = [ ]
bad_lines = 0
try :
text = path . read_text ( encoding = " utf-8 " )
except OSError as e :
logger . warning ( " [codex] Could not read %s : %s " , path , e )
return records
# split("\n"), not splitlines(): splitlines() also breaks on U+0085,
# U+2028/9 and friends, which are legal *inside* a JSON string. A NEL in
# captured command output shreds one record into unparseable fragments and
# loses it silently (observed 2026-08-18 in a real Codex rollout).
for line in text . split ( " \n " ) :
line = line . strip ( )
if not line :
continue
try :
records . append ( json . loads ( line ) )
except json . JSONDecodeError :
bad_lines + = 1
if bad_lines :
logger . warning ( " [codex] %s : skipped %d unparseable line(s) " , path . name , bad_lines )
return records
def _item ( rec : dict ) - > dict | None :
""" Return the typed item from an ``event_msg``/``item_completed`` record. """
if rec . get ( " type " ) != " event_msg " :
return None
payload = rec . get ( " payload " ) or { }
if payload . get ( " type " ) != " item_completed " :
return None
item = payload . get ( " item " )
return item if isinstance ( item , dict ) else None
def _item_kind ( item : dict ) - > str :
""" Typed item discriminator. 0.147.0 uses ``type``; ``item_type`` is a hedge. """
return str ( item . get ( " item_type " ) or item . get ( " type " ) or " " )
def _item_text ( item : dict ) - > str :
""" Concatenate the text of a UserMessage / AgentMessage item.
The two disagree on case — ``UserMessage`` blocks are `` " text " `` and
``AgentMessage`` blocks are `` " Text " `` — so the comparison is case-folded.
"""
content = item . get ( " content " )
if isinstance ( content , str ) :
return content . strip ( )
if not isinstance ( content , list ) :
return " "
parts = [ ]
for block in content :
if isinstance ( block , dict ) and str ( block . get ( " type " , " " ) ) . lower ( ) == " text " :
text = block . get ( " text " )
if isinstance ( text , str ) :
parts . append ( text )
return " " . join ( parts ) . strip ( )
def _read_session_meta ( path : Path ) - > tuple [ str , str | None , str ] :
""" Light single-pass scan for listing metadata: (title, project, created_at).
Substring guards keep this cheap — only candidate lines are JSON-parsed, and
the scan stops as soon as the title is found (the first UserMessage is
usually within the first few dozen lines of a multi-megabyte file).
"""
title = " "
project : str | None = None
created = " "
try :
with path . open ( encoding = " utf-8 " ) as fh :
for line in fh :
if not created and ' " session_meta " ' in line :
try :
rec = json . loads ( line )
except json . JSONDecodeError :
continue
payload = rec . get ( " payload " ) or { }
created = str ( payload . get ( " timestamp " ) or rec . get ( " timestamp " ) or " " )
cwd = payload . get ( " cwd " )
if cwd :
project = Path ( cwd ) . name or None
continue
if not title and ' " UserMessage " ' in line :
try :
rec = json . loads ( line )
except json . JSONDecodeError :
continue
item = _item ( rec )
if item is None or _item_kind ( item ) != " UserMessage " :
continue
text = _item_text ( item )
if text and not _HARNESS_USER_RE . match ( text ) :
title = text [ : 80 ]
break
except OSError as e :
logger . warning ( " [codex] Could not read %s : %s " , path , e )
return title or " Untitled session " , project , created
def _extract_title ( records : list [ dict ] ) - > str :
""" Codex has no AI-generated title — the first real user prompt is the title.
(``state_5.sqlite`` ' s ``title`` column is this same string truncated; see the
module docstring on why the sidecar DB is not consulted.)
"""
for rec in records :
item = _item ( rec )
if item is None or _item_kind ( item ) != " UserMessage " :
continue
text = _item_text ( item )
if text and not _HARNESS_USER_RE . match ( text ) :
return text [ : 80 ]
return " Untitled session "
def _session_meta_payload ( records : list [ dict ] ) - > dict :
for rec in records :
if rec . get ( " type " ) == " session_meta " :
payload = rec . get ( " payload " )
if isinstance ( payload , dict ) :
return payload
return { }
def _extract_launch_cwd ( records : list [ dict ] ) - > str | None :
""" The session ' s working directory (constant per session). """
cwd = _session_meta_payload ( records ) . get ( " cwd " )
return cwd if isinstance ( cwd , str ) and cwd else None
def _extract_project ( records : list [ dict ] ) - > str | None :
""" Project = basename of the session ' s working directory. """
cwd = _extract_launch_cwd ( records )
if cwd :
return Path ( cwd ) . name or None
return None
def _extract_created_at ( records : list [ dict ] ) - > str :
""" Session start time.
Prefers ``session_meta.payload.timestamp`` — the outer line ``timestamp`` is
when the record was *flushed*, which can trail the true start by minutes.
"""
payload = _session_meta_payload ( records )
ts = payload . get ( " timestamp " )
if isinstance ( ts , str ) and ts :
return ts
return next ( ( r . get ( " timestamp " ) for r in records if r . get ( " timestamp " ) ) , " " ) or " "
def _ignored_repos ( ) - > set [ str ] :
""" Optional ignore-list: git repos to never tag (comma-separated names). """
env = os . getenv ( " CODEX_REPO_TAG_IGNORE " , " " )
return { s . strip ( ) for s in env . split ( " , " ) if s . strip ( ) }
def _strip_file_uri ( path : str ) - > str :
""" ``CommandExecution.cwd`` is a ``file://`` URI; every other path is plain. """
return path [ 7 : ] if path . startswith ( " file:// " ) else path
# Absolute path-like tokens inside command strings.
_ABS_PATH_RE = re . compile ( r " /(?:[ \ w. \ -]+/)*[ \ w. \ -]+ " )
def _repos_touched (
records : list [ dict ] , launch_cwd : str , cap : int = 3 , ignore : set [ str ] | None = None
) - > list [ str ] :
""" Repos a session touched, for the title ' s ``[repo-a, repo-b]`` tag.
A file ' s repo is the git repository it lives in (nearest ancestor with a
``.git``). Paths come from the typed items: ``FileChange.changes`` keys
(absolute), and ``CommandExecution`` ' s argv plus its ``cwd``. Ordered by
touch frequency, capped with a trailing ``…``.
"""
ignore = _ignored_repos ( ) if ignore is None else ignore
counts : Counter = Counter ( )
seen : set [ str ] = set ( )
def note ( p ) - > None :
if not isinstance ( p , str ) or not p :
return
p = _strip_file_uri ( p )
if not p . startswith ( " / " ) : # resolve relative paths against the launch cwd
if not launch_cwd :
return
p = str ( Path ( launch_cwd ) / p )
if p in seen :
return
seen . add ( p )
name = git_root_name ( Path ( p ) )
if name and not name . startswith ( " . " ) and name not in ignore :
counts [ name ] + = 1
for rec in records :
item = _item ( rec )
if item is None :
continue
kind = _item_kind ( item )
if kind == " FileChange " :
changes = item . get ( " changes " )
if isinstance ( changes , dict ) :
for file_path in changes :
note ( file_path )
elif kind == " CommandExecution " :
note ( item . get ( " cwd " ) )
command = item . get ( " command " )
if isinstance ( command , list ) and command :
tail = command [ - 1 ]
if isinstance ( tail , str ) :
for m in _ABS_PATH_RE . finditer ( tail ) :
note ( m . group ( 0 ) )
ordered = [ name for name , _ in counts . most_common ( ) ]
if len ( ordered ) > cap :
return ordered [ : cap ] + [ " … " ]
return ordered
def _tool_label ( item : dict , kind : str ) - > str :
""" Placeholder label for a completed tool item.
``Extension`` covers everything routed through the model ' s own extensions
(``web.search`` so far), so its ``kind`` field is the useful name.
"""
if kind == " Extension " :
return str ( item . get ( " kind " ) or " extension " )
return _TOOL_ITEM_LABELS . get ( kind , kind )
def _attempt_label ( payload : dict ) - > str | None :
""" Label for a raw tool call, or None if it should not count as an attempt.
``exec`` custom_tool_calls carry JavaScript; the inner ``tools.<fn>`` name is
what lines up with the typed items ' labels. ``wait`` polls an already-running
process and starts no new work.
"""
ptype = payload . get ( " type " )
name = payload . get ( " name " ) or " "
if ptype == " custom_tool_call " :
if name == " exec " :
match = _TOOLS_CALL_RE . search ( payload . get ( " input " ) or " " )
if match :
# tools.web__run → the Extension item calls itself "web.search";
# both are "one web call", which is all the count claims.
return match . group ( 1 )
return " exec "
return name or " tool "
if ptype == " function_call " :
if name in _POLLING_FUNCTIONS :
return None
return name or " function "
return None
def _command_string ( item : dict ) - > str :
""" Human-readable command from a ``CommandExecution`` argv list.
The argv is ``[ " /bin/bash " , " -lc " , " <script> " ]``; the script is the part
worth showing.
"""
command = item . get ( " command " )
if isinstance ( command , list ) :
if len ( command ) > = 3 and command [ 0 ] . endswith ( " sh " ) and command [ 1 ] in ( " -lc " , " -c " ) :
return str ( command [ - 1 ] )
return " " . join ( str ( c ) for c in command )
return str ( command or " " )
def _extract_messages (
records : list [ dict ] ,
conv_id : str ,
report : LossReport ,
policy : str ,
) - > list [ dict ] :
""" Normalize Codex rollout records into messages.
Reads the typed ``item_completed`` layer for content and the raw
``response_item`` layer only to count attempted tool calls, so a collapsed
placeholder can report calls that never produced an item (sandbox failures,
user aborts, still-running processes). See the module docstring.
"""
messages : list [ dict ] = [ ]
# Pending collapsed tool activity for the current run.
pending_tools : Counter = Counter ( )
pending_bytes = 0
pending_attempted = 0
pending_completed = 0
def flush_pending ( ) - > None :
nonlocal pending_bytes , pending_attempted , pending_completed
# An activity run with only failed attempts still deserves a placeholder
# — silence would imply nothing happened.
shortfall = max ( pending_attempted - pending_completed , 0 )
if not pending_tools and not shortfall :
return
if policy == HIDDEN_CONTENT_FULL :
# Under `full` the completed calls are already rendered as tool
# blocks, so only the shortfall is left to report — and a *collapsed*
# block would render the "omitted (set full to keep)" suffix, which
# contradicts the policy in force. Say it plainly instead.
report . record_collapsed ( " tool_call_incomplete " , 0 )
note = make_tool_result_block (
f " { shortfall } tool call(s) produced no result — the process failed to "
" launch, was aborted, or was still running when the turn ended. " ,
tool_name = " incomplete " ,
is_error = True ,
)
if messages and messages [ - 1 ] [ " role " ] == " assistant " :
messages [ - 1 ] [ " blocks " ] . append ( note )
else :
messages . append (
{ " role " : " tool " , " content_type " : " text " , " timestamp " : None ,
" blocks " : [ note ] }
)
pending_tools . clear ( )
pending_bytes = 0
pending_attempted = 0
pending_completed = 0
return
if pending_tools :
shown = " , " . join (
f " { name } × { count } "
for name , count in pending_tools . most_common ( _TOOL_NAMES_SHOWN )
)
if len ( pending_tools ) > _TOOL_NAMES_SHOWN :
shown + = " , … "
origin = f " { sum ( pending_tools . values ( ) ) } calls: { shown } "
else :
origin = " 0 calls "
if shortfall :
origin + = f " (+ { shortfall } did not complete) "
report . record_collapsed ( " tool_call_incomplete " , 0 )
block = make_collapsed_block (
origin = origin ,
content_type = " tool_activity " ,
size_bytes = pending_bytes ,
kind = COLLAPSED_KIND_TOOL_DUMP ,
)
if messages and messages [ - 1 ] [ " role " ] == " assistant " :
messages [ - 1 ] [ " blocks " ] . append ( block )
else :
messages . append (
{ " role " : " tool " , " content_type " : " text " , " timestamp " : None , " blocks " : [ block ] }
)
pending_tools . clear ( )
pending_bytes = 0
pending_attempted = 0
pending_completed = 0
def append_message ( role : str , blocks : list [ dict ] , timestamp ) - > None :
# Attach the preceding run's tool activity to the previous message
# before starting a new one, so the placeholder lands between the
# dialogue turns it actually occurred between.
flush_pending ( )
messages . append (
{
" role " : role ,
" content_type " : " text " ,
" timestamp " : timestamp ,
" blocks " : blocks ,
}
)
for rec in records :
rec_type = rec . get ( " type " )
payload = rec . get ( " payload " ) or { }
timestamp = rec . get ( " timestamp " )
# ── Raw layer: attempt counting only ──────────────────────────────
if rec_type == " response_item " :
if payload . get ( " type " ) in ( " custom_tool_call " , " function_call " ) :
if _attempt_label ( payload ) is not None :
pending_attempted + = 1
continue
item = _item ( rec )
if item is None :
continue
kind = _item_kind ( item )
# ── Dialogue ──────────────────────────────────────────────────────
if kind in ( " UserMessage " , " AgentMessage " ) :
text = _item_text ( item )
if kind == " UserMessage " and _HARNESS_USER_RE . match ( text ) :
# Defensive: 0.147.0 gives these no typed item at all.
report . record_filtered_role ( " codex.harness_injection " )
continue
block = make_text_block ( text )
if block :
append_message (
" user " if kind == " UserMessage " else " assistant " , [ block ] , timestamp
)
continue
# ── Reasoning: encrypted at rest, nothing to keep under any policy ─
if kind == " Reasoning " :
report . record_collapsed ( " reasoning " , len ( json . dumps ( item , default = str ) ) )
continue
# ── Context compaction: a visible marker, not a silent drop ────────
if kind == " ContextCompaction " :
flush_pending ( )
marker = make_collapsed_block (
origin = " context compacted — earlier turns dropped from the model ' s context " ,
content_type = " context_compaction " ,
size_bytes = 0 ,
kind = COLLAPSED_KIND_HIDDEN_CONTEXT ,
)
report . record_collapsed ( " context_compaction " , 0 )
messages . append (
{ " role " : " tool " , " content_type " : " text " , " timestamp " : timestamp ,
" blocks " : [ marker ] }
)
continue
# ── Tool traffic ──────────────────────────────────────────────────
if kind in ( " CommandExecution " , " FileChange " , " Extension " ) :
pending_completed + = 1
label = _tool_label ( item , kind )
size = len ( json . dumps ( item , default = str ) )
if policy == HIDDEN_CONTENT_FULL :
use_block , result_block = _full_tool_blocks ( item , kind , label )
target = messages [ - 1 ] if messages and messages [ - 1 ] [ " role " ] == " assistant " else None
if target is None :
messages . append (
{ " role " : " tool " , " content_type " : " text " ,
" timestamp " : timestamp , " blocks " : [ ] }
)
target = messages [ - 1 ]
target [ " blocks " ] . append ( use_block )
if result_block :
target [ " blocks " ] . append ( result_block )
else :
pending_tools [ label ] + = 1
pending_bytes + = size
report . record_collapsed ( label , size )
continue
# ── Anything new in a future Codex version ────────────────────────
logger . warning ( " [codex] Unknown item type %r in session %s " , kind , conv_id [ : 8 ] )
report . record_unknown ( f " codex. { kind or ' ? ' } " )
append_message (
" tool " ,
[
make_unknown_block (
raw_type = f " codex. { kind or ' ? ' } " ,
observed_keys = list ( item . keys ( ) ) ,
reason = UNKNOWN_REASON_UNKNOWN_TYPE ,
)
] ,
timestamp ,
)
flush_pending ( )
return messages
def _full_tool_blocks ( item : dict , kind : str , label : str ) - > tuple [ dict , dict | None ] :
""" Decoded tool_use / tool_result blocks for EXPORTER_HIDDEN_CONTENT=full.
The typed item already carries structured fields, so this reads the command,
exit code and output directly rather than parsing the raw layer ' s JavaScript.
"""
if kind == " CommandExecution " :
exit_code = item . get ( " exit_code " )
use = make_tool_use_block (
label ,
{
" command " : _command_string ( item ) ,
" cwd " : _strip_file_uri ( str ( item . get ( " cwd " ) or " " ) ) ,
" status " : item . get ( " status " ) ,
" exit_code " : exit_code ,
} ,
item . get ( " id " ) ,
)
output = item . get ( " aggregated_output " ) or item . get ( " formatted_output " ) or " "
result = make_tool_result_block (
str ( output ) ,
tool_name = label ,
is_error = bool ( exit_code not in ( 0 , None ) ) ,
)
return use , result
if kind == " FileChange " :
changes = item . get ( " changes " ) if isinstance ( item . get ( " changes " ) , dict ) else { }
use = make_tool_use_block (
label ,
{
" files " : {
path : ( change . get ( " type " ) if isinstance ( change , dict ) else " ? " )
for path , change in changes . items ( )
} ,
" status " : item . get ( " status " ) ,
} ,
item . get ( " id " ) ,
)
output = " \n " . join (
part for part in ( item . get ( " stdout " ) , item . get ( " stderr " ) ) if part
)
result = make_tool_result_block (
output or f " { len ( changes ) } file(s) changed " ,
tool_name = label ,
is_error = item . get ( " status " ) not in ( None , " completed " , " success " ) ,
)
return use , result
# Extension (web.search and anything else routed through an extension).
use = make_tool_use_block (
label ,
{ " query " : item . get ( " query " ) , " action " : item . get ( " action " ) } ,
item . get ( " id " ) ,
)
results = item . get ( " results " )
result = make_tool_result_block (
json . dumps ( results , default = str , indent = 1 ) if results else " " ,
tool_name = label ,
)
return use , result
JesseMarkowitz