Skip to content

Commit 4763021

Browse files
mrcfpsmrcfps
authored
fix(daemon): parse Antigravity JSONL stdout (#4247)
* fix(daemon): parse antigravity jsonl stdout Generated-By: looper 0.9.8 (runner=worker, agent=codex) * fix(daemon): fail empty antigravity jsonl streams Generated-By: looper 0.9.8 (runner=fixer, agent=codex) * fix(daemon): require antigravity jsonl init marker Generated-By: looper 0.9.8 (runner=fixer, agent=codex) * fix(daemon): preserve antigravity jsonl first-token timing Generated-By: looper 0.9.9 (runner=fixer, agent=opencode) * fix(daemon): stamp Antigravity TTFT from assistant chunks Generated-By: looper 0.9.9 (runner=fixer, agent=opencode)
1 parent 3fc9ead commit 4763021

2 files changed

Lines changed: 300 additions & 4 deletions

File tree

‎apps/daemon/src/server.ts‎

Lines changed: 124 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -4638,6 +4638,103 @@ function resolveAcpStageTimeoutMs(): number | undefined {
46384638
return Math.min(MAX_CHAT_RUN_INACTIVITY_TIMEOUT_MS, Math.max(0, Math.floor(raw)));
46394639
}
46404640

4641+
type GeminiJsonEventStreamEvent = Record<string, unknown>;
4642+
type BufferedStdoutChunk = { text: string; receivedAt: number };
4643+
4644+
function parseGeminiJsonEventStreamEvents(text: string): GeminiJsonEventStreamEvent[] | null {
4645+
const lines = text
4646+
.split(/\r?\n/u)
4647+
.map((line) => line.trim())
4648+
.filter(Boolean);
4649+
if (lines.length === 0) return null;
4650+
const events: GeminiJsonEventStreamEvent[] = [];
4651+
for (const line of lines) {
4652+
try {
4653+
const obj = JSON.parse(line);
4654+
if (!obj || typeof obj !== 'object' || Array.isArray(obj)) return null;
4655+
events.push(obj as GeminiJsonEventStreamEvent);
4656+
} catch {
4657+
return null;
4658+
}
4659+
}
4660+
return events;
4661+
}
4662+
4663+
function isGeminiJsonEventStream(events: GeminiJsonEventStreamEvent[] | null): boolean {
4664+
if (!events || events.length === 0) return false;
4665+
const [firstEvent] = events;
4666+
if (
4667+
!firstEvent ||
4668+
firstEvent.type !== 'init' ||
4669+
typeof firstEvent.session_id !== 'string' ||
4670+
firstEvent.session_id.length === 0 ||
4671+
typeof firstEvent.model !== 'string' ||
4672+
firstEvent.model.length === 0
4673+
) {
4674+
return false;
4675+
}
4676+
return events.every((event) => {
4677+
const type = event?.type;
4678+
return (
4679+
type === 'init' ||
4680+
type === 'message' ||
4681+
type === 'tool_use' ||
4682+
type === 'tool_result' ||
4683+
type === 'error' ||
4684+
type === 'result'
4685+
);
4686+
});
4687+
}
4688+
4689+
function geminiJsonEventStreamHasVisibleAssistantText(
4690+
events: GeminiJsonEventStreamEvent[] | null,
4691+
): boolean {
4692+
if (!events) return false;
4693+
return events.some((event) => (
4694+
event.type === 'message' &&
4695+
event.role === 'assistant' &&
4696+
typeof event.content === 'string' &&
4697+
event.content.length > 0
4698+
));
4699+
}
4700+
4701+
export function bufferedAntigravityGeminiFirstTokenAt(
4702+
chunks: readonly BufferedStdoutChunk[],
4703+
): number | null {
4704+
if (chunks.length === 0) return null;
4705+
const text = chunks.map((chunk) => chunk.text).join('');
4706+
const events = parseGeminiJsonEventStreamEvents(text);
4707+
if (!isGeminiJsonEventStream(events)) return null;
4708+
if (!geminiJsonEventStreamHasVisibleAssistantText(events)) return null;
4709+
4710+
let offset = 0;
4711+
for (const line of text.split(/(\r?\n)/u)) {
4712+
const nextOffset = offset + line.length;
4713+
if (line.length > 0 && line.trim().length > 0) {
4714+
try {
4715+
const event = JSON.parse(line) as GeminiJsonEventStreamEvent;
4716+
if (
4717+
event?.type === 'message' &&
4718+
event.role === 'assistant' &&
4719+
typeof event.content === 'string' &&
4720+
event.content.length > 0
4721+
) {
4722+
let consumed = 0;
4723+
for (const chunk of chunks) {
4724+
consumed += chunk.text.length;
4725+
if (consumed >= nextOffset) return chunk.receivedAt;
4726+
}
4727+
return chunks.at(-1)?.receivedAt ?? null;
4728+
}
4729+
} catch {
4730+
return null;
4731+
}
4732+
}
4733+
offset = nextOffset;
4734+
}
4735+
return null;
4736+
}
4737+
46414738
export async function startServer({
46424739
port = 7456,
46434740
host = normalizeDaemonBindHost(process.env.OD_BIND_HOST),
@@ -13397,7 +13494,7 @@ export async function startServer({
1339713494
// time before deciding whether to forward it. The auth-prompt guard
1339813495
// in the close handler suppresses the buffer when the output is an
1339913496
// OAuth prompt; otherwise the flush below sends the chunks in order.
13400-
const plaintextStdoutBuffer: string[] = [];
13497+
const plaintextStdoutBuffer: BufferedStdoutChunk[] = [];
1340113498
// Arrival time of the first buffered plain-text stdout chunk
1340213499
// (antigravity). First-token timing is stamped from this value only
1340313500
// when the buffer is actually flushed to the client at close time. If
@@ -13412,6 +13509,9 @@ export async function startServer({
1341213509
// guard below skips them via `trackingSubstantiveOutput`.
1341313510
let agentProducedOutput = false;
1341413511
let trackingSubstantiveOutput = false;
13512+
const looksLikeGeminiJsonEventStream = (text: string) => (
13513+
isGeminiJsonEventStream(parseGeminiJsonEventStreamEvents(text))
13514+
);
1341513515
// Event types that count as "the agent actually produced something the
1341613516
// user can see." Lifecycle markers (`status`) and meter readings
1341713517
// (`usage`) deliberately do NOT count — a model can emit token-usage
@@ -13576,6 +13676,24 @@ export async function startServer({
1357613676
}
1357713677
send('agent', ev);
1357813678
};
13679+
const parseBufferedAntigravityGeminiJsonEventStream = () => {
13680+
if (
13681+
def.id !== 'antigravity' ||
13682+
plaintextStdoutBuffer.length === 0
13683+
) {
13684+
return false;
13685+
}
13686+
const bufferedStdout = plaintextStdoutBuffer.map((chunk) => chunk.text).join('');
13687+
if (!looksLikeGeminiJsonEventStream(bufferedStdout)) return false;
13688+
trackingSubstantiveOutput = true;
13689+
const firstTokenAt = bufferedAntigravityGeminiFirstTokenAt(plaintextStdoutBuffer);
13690+
if (firstTokenAt !== null) noteFirstTokenAt(firstTokenAt);
13691+
const handler = createJsonEventStreamHandler('gemini', sendAgentEvent);
13692+
handler.feed(bufferedStdout);
13693+
handler.flush();
13694+
plaintextStdoutBuffer.length = 0;
13695+
return true;
13696+
};
1357913697

1358013698
if (def.streamFormat === 'claude-stream-json') {
1358113699
const claude = createClaudeStreamHandler((ev) => {
@@ -13782,8 +13900,9 @@ export async function startServer({
1378213900
// suppressed OAuth-prompt path never reports a TTFT (PR #3412).
1378313901
child.stdout.on('data', (chunk) => {
1378413902
noteAgentActivity();
13785-
if (firstBufferedStdoutAt === null) firstBufferedStdoutAt = Date.now();
13786-
plaintextStdoutBuffer.push(String(chunk));
13903+
const receivedAt = Date.now();
13904+
if (firstBufferedStdoutAt === null) firstBufferedStdoutAt = receivedAt;
13905+
plaintextStdoutBuffer.push({ text: String(chunk), receivedAt });
1378713906
});
1378813907
} else {
1378913908
// Plain / BYOK mode: guard raw stdout chunks (#3247).
@@ -13844,6 +13963,7 @@ export async function startServer({
1384413963
markRpcCloseReason('fatal_rpc_error');
1384513964
return finishWithRetryDecision('failed', code ?? 1, signal ?? null);
1384613965
}
13966+
parseBufferedAntigravityGeminiJsonEventStream();
1384713967
if (agentStreamError) {
1384813968
markRpcCloseReason('stream_error');
1384913969
return finishWithRetryDecision('failed', code === 0 ? 1 : (code ?? 1), signal ?? null);
@@ -14161,7 +14281,7 @@ export async function startServer({
1416114281
noteFirstTokenAt(firstBufferedStdoutAt);
1416214282
}
1416314283
for (const chunk of plaintextStdoutBuffer) {
14164-
send('stdout', { chunk });
14284+
send('stdout', { chunk: chunk.text });
1416514285
}
1416614286
// Capture the pi session file path for conversational continuity.
1416714287
// The session path is discovered by attachPiRpcSession when it

‎apps/daemon/tests/chat-route.test.ts‎

Lines changed: 176 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ import { tmpdir } from 'node:os';
1818
import { delimiter, join, resolve } from 'node:path';
1919
import { afterAll, afterEach, beforeAll, describe, expect, it } from 'vitest';
2020
import {
21+
bufferedAntigravityGeminiFirstTokenAt,
2122
composeLiveInstructionPrompt,
2223
resolveGrantedCodexImagegenOverride,
2324
resolveCodexGeneratedImagesDir,
@@ -1978,6 +1979,181 @@ process.exit(0);
19781979
);
19791980
});
19801981

1982+
it('parses successful Antigravity Gemini JSONL output instead of forwarding raw stdout', async () => {
1983+
await withFakeAgent(
1984+
'agy',
1985+
`
1986+
const args = process.argv.slice(2);
1987+
if (args[0] === '--version') {
1988+
console.log('1.107.0-test');
1989+
process.exit(0);
1990+
}
1991+
process.stdout.write(JSON.stringify({ type: 'init', session_id: 'agy-1', model: 'gemini-3.5-flash' }) + '\\n');
1992+
process.stdout.write(JSON.stringify({ type: 'message', role: 'assistant', content: 'Hello from Antigravity.', delta: true }) + '\\n');
1993+
process.stdout.write(JSON.stringify({ type: 'result', status: 'success', stats: { input_tokens: 4, output_tokens: 5, cached: 0, duration_ms: 25 } }) + '\\n');
1994+
process.exit(0);
1995+
`,
1996+
async () => {
1997+
const createResponse = await fetch(`${baseUrl}/api/runs`, {
1998+
method: 'POST',
1999+
headers: { 'Content-Type': 'application/json' },
2000+
body: JSON.stringify({
2001+
agentId: 'antigravity',
2002+
message: 'hello',
2003+
}),
2004+
});
2005+
expect(createResponse.status).toBe(202);
2006+
const { runId } = await createResponse.json() as { runId: string };
2007+
2008+
const eventsController = new AbortController();
2009+
const eventsResponse = await fetch(`${baseUrl}/api/runs/${runId}/events`, {
2010+
signal: eventsController.signal,
2011+
});
2012+
const eventsBody = await readSseUntil(eventsResponse, 'event: final');
2013+
eventsController.abort();
2014+
const statusBody = await waitForRunStatus(baseUrl, runId);
2015+
2016+
expect(eventsBody).toContain('event: agent');
2017+
expect(eventsBody).toContain('"type":"text_delta","delta":"Hello from Antigravity."');
2018+
expect(eventsBody).toContain('"type":"usage"');
2019+
expect(eventsBody).not.toContain('event: stdout');
2020+
expect(eventsBody).not.toContain('"role":"assistant"');
2021+
expect(statusBody.status).toBe('succeeded');
2022+
},
2023+
);
2024+
});
2025+
2026+
it('forwards Antigravity plain stdout JSONL when it lacks the Gemini init marker', async () => {
2027+
await withFakeAgent(
2028+
'agy',
2029+
`
2030+
const args = process.argv.slice(2);
2031+
if (args[0] === '--version') {
2032+
console.log('1.107.0-test');
2033+
process.exit(0);
2034+
}
2035+
process.stdout.write(JSON.stringify({ type: 'error', message: 'requested JSONL output' }) + '\\n');
2036+
process.exit(0);
2037+
`,
2038+
async () => {
2039+
const createResponse = await fetch(`${baseUrl}/api/runs`, {
2040+
method: 'POST',
2041+
headers: { 'Content-Type': 'application/json' },
2042+
body: JSON.stringify({
2043+
agentId: 'antigravity',
2044+
message: 'return JSONL',
2045+
}),
2046+
});
2047+
expect(createResponse.status).toBe(202);
2048+
const { runId } = await createResponse.json() as { runId: string };
2049+
2050+
const eventsController = new AbortController();
2051+
const eventsResponse = await fetch(`${baseUrl}/api/runs/${runId}/events`, {
2052+
signal: eventsController.signal,
2053+
});
2054+
const eventsBody = await readSseUntil(eventsResponse, 'event: final');
2055+
eventsController.abort();
2056+
const statusBody = await waitForRunStatus(baseUrl, runId);
2057+
2058+
expect(eventsBody).toContain('event: stdout');
2059+
expect(eventsBody).toContain('requested JSONL output');
2060+
expect(eventsBody).not.toContain('event: error');
2061+
expect(statusBody.status).toBe('succeeded');
2062+
},
2063+
);
2064+
});
2065+
2066+
it('fails Antigravity Gemini JSONL output with no visible assistant content', async () => {
2067+
await withFakeAgent(
2068+
'agy',
2069+
`
2070+
const args = process.argv.slice(2);
2071+
if (args[0] === '--version') {
2072+
console.log('1.107.0-test');
2073+
process.exit(0);
2074+
}
2075+
process.stdout.write(JSON.stringify({ type: 'init', session_id: 'agy-1', model: 'gemini-3.5-flash' }) + '\\n');
2076+
process.stdout.write(JSON.stringify({ type: 'result', status: 'success', stats: { input_tokens: 4, output_tokens: 0, cached: 0, duration_ms: 25 } }) + '\\n');
2077+
process.exit(0);
2078+
`,
2079+
async () => {
2080+
const createResponse = await fetch(`${baseUrl}/api/runs`, {
2081+
method: 'POST',
2082+
headers: { 'Content-Type': 'application/json' },
2083+
body: JSON.stringify({
2084+
agentId: 'antigravity',
2085+
message: 'hello',
2086+
}),
2087+
});
2088+
expect(createResponse.status).toBe(202);
2089+
const { runId } = await createResponse.json() as { runId: string };
2090+
2091+
const eventsController = new AbortController();
2092+
const eventsResponse = await fetch(`${baseUrl}/api/runs/${runId}/events`, {
2093+
signal: eventsController.signal,
2094+
});
2095+
const eventsBody = await readSseUntil(eventsResponse, 'Agent completed without producing any output');
2096+
eventsController.abort();
2097+
const statusBody = await waitForRunStatus(baseUrl, runId);
2098+
2099+
expect(eventsBody).toContain('event: agent');
2100+
expect(eventsBody).toContain('"type":"usage"');
2101+
expect(eventsBody).toContain('event: error');
2102+
expect(eventsBody).toContain('AGENT_EXECUTION_FAILED');
2103+
expect(eventsBody).not.toContain('event: stdout');
2104+
expect(statusBody.status).toBe('failed');
2105+
},
2106+
);
2107+
});
2108+
2109+
it('preserves the first buffered stdout timestamp for Antigravity Gemini assistant text', () => {
2110+
const timestamp = bufferedAntigravityGeminiFirstTokenAt(
2111+
[{
2112+
receivedAt: 1_234,
2113+
text: [
2114+
JSON.stringify({ type: 'init', session_id: 'agy-1', model: 'gemini-3.5-flash' }),
2115+
JSON.stringify({ type: 'message', role: 'assistant', content: 'Hello from Antigravity.', delta: true }),
2116+
JSON.stringify({ type: 'result', status: 'success', stats: { input_tokens: 4, output_tokens: 5 } }),
2117+
].join('\n'),
2118+
}],
2119+
);
2120+
2121+
expect(timestamp).toBe(1_234);
2122+
});
2123+
2124+
it('stamps Antigravity Gemini assistant text from the chunk that completes the first assistant message', () => {
2125+
const timestamp = bufferedAntigravityGeminiFirstTokenAt([
2126+
{
2127+
receivedAt: 1_234,
2128+
text: `${JSON.stringify({ type: 'init', session_id: 'agy-1', model: 'gemini-3.5-flash' })}\n`,
2129+
},
2130+
{
2131+
receivedAt: 5_678,
2132+
text: `${JSON.stringify({ type: 'message', role: 'assistant', content: 'Hello from Antigravity.', delta: true })}\n`,
2133+
},
2134+
{
2135+
receivedAt: 9_999,
2136+
text: `${JSON.stringify({ type: 'result', status: 'success', stats: { input_tokens: 4, output_tokens: 5 } })}\n`,
2137+
},
2138+
]);
2139+
2140+
expect(timestamp).toBe(5_678);
2141+
});
2142+
2143+
it('does not stamp a first token timestamp for Antigravity Gemini streams without assistant text', () => {
2144+
const timestamp = bufferedAntigravityGeminiFirstTokenAt(
2145+
[{
2146+
receivedAt: 1_234,
2147+
text: [
2148+
JSON.stringify({ type: 'init', session_id: 'agy-1', model: 'gemini-3.5-flash' }),
2149+
JSON.stringify({ type: 'result', status: 'success', stats: { input_tokens: 4, output_tokens: 0 } }),
2150+
].join('\n'),
2151+
}],
2152+
);
2153+
2154+
expect(timestamp).toBeNull();
2155+
});
2156+
19812157
it('surfaces Qoder assistant error records through the SSE error channel', async () => {
19822158
const qoderErrorLine = JSON.stringify({
19832159
type: 'assistant',

0 commit comments

Comments
 (0)