1/**
2 * Usage history for Codex: the rollout format of `<home>/sessions/**.jsonl`
3 * and where each instance's shared home keeps them.
4 *
5 * @module provider/Drivers/codexUsage
6 */
7import type { CodexSettings } from "@t3tools/contracts";
8import * as Effect from "effect/Effect";
9import * as Path from "effect/Path";
10import * as Schema from "effect/Schema";
11
12import {
13 parseTimestampMs,
14 tokenCount,
15 totalTokens,
16 type ProviderUsageReader,
17 type TranscriptUsageFormat,
18 type UsageRecord,
19 type UsageSpeed,
20} from "@t3tools/provider-core/server/usage";
21import type { UsageTokenTotals } from "@t3tools/contracts";
22
23import { resolveCodexHomeLayout } from "./CodexHomeLayout.ts";
24import * as ServerConfig from "../../config.ts";
25import { sharedProviderHomePaths } from "../ManagedProviderHome.ts";
26
27/**
28 * Rolling state for a single Codex rollout file.
29 *
30 * Codex `token_count` events carry no model or service tier, so both are
31 * carried forward: the model from the most recent `turn_context`, the tier from
32 * the most recent `thread_settings_applied`. Sessions that switch either
33 * mid-run attribute correctly from the switch onward.
34 *
35 * Persisted in the scan cache at each resume point. The field order is the
36 * stored JSON's, so keep it stable.
37 */
38export const CodexScanState = Schema.Struct({
39 model: Schema.mutableKey(Schema.String),
40 speed: Schema.mutableKey(Schema.Literals(["standard", "fast", "ultrafast"])),
41 sessionId: Schema.mutableKey(Schema.String),
42 lastUsageSignature: Schema.mutableKey(Schema.NullOr(Schema.String)),
43 sawSessionMeta: Schema.mutableKey(Schema.Boolean),
44 /** While true, leading usage events are re-stamped copies of parent history. */
45 suppressingForkCopies: Schema.mutableKey(Schema.Boolean),
46 forkCopyAnchorMs: Schema.mutableKey(Schema.Finite),
47});
48export type CodexScanState = typeof CodexScanState.Type;
49
50export function initialCodexScanState(): CodexScanState {
51 return {
52 model: "",
53 speed: "standard",
54 sessionId: "",
55 lastUsageSignature: null,
56 sawSessionMeta: false,
57 suppressingForkCopies: false,
58 forkCopyAnchorMs: 0,
59 };
60}
61
62/**
63 * A forked or subagent rollout opens with the parent's full history copied in,
64 * every line re-stamped to the fork instant. Those copies are written in one
65 * synchronous burst (observed gaps 0-40ms), while the child's first genuine
66 * usage event only lands after a real model turn (observed 5s+). One second of
67 * separation splits the two cleanly; `ccusage` uses the same threshold.
68 */
69const FORK_COPY_MAX_GAP_MS = 1000;
70
71/** Whether a `session_meta` payload marks the rollout as a fork or subagent. */
72function isForkedSessionMeta(payload: Record<string, unknown>): boolean {
73 if (typeof payload["forked_from_id"] === "string") return true;
74 const source = payload["source"];
75 if (typeof source !== "object" || source === null) return false;
76 const subagent = (source as Record<string, unknown>)["subagent"];
77 if (typeof subagent !== "object" || subagent === null) return false;
78 const spawn = (subagent as Record<string, unknown>)["thread_spawn"];
79 if (typeof spawn !== "object" || spawn === null) return false;
80 return typeof (spawn as Record<string, unknown>)["parent_thread_id"] === "string";
81}
82
83/**
84 * Feeds one line of a Codex rollout into `state`, returning a record when the
85 * line was a usage event.
86 *
87 * Deltas come from `last_token_usage`. Summing those across a session
88 * reconciles with the session's final `total_token_usage`, provided
89 * consecutive duplicate events are dropped, which this does.
90 */
91export function parseCodexLine(line: string, state: CodexScanState): UsageRecord | null {
92 let parsed: unknown;
93 try {
94 parsed = JSON.parse(line);
95 } catch {
96 return null;
97 }
98 return parseCodexRecord(parsed, state);
99}
100
101function parseCodexRecord(parsed: unknown, state: CodexScanState): UsageRecord | null {
102 if (typeof parsed !== "object" || parsed === null) return null;
103
104 const record = parsed as Record<string, unknown>;
105 const payload = record["payload"];
106 if (typeof payload !== "object" || payload === null) return null;
107 const payloadRecord = payload as Record<string, unknown>;
108 const payloadType = payloadRecord["type"];
109
110 if (record["type"] === "session_meta") {
111 // Only the first meta describes this file's own session. A forked rollout
112 // repeats the ancestors' metas right after it; letting those through would
113 // reassign every subsequent record to an ancestor session.
114 if (state.sawSessionMeta) return null;
115 state.sawSessionMeta = true;
116 const id = payloadRecord["id"] ?? payloadRecord["session_id"];
117 if (typeof id === "string") state.sessionId = id;
118 const metaTimestampMs = parseTimestampMs(record["timestamp"]);
119 if (metaTimestampMs !== null && isForkedSessionMeta(payloadRecord)) {
120 state.suppressingForkCopies = true;
121 state.forkCopyAnchorMs = metaTimestampMs;
122 }
123 return null;
124 }
125
126 if (record["type"] === "turn_context") {
127 if (typeof payloadRecord["model"] === "string") state.model = payloadRecord["model"];
128 return null;
129 }
130
131 if (payloadType === "thread_settings_applied") {
132 const settings = payloadRecord["thread_settings"];
133 if (typeof settings === "object" && settings !== null) {
134 state.speed = codexSpeed((settings as Record<string, unknown>)["service_tier"]);
135 }
136 return null;
137 }
138
139 if (payloadType !== "token_count") return null;
140
141 const info = payloadRecord["info"];
142 if (typeof info !== "object" || info === null) return null;
143 const last = (info as Record<string, unknown>)["last_token_usage"];
144 if (typeof last !== "object" || last === null) return null;
145 const lastRecord = last as Record<string, unknown>;
146
147 // Only an event that is otherwise eligible may consume the duplicate
148 // signature. A token_count arriving before its turn_context (no model yet)
149 // must not poison it, or the re-emitted copy after the model is known would
150 // be skipped as a duplicate and those tokens never counted.
151 const timestampMs = parseTimestampMs(record["timestamp"]);
152 if (timestampMs === null) return null;
153 if (state.model.length === 0) return null;
154
155 // Codex re-emits an unchanged token_count on some stream boundaries. Summing
156 // those would double count, so identical consecutive payloads are skipped.
157 const signature = JSON.stringify(lastRecord);
158 if (signature === state.lastUsageSignature) return null;
159 state.lastUsageSignature = signature;
160
161 // In a forked rollout the copied parent history was already counted from the
162 // parent's own file. Drop the leading burst; the first usage event separated
163 // from its predecessor by a real turn's worth of time ends it for good.
164 if (state.suppressingForkCopies) {
165 if (timestampMs - state.forkCopyAnchorMs < FORK_COPY_MAX_GAP_MS) {
166 state.forkCopyAnchorMs = timestampMs;
167 return null;
168 }
169 state.suppressingForkCopies = false;
170 }
171
172 const inputTokens = tokenCount(lastRecord["input_tokens"]);
173 const cachedInputTokens = tokenCount(lastRecord["cached_input_tokens"]);
174 const cacheCreationTokens = tokenCount(lastRecord["cache_write_input_tokens"]);
175 const outputTokens = tokenCount(lastRecord["output_tokens"]);
176
177 const totals: UsageTokenTotals = {
178 // Codex reports `input_tokens` inclusive of the cached portion.
179 uncachedInputTokens: Math.max(0, inputTokens - cachedInputTokens - cacheCreationTokens),
180 cachedInputTokens,
181 cacheCreationTokens,
182 outputTokens,
183 // Reported inside output_tokens, surfaced separately for the token mix.
184 reasoningTokens: Math.min(outputTokens, tokenCount(lastRecord["reasoning_output_tokens"])),
185 };
186
187 if (totalTokens(totals) === 0) return null;
188
189 return {
190 provider: "codex",
191 timestampMs,
192 model: state.model,
193 sessionId: state.sessionId,
194 totals,
195 // Codex does not report cost in the rollout.
196 reportedCostUsd: null,
197 speed: state.speed,
198 // Events surviving the fork-copy suppression above are unique to this
199 // rollout, so they need no global dedup.
200 dedupeKey: null,
201 };
202}
203
204/**
205 * Maps a Codex `service_tier` to its billing speed. Codex omits the field when
206 * no tier was requested, which bills as standard, as do `default` and
207 * `standard`. `fast` is accepted as an alias of `priority`.
208 */
209function codexSpeed(serviceTier: unknown): UsageSpeed {
210 if (serviceTier === "priority" || serviceTier === "fast") return "fast";
211 if (serviceTier === "ultrafast") return "ultrafast";
212 return "standard";
213}
214
215const orEmpty = (record: UsageRecord | null): readonly UsageRecord[] =>
216 record === null ? [] : [record];
217
218export const codexUsageFormat: TranscriptUsageFormat<CodexScanState> = {
219 // Keeps the fields the reducer reads as well as the usage itself.
220 selectFields: {
221 type: true,
222 timestamp: true,
223 payload: {
224 type: true,
225 id: true,
226 session_id: true,
227 model: true,
228 thread_settings: { service_tier: true },
229 forked_from_id: true,
230 source: { subagent: { thread_spawn: { parent_thread_id: true } } },
231 info: { last_token_usage: true },
232 },
233 },
234 // `turn_context`, `thread_settings_applied` and `session_meta` lines hold no
235 // usage, but they carry the model, tier and session the reducer needs.
236 usageMarkers: ['"token_count"', '"turn_context"', '"thread_settings_applied"', '"session_meta"'],
237 parseLine: (line, state) => orEmpty(parseCodexLine(line, state)),
238 parseProjected: (projected, state) => orEmpty(parseCodexRecord(projected, state)),
239 state: { initial: initialCodexScanState, schema: CodexScanState },
240 // A rollout that moved after it was read leaves a copy in both places.
241 sharedSessionsAcrossFiles: true,
242};
243
244export const codexUsageReader = {
245 kind: "transcripts",
246 provider: "codex",
247 format: codexUsageFormat,
248 directories: Effect.fn("codexUsageReader.directories")(function* ({ config, environment }) {
249 // An undecodable config has no trustworthy home to read.
250 if (config === undefined) return [];
251 const path = yield* Path.Path;
252 if (config.sharedAccountId) {
253 const { stateDir } = yield* ServerConfig.ServerConfig;
254 const home = sharedProviderHomePaths(path, stateDir, "codex", config.sharedAccountId);
255 return [{ dir: path.join(home.shared, "sessions") }];
256 }
257 const environmentHome = environment.CODEX_HOME?.trim();
258 const layout = yield* resolveCodexHomeLayout(
259 config.setupMode !== "managed" &&
260 !config.homePath.trim() &&
261 !config.shadowHomePath.trim() &&
262 environmentHome
263 ? { ...config, homePath: environmentHome }
264 : config,
265 );
266 return [{ dir: path.resolve(layout.sharedHomePath, "sessions") }];
267 }),
268} satisfies ProviderUsageReader<CodexSettings, Path.Path | ServerConfig.ServerConfig>;