1// Minimal codex app-server stand-in, spawned as the Codex binary by the
2// provider readiness probe tests. Answers the handshake and, for session
3// requests, returns REAL captured responses (codexMultiAgentWire.json), then
4// replays a scripted multi-agent notification sequence read from the
5// T3_CODEX_COLLAB_SCRIPT env var (a JSON file path) when the first turn
6// starts. Runs as a plain Node process — stdlib only.
7import * as NodeFS from "node:fs";
8import * as NodeReadline from "node:readline";
9import * as NodePath from "node:path";
10import * as NodeURL from "node:url";
11
12const here = NodePath.dirname(NodeURL.fileURLToPath(import.meta.url));
13const fixture = JSON.parse(
14 NodeFS.readFileSync(NodePath.join(here, "codexMultiAgentWire.json"), "utf8"),
15);
16const script = JSON.parse(NodeFS.readFileSync(process.env.T3_CODEX_COLLAB_SCRIPT, "utf8"));
17
18const write = (message) => process.stdout.write(`${JSON.stringify(message)}\n`);
19let turnStartCount = 0;
20let activeTurn;
21// Server->client requests the runtime must answer (approval prompts), keyed
22// by the numeric JSON-RPC id this peer allocated for them.
23const openServerRequests = new Map();
24const rl = NodeReadline.createInterface({ input: process.stdin });
25rl.on("line", (line) => {
26 let message;
27 try {
28 message = JSON.parse(line);
29 } catch {
30 return;
31 }
32 const { id, method } = message;
33 if (openServerRequests.has(id)) {
34 // The runtime answered an approval request. Record the response so tests
35 // can assert settlement behavior, then emit serverRequest/resolved as a
36 // deterministic receipt (tests wait on the runtime's event stream rather
37 // than polling the sidecar file). Real codex identifies the resolved
38 // request by its item id — the key the runtime correlates on — so the
39 // receipt exercises the same path that closes the approval card in the
40 // UI.
41 const request = openServerRequests.get(id);
42 openServerRequests.delete(id);
43 NodeFS.appendFileSync(
44 `${process.env.T3_CODEX_COLLAB_SCRIPT}.approvalResponses`,
45 `${JSON.stringify({ id, label: request.label, result: message.result ?? null })}\n`,
46 );
47 write({
48 jsonrpc: "2.0",
49 method: "serverRequest/resolved",
50 params: {
51 threadId: script.rootThreadId,
52 requestId: request.itemId ?? request.label,
53 },
54 });
55 return;
56 }
57 if (method === undefined && script.serverRequests?.some((request) => request.id === id)) {
58 NodeFS.appendFileSync(
59 `${process.env.T3_CODEX_COLLAB_SCRIPT}.responses`,
60 `${JSON.stringify({ id, result: message.result, error: message.error })}\n`,
61 );
62 if (script.completeTurnOnServerResponse && activeTurn) {
63 write({
64 jsonrpc: "2.0",
65 method: "turn/completed",
66 params: {
67 threadId: script.rootThreadId,
68 turn: { ...activeTurn, status: "completed" },
69 },
70 });
71 }
72 return;
73 }
74 if (method === "initialize") {
75 write({
76 id,
77 result: {
78 userAgent: "t3-collab-mock/0.0.0",
79 codexHome: "/tmp",
80 platformFamily: "unix",
81 platformOs: "linux",
82 },
83 });
84 return;
85 }
86 if (method === "account/read") {
87 write({
88 id,
89 result: { account: script.account ?? { type: "apiKey" }, requiresOpenaiAuth: false },
90 });
91 return;
92 }
93 if (method === "account/rateLimits/read" && script.failRateLimitsRead) {
94 write({ id, error: { code: -32000, message: "usage unavailable" } });
95 return;
96 }
97 if (method === "account/rateLimitResetCredit/consume" && script.resetCreditOutcome) {
98 write({ id, result: { outcome: script.resetCreditOutcome } });
99 return;
100 }
101 if (method === "skills/list" || method === "model/list") {
102 write({ id, result: { data: [] } });
103 return;
104 }
105 if (method === "thread/start") {
106 write({ id, result: fixture.responses.threadStart });
107 return;
108 }
109 if (method === "thread/inject_items" && script.recordRequests) {
110 NodeFS.appendFileSync(
111 `${process.env.T3_CODEX_COLLAB_SCRIPT}.requests`,
112 `${JSON.stringify({ method, params: message.params })}\n`,
113 );
114 write({ id, result: {} });
115 return;
116 }
117 if (method === "thread/resume") {
118 if (script.recordRequests) {
119 NodeFS.appendFileSync(
120 `${process.env.T3_CODEX_COLLAB_SCRIPT}.requests`,
121 `${JSON.stringify({ method, params: message.params })}\n`,
122 );
123 }
124 const threadId = message.params?.threadId;
125 const childSnapshot = script.childResumeSnapshots?.[threadId];
126 if (script.resumeRequestMarker) {
127 write({
128 jsonrpc: "2.0",
129 method: "serverRequest/resolved",
130 params: {
131 threadId: script.rootThreadId,
132 requestId: script.resumeRequestMarker,
133 },
134 });
135 }
136 if (childSnapshot?.hang) {
137 return;
138 }
139 if (childSnapshot?.error) {
140 write({ id, error: { code: -32000, message: childSnapshot.error } });
141 return;
142 }
143 if (childSnapshot) {
144 write({
145 id,
146 result: {
147 ...fixture.responses.threadStart,
148 model: childSnapshot.model,
149 reasoningEffort: childSnapshot.reasoningEffort,
150 thread: {
151 ...fixture.responses.threadStart.thread,
152 id: threadId,
153 sessionId: threadId,
154 },
155 },
156 });
157 for (const notification of childSnapshot.notifications ?? []) {
158 write({ jsonrpc: "2.0", method: notification.method, params: notification.params });
159 }
160 return;
161 }
162 write({ id, result: fixture.responses.threadStart });
163 return;
164 }
165 if (method === "turn/start") {
166 const turnId = script.turnIds?.[turnStartCount];
167 const turn = turnId
168 ? { ...fixture.responses.turnStart.turn, id: turnId }
169 : fixture.responses.turnStart.turn;
170 activeTurn = turn;
171 turnStartCount += 1;
172 write({ id, result: { ...fixture.responses.turnStart, turn } });
173 const rootThreadId = script.rootThreadId;
174 if (script.onlyFirstTurnStarts !== true || turnStartCount === 1) {
175 write({
176 jsonrpc: "2.0",
177 method: "turn/started",
178 params: { threadId: rootThreadId, turn },
179 });
180 }
181 for (const notification of script.notifications) {
182 write({ jsonrpc: "2.0", method: notification.method, params: notification.params });
183 }
184 for (const [index, request] of (script.serverRequests ?? []).entries()) {
185 if (request.id !== undefined) {
186 // Legacy form: caller-supplied JSON-RPC ids recorded in .responses.
187 write({ jsonrpc: "2.0", id: request.id, method: request.method, params: request.params });
188 continue;
189 }
190 // Scripted server->client requests (approval prompts). String values in
191 // params may reference "${threadId}" / "${turnId}" placeholders that are
192 // substituted with the ids this peer actually allocated.
193 const requestId = 9000 + index;
194 const label = request.label ?? `approval-${requestId}`;
195 openServerRequests.set(requestId, { label, itemId: request.params?.itemId });
196 const params = JSON.parse(
197 JSON.stringify(request.params)
198 .replaceAll("${threadId}", String(rootThreadId))
199 .replaceAll("${turnId}", String(turn.id)),
200 );
201 write({ jsonrpc: "2.0", id: requestId, method: request.method, params });
202 }
203 if (script.holdTurnOpen !== true) {
204 write({
205 jsonrpc: "2.0",
206 method: "turn/completed",
207 params: {
208 threadId: rootThreadId,
209 turn: { ...turn, status: "completed" },
210 },
211 });
212 }
213 return;
214 }
215 if (method === "turn/interrupt") {
216 // Record which thread/turn was interrupted (append-only sidecar file the
217 // test reads) so Stop coverage can assert every live child was reached.
218 // failInterruptFor simulates a dead child whose interrupt errors.
219 const target = message.params?.threadId;
220 NodeFS.appendFileSync(
221 `${process.env.T3_CODEX_COLLAB_SCRIPT}.interrupts`,
222 `${JSON.stringify({ threadId: target, turnId: message.params?.turnId })}\n`,
223 );
224 if (
225 script.expectedActiveTurnId &&
226 message.params?.threadId === script.rootThreadId &&
227 message.params?.turnId !== script.expectedActiveTurnId
228 ) {
229 write({
230 id,
231 error: {
232 code: -32000,
233 message: `expected active turn id ${message.params?.turnId} but found ${script.expectedActiveTurnId}`,
234 },
235 });
236 return;
237 }
238 if (script.failInterruptFor && script.failInterruptFor === target) {
239 write({ id, error: { code: -32000, message: "thread already closed" } });
240 return;
241 }
242 if (script.hangInterruptFor && script.hangInterruptFor === target) {
243 // Never respond: simulates a wedged child whose RPC neither resolves
244 // nor rejects. The runtime's bounded deadline must move on.
245 return;
246 }
247 write({ id, result: {} });
248 return;
249 }
250 if (id !== undefined) {
251 write({ id, result: {} });
252 }
253});