From c92fb003051c1f3182b744d00df4e677bb473d21 Mon Sep 17 00:00:00 2001 From: Timo Date: Mon, 31 Aug 2026 21:02:51 +0300 Subject: [PATCH 1/5] fix(agy): stream structured print-mode events --- .../src/agent/acp/AcpSessionManager.test.ts | 8 +++ .../src/agent/acp/AcpSessionManager.ts | 2 +- .../happy-cli/src/agent/core/AgentBackend.ts | 2 +- packages/happy-cli/src/agy/AgyBackend.test.ts | 28 ++++++-- packages/happy-cli/src/agy/AgyBackend.ts | 69 +++++++++++++++++-- packages/happy-cli/src/agy/cliArgs.test.ts | 1 + packages/happy-cli/src/agy/cliArgs.ts | 2 +- 7 files changed, 99 insertions(+), 13 deletions(-) diff --git a/packages/happy-cli/src/agent/acp/AcpSessionManager.test.ts b/packages/happy-cli/src/agent/acp/AcpSessionManager.test.ts index d8abfd635d..265c43ecf3 100644 --- a/packages/happy-cli/src/agent/acp/AcpSessionManager.test.ts +++ b/packages/happy-cli/src/agent/acp/AcpSessionManager.test.ts @@ -139,6 +139,14 @@ describe('AcpSessionManager text mapping', () => { expect(envelopes[1].ev).toEqual({ t: 'turn-end', status: 'completed' }); }); + it('flushes model output marked for immediate delivery', () => { + const mapper = new AcpSessionManager(); + mapper.startTurn(); + const envelopes = mapper.mapMessage({ type: 'model-output', textDelta: 'live', flush: true }); + expect(envelopes).toHaveLength(1); + expect(envelopes[0].ev).toMatchObject({ t: 'text', text: 'live' }); + }); + it('flushes accumulated output when thinking starts', () => { const mapper = new AcpSessionManager(); mapper.startTurn(); diff --git a/packages/happy-cli/src/agent/acp/AcpSessionManager.ts b/packages/happy-cli/src/agent/acp/AcpSessionManager.ts index 9a1c12e70a..f0b5566838 100644 --- a/packages/happy-cli/src/agent/acp/AcpSessionManager.ts +++ b/packages/happy-cli/src/agent/acp/AcpSessionManager.ts @@ -139,7 +139,7 @@ export class AcpSessionManager { const flushed = this.pendingType !== 'output' ? this.flush() : []; this.pendingType = 'output'; this.pendingText += text; - return flushed; + return msg.flush ? [...flushed, ...this.flush()] : flushed; } if (msg.type === 'tool-call') { diff --git a/packages/happy-cli/src/agent/core/AgentBackend.ts b/packages/happy-cli/src/agent/core/AgentBackend.ts index cec9b73f44..e128a10288 100644 --- a/packages/happy-cli/src/agent/core/AgentBackend.ts +++ b/packages/happy-cli/src/agent/core/AgentBackend.ts @@ -23,7 +23,7 @@ export type ToolCallId = string; * These messages are forwarded to the Happy server and mobile app. */ export type AgentMessage = - | { type: 'model-output'; textDelta?: string; fullText?: string } + | { type: 'model-output'; textDelta?: string; fullText?: string; flush?: boolean } | { type: 'status'; status: 'starting' | 'running' | 'idle' | 'stopped' | 'error'; detail?: string } | { type: 'tool-call'; toolName: string; args: Record; callId: ToolCallId } | { type: 'tool-result'; toolName: string; result: unknown; callId: ToolCallId } diff --git a/packages/happy-cli/src/agy/AgyBackend.test.ts b/packages/happy-cli/src/agy/AgyBackend.test.ts index e545b9621d..f6baf39684 100644 --- a/packages/happy-cli/src/agy/AgyBackend.test.ts +++ b/packages/happy-cli/src/agy/AgyBackend.test.ts @@ -39,9 +39,9 @@ describe('AgyBackend', () => { await backend.startSession(); const turn = backend.sendPrompt('/work', 'hi'); - // Stream two chunks then exit cleanly. - stdout.emit('data', 'Hello '); - stdout.emit('data', 'world'); + // Split NDJSON records across arbitrary stdout chunks. + stdout.emit('data', '{"event":"init","conversation_id":"c1"}\n{"event":"step_update","step_update":{"step_type":"agent_response","text_'); + stdout.emit('data', 'delta":"Hello "}}\n{"event":"step_update","step_update":{"step_type":"agent_response","text_delta":"world"}}\n'); child.emit('close', 0); await expect(turn).resolves.toBeUndefined(); @@ -55,12 +55,30 @@ describe('AgyBackend', () => { expect(spawnOpts.stdio).toEqual(['ignore', 'pipe', 'pipe']); expect(messages.filter((m) => m.type === 'model-output')).toEqual([ - { type: 'model-output', textDelta: 'Hello ' }, - { type: 'model-output', textDelta: 'world' }, + { type: 'model-output', textDelta: 'Hello ', flush: true }, + { type: 'model-output', textDelta: 'world', flush: true }, ]); expect(messages.at(-1)).toMatchObject({ type: 'status', status: 'idle' }); }); + it('maps tool lifecycle records with paired ids and redacted args', async () => { + const { child, stdout } = makeFakeChild(); + const spawnFn = vi.fn(() => child) as unknown as SpawnFn; + const backend = new AgyBackend({ cwd: '/work', permissionMode: 'default', spawnFn, resolveConversationId: () => null }); + const messages: AgentMessage[] = []; + backend.onMessage((m) => messages.push(m)); + + const turn = backend.sendPrompt('/work', 'hi'); + stdout.emit('data', '{"event":"init","conversation_id":"c1"}\n{"event":"step_update","step_update":{"step_index":2,"state":"ACTIVE","step_type":"tool","tool_name":"ReadFile","tool_info":{"parameters":{"path":"a.txt","token":"nope"}}}}\n{"event":"step_update","step_update":{"step_index":2,"state":"DONE","step_type":"tool","tool_name":"ReadFile"}}\n'); + child.emit('close', 0); + await turn; + + expect(messages.filter((m) => m.type === 'tool-call' || m.type === 'tool-result')).toEqual([ + { type: 'tool-call', toolName: 'ReadFile', callId: 'c1:2', args: { path: 'a.txt' } }, + { type: 'tool-result', toolName: 'ReadFile', callId: 'c1:2', result: { status: 'DONE' } }, + ]); + }); + it('emits an error status and rejects on non-zero exit', async () => { const { child } = makeFakeChild(); const spawnFn = vi.fn(() => child) as unknown as SpawnFn; diff --git a/packages/happy-cli/src/agy/AgyBackend.ts b/packages/happy-cli/src/agy/AgyBackend.ts index 305d468ab3..94331bc899 100644 --- a/packages/happy-cli/src/agy/AgyBackend.ts +++ b/packages/happy-cli/src/agy/AgyBackend.ts @@ -48,6 +48,29 @@ export interface AgyBackendOptions { resolveConversationId?: (cwd: string) => string | null; } +type AgyStepUpdate = { + conversation_id?: unknown; + step_index?: unknown; + state?: unknown; + step_type?: unknown; + text_delta?: unknown; + tool_name?: unknown; + tool_info?: { parameters?: unknown }; +}; + +function safeToolArgs(value: unknown, depth = 0): Record { + if (!value || typeof value !== 'object' || Array.isArray(value) || depth > 2) return {}; + return Object.fromEntries(Object.entries(value as Record) + .filter(([key]) => !/(token|secret|password|authorization|cookie|api[_-]?key)/i.test(key)) + .slice(0, 12) + .map(([key, item]) => { + if (typeof item === 'string') return [key, item.slice(0, 512)]; + if (typeof item === 'number' || typeof item === 'boolean' || item === null) return [key, item]; + if (Array.isArray(item)) return [key, item.slice(0, 8).map((entry) => typeof entry === 'string' ? entry.slice(0, 256) : String(entry).slice(0, 256))]; + return [key, safeToolArgs(item, depth + 1)]; + })); +} + /** Parse an agy duration string ("10m", "30s", "1h") to milliseconds; defaults to 10m. */ function parsePrintTimeoutMs(value: string): number { const m = /^(\d+)\s*(s|m|h)$/.exec(value.trim()); @@ -127,6 +150,40 @@ export class AgyBackend implements AgentBackend { stdio: ['ignore', 'pipe', 'pipe'], }); this.child = child; + let streamBuffer = ''; + let streamConversationId: string | null = null; + let streamFailure: string | null = null; + const handleRecord = (line: string) => { + let record: { event?: unknown; conversation_id?: unknown; step_update?: AgyStepUpdate; result?: { status?: unknown } }; + try { + record = JSON.parse(line) as typeof record; + } catch { + this.log(`ignored malformed agy stream-json record: ${line.slice(0, 256)}`); + return; + } + if (record.event === 'init' && typeof record.conversation_id === 'string') { + streamConversationId = record.conversation_id; + return; + } + if (record.event === 'result' && record.result?.status !== 'SUCCESS') { + streamFailure = `agy stream result status ${String(record.result?.status)}`; + return; + } + if (record.event !== 'step_update' || !record.step_update) return; + const update = record.step_update; + if (typeof update.conversation_id === 'string') streamConversationId = update.conversation_id; + if (update.step_type === 'agent_response' && typeof update.text_delta === 'string' && update.text_delta) { + this.emit({ type: 'model-output', textDelta: update.text_delta, flush: true }); + return; + } + if (update.step_type !== 'tool' || typeof update.step_index !== 'number' || typeof update.tool_name !== 'string') return; + const callId = `${streamConversationId ?? 'agy'}:${update.step_index}`; + if (update.state === 'ACTIVE') { + this.emit({ type: 'tool-call', toolName: update.tool_name, callId, args: safeToolArgs(update.tool_info?.parameters) }); + } else if (update.state === 'DONE' || update.state === 'ERROR') { + this.emit({ type: 'tool-result', toolName: update.tool_name, callId, result: { status: update.state } }); + } + }; // Node can fire both 'error' and 'close' on spawn failure; act on the first only. let settled = false; @@ -143,9 +200,10 @@ export class AgyBackend implements AgentBackend { child.stdout?.setEncoding('utf8'); child.stdout?.on('data', (chunk: string) => { - if (chunk) { - this.emit({ type: 'model-output', textDelta: chunk }); - } + streamBuffer += chunk; + const records = streamBuffer.split('\n'); + streamBuffer = records.pop() ?? ''; + for (const record of records) if (record.trim()) handleRecord(record); }); child.stderr?.setEncoding('utf8'); @@ -171,6 +229,7 @@ export class AgyBackend implements AgentBackend { if (settled) return; settled = true; cleanup(); + if (streamBuffer.trim()) this.log(`ignored unterminated agy stream-json record: ${streamBuffer.slice(0, 256)}`); // Pin the conversation our first turn created so later turns resume it. // Once pinned, never re-read the cache: another session in the same cwd // may have updated it since, and adopting that id would cross-resume. @@ -190,11 +249,11 @@ export class AgyBackend implements AgentBackend { } } - if (code === 0) { + if (code === 0 && !streamFailure) { this.emit({ type: 'status', status: 'idle' }); resolve(); } else { - const detail = `agy exited with code ${code ?? 'null'}`; + const detail = streamFailure ?? `agy exited with code ${code ?? 'null'}`; this.emit({ type: 'status', status: 'error', detail }); reject(new Error(detail)); } diff --git a/packages/happy-cli/src/agy/cliArgs.test.ts b/packages/happy-cli/src/agy/cliArgs.test.ts index b1b356eb2a..d2474b787a 100644 --- a/packages/happy-cli/src/agy/cliArgs.test.ts +++ b/packages/happy-cli/src/agy/cliArgs.test.ts @@ -8,6 +8,7 @@ describe('buildAgyArgs', () => { expect(args).toContain('--sandbox'); expect(args).not.toContain('--dangerously-skip-permissions'); + expect(args.slice(-4)).toEqual(['--output-format', 'stream-json', '--print', 'hello world']); expect(args.slice(-2)).toEqual(['--print', 'hello world']); }); diff --git a/packages/happy-cli/src/agy/cliArgs.ts b/packages/happy-cli/src/agy/cliArgs.ts index 07d19c25da..5ae40e3020 100644 --- a/packages/happy-cli/src/agy/cliArgs.ts +++ b/packages/happy-cli/src/agy/cliArgs.ts @@ -61,6 +61,6 @@ export function buildAgyArgs(opts: BuildAgyArgsOptions): string[] { args.push('--print-timeout', opts.printTimeout); } - args.push('--print', opts.prompt); + args.push('--output-format', 'stream-json', '--print', opts.prompt); return args; } From 201eb662b072b7685205928a10a659ad084e6e7c Mon Sep 17 00:00:00 2001 From: Timo Date: Mon, 31 Aug 2026 21:12:24 +0300 Subject: [PATCH 2/5] fix(agy): avoid exposing stream payloads --- packages/happy-cli/src/agy/AgyBackend.test.ts | 20 +++++++++++++++++-- packages/happy-cli/src/agy/AgyBackend.ts | 19 +++--------------- 2 files changed, 21 insertions(+), 18 deletions(-) diff --git a/packages/happy-cli/src/agy/AgyBackend.test.ts b/packages/happy-cli/src/agy/AgyBackend.test.ts index f6baf39684..7e4b124289 100644 --- a/packages/happy-cli/src/agy/AgyBackend.test.ts +++ b/packages/happy-cli/src/agy/AgyBackend.test.ts @@ -61,7 +61,7 @@ describe('AgyBackend', () => { expect(messages.at(-1)).toMatchObject({ type: 'status', status: 'idle' }); }); - it('maps tool lifecycle records with paired ids and redacted args', async () => { + it('maps tool lifecycle records with paired ids and no arguments', async () => { const { child, stdout } = makeFakeChild(); const spawnFn = vi.fn(() => child) as unknown as SpawnFn; const backend = new AgyBackend({ cwd: '/work', permissionMode: 'default', spawnFn, resolveConversationId: () => null }); @@ -74,11 +74,27 @@ describe('AgyBackend', () => { await turn; expect(messages.filter((m) => m.type === 'tool-call' || m.type === 'tool-result')).toEqual([ - { type: 'tool-call', toolName: 'ReadFile', callId: 'c1:2', args: { path: 'a.txt' } }, + { type: 'tool-call', toolName: 'ReadFile', callId: 'c1:2', args: {} }, { type: 'tool-result', toolName: 'ReadFile', callId: 'c1:2', result: { status: 'DONE' } }, ]); }); + it('logs malformed records without including their contents', async () => { + const { child, stdout } = makeFakeChild(); + const spawnFn = vi.fn(() => child) as unknown as SpawnFn; + const log = vi.fn(); + const backend = new AgyBackend({ cwd: '/work', permissionMode: 'default', spawnFn, log, resolveConversationId: () => null }); + + const turn = backend.sendPrompt('/work', 'hi'); + stdout.emit('data', 'secret-token\nunterminated-secret'); + child.emit('close', 0); + await turn; + + expect(log).toHaveBeenCalledWith('ignored malformed agy stream-json record (12 bytes)'); + expect(log).toHaveBeenCalledWith('ignored unterminated agy stream-json record (19 bytes)'); + expect(log.mock.calls.flat().join(' ')).not.toContain('secret'); + }); + it('emits an error status and rejects on non-zero exit', async () => { const { child } = makeFakeChild(); const spawnFn = vi.fn(() => child) as unknown as SpawnFn; diff --git a/packages/happy-cli/src/agy/AgyBackend.ts b/packages/happy-cli/src/agy/AgyBackend.ts index 94331bc899..eff2222fdb 100644 --- a/packages/happy-cli/src/agy/AgyBackend.ts +++ b/packages/happy-cli/src/agy/AgyBackend.ts @@ -58,19 +58,6 @@ type AgyStepUpdate = { tool_info?: { parameters?: unknown }; }; -function safeToolArgs(value: unknown, depth = 0): Record { - if (!value || typeof value !== 'object' || Array.isArray(value) || depth > 2) return {}; - return Object.fromEntries(Object.entries(value as Record) - .filter(([key]) => !/(token|secret|password|authorization|cookie|api[_-]?key)/i.test(key)) - .slice(0, 12) - .map(([key, item]) => { - if (typeof item === 'string') return [key, item.slice(0, 512)]; - if (typeof item === 'number' || typeof item === 'boolean' || item === null) return [key, item]; - if (Array.isArray(item)) return [key, item.slice(0, 8).map((entry) => typeof entry === 'string' ? entry.slice(0, 256) : String(entry).slice(0, 256))]; - return [key, safeToolArgs(item, depth + 1)]; - })); -} - /** Parse an agy duration string ("10m", "30s", "1h") to milliseconds; defaults to 10m. */ function parsePrintTimeoutMs(value: string): number { const m = /^(\d+)\s*(s|m|h)$/.exec(value.trim()); @@ -158,7 +145,7 @@ export class AgyBackend implements AgentBackend { try { record = JSON.parse(line) as typeof record; } catch { - this.log(`ignored malformed agy stream-json record: ${line.slice(0, 256)}`); + this.log(`ignored malformed agy stream-json record (${Buffer.byteLength(line, 'utf8')} bytes)`); return; } if (record.event === 'init' && typeof record.conversation_id === 'string') { @@ -179,7 +166,7 @@ export class AgyBackend implements AgentBackend { if (update.step_type !== 'tool' || typeof update.step_index !== 'number' || typeof update.tool_name !== 'string') return; const callId = `${streamConversationId ?? 'agy'}:${update.step_index}`; if (update.state === 'ACTIVE') { - this.emit({ type: 'tool-call', toolName: update.tool_name, callId, args: safeToolArgs(update.tool_info?.parameters) }); + this.emit({ type: 'tool-call', toolName: update.tool_name, callId, args: {} }); } else if (update.state === 'DONE' || update.state === 'ERROR') { this.emit({ type: 'tool-result', toolName: update.tool_name, callId, result: { status: update.state } }); } @@ -229,7 +216,7 @@ export class AgyBackend implements AgentBackend { if (settled) return; settled = true; cleanup(); - if (streamBuffer.trim()) this.log(`ignored unterminated agy stream-json record: ${streamBuffer.slice(0, 256)}`); + if (streamBuffer.trim()) this.log(`ignored unterminated agy stream-json record (${Buffer.byteLength(streamBuffer, 'utf8')} bytes)`); // Pin the conversation our first turn created so later turns resume it. // Once pinned, never re-read the cache: another session in the same cwd // may have updated it since, and adopting that id would cross-resume. From 9aa66d9bdbc838b3d6d3dd18a7397d9288f5e624 Mon Sep 17 00:00:00 2001 From: Timo Date: Tue, 1 Sep 2026 14:13:33 +0300 Subject: [PATCH 3/5] fix(agy): simplify structured progress mapping --- .../src/agent/acp/AcpSessionManager.test.ts | 8 - .../src/agent/acp/AcpSessionManager.ts | 2 +- .../happy-cli/src/agent/core/AgentBackend.ts | 2 +- packages/happy-cli/src/agy/AgyBackend.test.ts | 262 ++++++++++++++++- packages/happy-cli/src/agy/AgyBackend.ts | 270 ++++++++++++++---- 5 files changed, 475 insertions(+), 69 deletions(-) diff --git a/packages/happy-cli/src/agent/acp/AcpSessionManager.test.ts b/packages/happy-cli/src/agent/acp/AcpSessionManager.test.ts index 265c43ecf3..d8abfd635d 100644 --- a/packages/happy-cli/src/agent/acp/AcpSessionManager.test.ts +++ b/packages/happy-cli/src/agent/acp/AcpSessionManager.test.ts @@ -139,14 +139,6 @@ describe('AcpSessionManager text mapping', () => { expect(envelopes[1].ev).toEqual({ t: 'turn-end', status: 'completed' }); }); - it('flushes model output marked for immediate delivery', () => { - const mapper = new AcpSessionManager(); - mapper.startTurn(); - const envelopes = mapper.mapMessage({ type: 'model-output', textDelta: 'live', flush: true }); - expect(envelopes).toHaveLength(1); - expect(envelopes[0].ev).toMatchObject({ t: 'text', text: 'live' }); - }); - it('flushes accumulated output when thinking starts', () => { const mapper = new AcpSessionManager(); mapper.startTurn(); diff --git a/packages/happy-cli/src/agent/acp/AcpSessionManager.ts b/packages/happy-cli/src/agent/acp/AcpSessionManager.ts index f0b5566838..9a1c12e70a 100644 --- a/packages/happy-cli/src/agent/acp/AcpSessionManager.ts +++ b/packages/happy-cli/src/agent/acp/AcpSessionManager.ts @@ -139,7 +139,7 @@ export class AcpSessionManager { const flushed = this.pendingType !== 'output' ? this.flush() : []; this.pendingType = 'output'; this.pendingText += text; - return msg.flush ? [...flushed, ...this.flush()] : flushed; + return flushed; } if (msg.type === 'tool-call') { diff --git a/packages/happy-cli/src/agent/core/AgentBackend.ts b/packages/happy-cli/src/agent/core/AgentBackend.ts index e128a10288..cec9b73f44 100644 --- a/packages/happy-cli/src/agent/core/AgentBackend.ts +++ b/packages/happy-cli/src/agent/core/AgentBackend.ts @@ -23,7 +23,7 @@ export type ToolCallId = string; * These messages are forwarded to the Happy server and mobile app. */ export type AgentMessage = - | { type: 'model-output'; textDelta?: string; fullText?: string; flush?: boolean } + | { type: 'model-output'; textDelta?: string; fullText?: string } | { type: 'status'; status: 'starting' | 'running' | 'idle' | 'stopped' | 'error'; detail?: string } | { type: 'tool-call'; toolName: string; args: Record; callId: ToolCallId } | { type: 'tool-result'; toolName: string; result: unknown; callId: ToolCallId } diff --git a/packages/happy-cli/src/agy/AgyBackend.test.ts b/packages/happy-cli/src/agy/AgyBackend.test.ts index 7e4b124289..c9af14026a 100644 --- a/packages/happy-cli/src/agy/AgyBackend.test.ts +++ b/packages/happy-cli/src/agy/AgyBackend.test.ts @@ -3,6 +3,7 @@ import { describe, expect, it, vi } from 'vitest'; import { AgyBackend, type SpawnFn } from './AgyBackend'; import type { AgentMessage } from '@/agent/core/AgentBackend'; +import { AcpSessionManager } from '@/agent/acp/AcpSessionManager'; /** Minimal fake of a spawned child process for driving AgyBackend in tests. */ function makeFakeChild() { @@ -22,7 +23,7 @@ function makeFakeChild() { } describe('AgyBackend', () => { - it('maps a successful turn: running → model-output(s) → idle', async () => { + it('maps a successful turn: running → one complete terminal response → idle', async () => { const { child, stdout } = makeFakeChild(); const spawnFn = vi.fn(() => child) as unknown as SpawnFn; @@ -41,7 +42,7 @@ describe('AgyBackend', () => { // Split NDJSON records across arbitrary stdout chunks. stdout.emit('data', '{"event":"init","conversation_id":"c1"}\n{"event":"step_update","step_update":{"step_type":"agent_response","text_'); - stdout.emit('data', 'delta":"Hello "}}\n{"event":"step_update","step_update":{"step_type":"agent_response","text_delta":"world"}}\n'); + stdout.emit('data', 'delta":"Hello "}}\n{"event":"step_update","step_update":{"step_type":"agent_response","text_delta":"world"}}\n{"event":"result","result":{"status":"SUCCESS","response":"Hello world"}}\n'); child.emit('close', 0); await expect(turn).resolves.toBeUndefined(); @@ -55,13 +56,12 @@ describe('AgyBackend', () => { expect(spawnOpts.stdio).toEqual(['ignore', 'pipe', 'pipe']); expect(messages.filter((m) => m.type === 'model-output')).toEqual([ - { type: 'model-output', textDelta: 'Hello ', flush: true }, - { type: 'model-output', textDelta: 'world', flush: true }, + { type: 'model-output', textDelta: 'Hello world' }, ]); expect(messages.at(-1)).toMatchObject({ type: 'status', status: 'idle' }); }); - it('maps tool lifecycle records with paired ids and no arguments', async () => { + it('ignores response fragments and emits result.response exactly once', async () => { const { child, stdout } = makeFakeChild(); const spawnFn = vi.fn(() => child) as unknown as SpawnFn; const backend = new AgyBackend({ cwd: '/work', permissionMode: 'default', spawnFn, resolveConversationId: () => null }); @@ -69,32 +69,276 @@ describe('AgyBackend', () => { backend.onMessage((m) => messages.push(m)); const turn = backend.sendPrompt('/work', 'hi'); - stdout.emit('data', '{"event":"init","conversation_id":"c1"}\n{"event":"step_update","step_update":{"step_index":2,"state":"ACTIVE","step_type":"tool","tool_name":"ReadFile","tool_info":{"parameters":{"path":"a.txt","token":"nope"}}}}\n{"event":"step_update","step_update":{"step_index":2,"state":"DONE","step_type":"tool","tool_name":"ReadFile"}}\n'); + stdout.emit('data', `${JSON.stringify({ event: 'step_update', step_update: { step_type: 'agent_response', text_delta: 'split frag' } })}\n`); + stdout.emit('data', `${JSON.stringify({ event: 'step_update', step_update: { step_type: 'agent_response', text_delta: 'ment' } })}\n`); + const result = `${JSON.stringify({ event: 'result', result: { status: 'SUCCESS', response: 'split fragment' } })}\n`; + stdout.emit('data', result); + stdout.emit('data', result); + child.emit('close', 0); + await turn; + + expect(messages.filter((m) => m.type === 'model-output')).toEqual([ + { type: 'model-output', textDelta: 'split fragment' }, + ]); + }); + + it('produces one intact Happy text envelope after fragmented agy output', async () => { + const { child, stdout } = makeFakeChild(); + const spawnFn = vi.fn(() => child) as unknown as SpawnFn; + const backend = new AgyBackend({ cwd: '/work', permissionMode: 'default', spawnFn, resolveConversationId: () => null }); + const mapper = new AcpSessionManager(); + const envelopes = mapper.startTurn(); + backend.onMessage((message) => envelopes.push(...mapper.mapMessage(message))); + + const turn = backend.sendPrompt('/work', 'hi'); + const deltas = ['# Head', 'ing\n', '* first', ' item\n', '* second item']; + for (const text_delta of deltas) { + stdout.emit('data', `${JSON.stringify({ event: 'step_update', step_update: { step_type: 'agent_response', text_delta } })}\n`); + } + const response = deltas.join(''); + stdout.emit('data', `${JSON.stringify({ event: 'result', result: { status: 'SUCCESS', response } })}\n`); + child.emit('close', 0); + await turn; + envelopes.push(...mapper.endTurn('completed')); + + const textEnvelopes = envelopes.filter((envelope) => envelope.ev.t === 'text'); + expect(textEnvelopes).toHaveLength(1); + expect(textEnvelopes[0].ev).toEqual({ t: 'text', text: response }); + expect(envelopes.at(-1)?.ev).toMatchObject({ t: 'turn-end', status: 'completed' }); + }); + + it('maps tool lifecycle records with paired ids and useful inputs', async () => { + const { child, stdout } = makeFakeChild(); + const spawnFn = vi.fn(() => child) as unknown as SpawnFn; + const backend = new AgyBackend({ cwd: '/work', permissionMode: 'default', spawnFn, resolveConversationId: () => null }); + const messages: AgentMessage[] = []; + backend.onMessage((m) => messages.push(m)); + + const turn = backend.sendPrompt('/work', 'hi'); + stdout.emit('data', `${JSON.stringify({ + event: 'step_update', + step_update: { + step_index: 2, + state: 'ACTIVE', + step_type: 'tool', + tool_name: 'list_dir', + tool_info: { parameters: { DirectoryPath: '/work/tasks', token: 'nope' } }, + }, + })}\n`); + stdout.emit('data', `${JSON.stringify({ + event: 'step_update', + step_update: { + step_index: 2, + state: 'DONE', + step_type: 'tool', + tool_name: 'list_dir', + tool_info: { parameters: { DirectoryPath: '/work/tasks' }, output: 'private listing' }, + }, + })}\n`); child.emit('close', 0); await turn; expect(messages.filter((m) => m.type === 'tool-call' || m.type === 'tool-result')).toEqual([ - { type: 'tool-call', toolName: 'ReadFile', callId: 'c1:2', args: {} }, - { type: 'tool-result', toolName: 'ReadFile', callId: 'c1:2', result: { status: 'DONE' } }, + { type: 'tool-call', toolName: 'LS', callId: 'agy:1:2', args: { path: '/work/tasks' } }, + { type: 'tool-result', toolName: 'LS', callId: 'agy:1:2', result: { status: 'DONE' } }, ]); + expect(JSON.stringify(messages)).not.toContain('private listing'); + expect(JSON.stringify(messages)).not.toContain('nope'); + }); + + it('maps file and command cards while hiding credential-bearing commands', async () => { + const { child, stdout } = makeFakeChild(); + const spawnFn = vi.fn(() => child) as unknown as SpawnFn; + const backend = new AgyBackend({ cwd: '/work', permissionMode: 'default', spawnFn, resolveConversationId: () => null }); + const messages: AgentMessage[] = []; + backend.onMessage((m) => messages.push(m)); + + const secrets = [ + 'header-secret-123', + 'flag-secret-123', + 'env-secret-123', + 'url-secret-123', + 'json-secret-123', + 'curl-user-secret-123', + 'multiline-secret-123', + 'escaped-secret-123', + ]; + const commands = [ + `curl -H "Authorization: Bearer ${secrets[0]}"`, + `--token=${secrets[1]}`, + `TOKEN=${secrets[2]}`, + `https://user:${secrets[3]}@example.test/path`, + `--data '{"client_secret":"${secrets[4]}"}'`, + `curl -u user:${secrets[5]} https://example.test`, + `cat < [ + { step_index: index + 5, state: 'ACTIVE', step_type: 'tool', tool_name: 'run_command', tool_info: { parameters: { CommandLine } } }, + { step_index: index + 5, state: 'ERROR', step_type: 'tool', tool_name: 'run_command', tool_info: { error: { message: secrets[index] } } }, + ]), + ]; + const turn = backend.sendPrompt('/work', 'hi'); + for (const step_update of records) { + stdout.emit('data', `${JSON.stringify({ event: 'step_update', step_update })}\n`); + } + child.emit('close', 0); + await turn; + + const calls = messages.filter((m) => m.type === 'tool-call'); + expect(calls[0]).toEqual({ + type: 'tool-call', + toolName: 'Read', + callId: 'agy:1:3', + args: { file_path: '/work/package.json', offset: 1, limit: 20 }, + }); + expect(calls[1]).toMatchObject({ + type: 'tool-call', + toolName: 'Bash', + callId: 'agy:1:4', + args: { command: 'pwd' }, + }); + for (const [index, call] of calls.slice(2).entries()) { + expect(call).toMatchObject({ type: 'tool-call', toolName: 'Bash', callId: `agy:1:${index + 5}` }); + if (call?.type !== 'tool-call') throw new Error('expected run_command call'); + expect(call.args.command, `command ${index}`).toBe('[redacted command]'); + } + for (const secret of secrets) expect(JSON.stringify(messages)).not.toContain(secret); + }); + + it('copies only known scalar tool fields', async () => { + const { child, stdout } = makeFakeChild(); + const spawnFn = vi.fn(() => child) as unknown as SpawnFn; + const backend = new AgyBackend({ cwd: '/work', permissionMode: 'default', spawnFn, resolveConversationId: () => null }); + const messages: AgentMessage[] = []; + backend.onMessage((m) => messages.push(m)); + + const parameters = JSON.parse('{"DirectoryPath":{"safe":{"next":{"tooDeep":"hidden"}},"__proto__":{"polluted":true}},"Unexpected":"hidden"}'); + const turn = backend.sendPrompt('/work', 'hi'); + stdout.emit('data', `${JSON.stringify({ + event: 'step_update', + step_update: { step_index: 5, state: 'ACTIVE', step_type: 'tool', tool_name: 'list_dir', tool_info: { parameters } }, + })}\n`); + stdout.emit('data', `${JSON.stringify({ + event: 'step_update', + step_update: { step_index: 6, state: 'ACTIVE', step_type: 'tool', tool_name: 'grep_search', tool_info: { parameters: { Query: 'needle', SearchPath: Array.from({ length: 20 }, (_, index) => [index]) } } }, + })}\n`); + stdout.emit('data', `${JSON.stringify({ + event: 'step_update', + step_update: { step_index: 7, state: 'ACTIVE', step_type: 'tool', tool_name: 'unknown_tool', tool_info: { parameters: { CommandLine: 'echo visible' } } }, + })}\n`); + child.emit('close', 0); + await turn; + + const serialized = JSON.stringify(messages); + expect(serialized).not.toContain('polluted'); + expect(serialized).not.toContain('tooDeep'); + expect(serialized).not.toContain('Unexpected'); + const grepCall = messages.find((message) => message.type === 'tool-call' && message.toolName === 'Grep'); + expect(grepCall).toMatchObject({ args: { pattern: 'needle' } }); + if (grepCall?.type !== 'tool-call') throw new Error('expected Grep call'); + expect(grepCall.args).not.toHaveProperty('path'); + expect(messages.filter((m) => m.type === 'tool-call').at(-1)).toMatchObject({ args: {} }); }); - it('logs malformed records without including their contents', async () => { + it('deduplicates repeated tool lifecycle records', async () => { const { child, stdout } = makeFakeChild(); const spawnFn = vi.fn(() => child) as unknown as SpawnFn; + const backend = new AgyBackend({ cwd: '/work', permissionMode: 'default', spawnFn, resolveConversationId: () => null }); + const messages: AgentMessage[] = []; + backend.onMessage((m) => messages.push(m)); + + const active = { event: 'step_update', step_update: { step_index: 7, state: 'ACTIVE', step_type: 'tool', tool_name: 'list_dir', tool_info: { parameters: { DirectoryPath: '/work' } } } }; + const done = { event: 'step_update', step_update: { step_index: 7, state: 'DONE', step_type: 'tool', tool_name: 'list_dir' } }; + const turn = backend.sendPrompt('/work', 'hi'); + stdout.emit('data', `${JSON.stringify(active)}\n${JSON.stringify(active)}\n${JSON.stringify(done)}\n${JSON.stringify(done)}\n`); + child.emit('close', 0); + await turn; + + expect(messages.filter((m) => m.type === 'tool-call')).toHaveLength(1); + expect(messages.filter((m) => m.type === 'tool-result')).toHaveLength(1); + }); + + it('logs malformed records and stderr without including their contents', async () => { + const { child, stdout, stderr } = makeFakeChild(); + const spawnFn = vi.fn(() => child) as unknown as SpawnFn; const log = vi.fn(); const backend = new AgyBackend({ cwd: '/work', permissionMode: 'default', spawnFn, log, resolveConversationId: () => null }); const turn = backend.sendPrompt('/work', 'hi'); stdout.emit('data', 'secret-token\nunterminated-secret'); + stderr.emit('data', 'Authorization: Bearer stderr-secret'); child.emit('close', 0); await turn; expect(log).toHaveBeenCalledWith('ignored malformed agy stream-json record (12 bytes)'); expect(log).toHaveBeenCalledWith('ignored unterminated agy stream-json record (19 bytes)'); + expect(log).toHaveBeenCalledWith('agy stderr suppressed (35 bytes)'); + expect(log.mock.calls.flat().join(' ')).not.toContain('secret'); + }); + + it('ignores valid non-object records and continues with the stream', async () => { + const { child, stdout } = makeFakeChild(); + const spawnFn = vi.fn(() => child) as unknown as SpawnFn; + const log = vi.fn(); + const backend = new AgyBackend({ cwd: '/work', permissionMode: 'default', spawnFn, log, resolveConversationId: () => null }); + const messages: AgentMessage[] = []; + backend.onMessage((m) => messages.push(m)); + + const turn = backend.sendPrompt('/work', 'hi'); + stdout.emit('data', 'null\n42\n[]\n"hidden-string"\n'); + stdout.emit('data', `${JSON.stringify({ event: 'result', result: { status: 'SUCCESS', response: 'recovered' } })}\n`); + child.emit('close', 0); + await turn; + + expect(messages.filter((message) => message.type === 'model-output')).toEqual([ + { type: 'model-output', textDelta: 'recovered' }, + ]); + expect(log.mock.calls.flat().join(' ')).not.toContain('hidden-string'); + expect(log.mock.calls.filter(([entry]) => /^ignored non-object agy stream-json record \(\d+ bytes\)$/.test(entry))).toHaveLength(4); + }); + + it('drops oversized records, resumes at newline, and keeps payloads out of logs', async () => { + const { child, stdout } = makeFakeChild(); + const spawnFn = vi.fn(() => child) as unknown as SpawnFn; + const log = vi.fn(); + const backend = new AgyBackend({ cwd: '/work', permissionMode: 'default', spawnFn, log, resolveConversationId: () => null }); + const messages: AgentMessage[] = []; + backend.onMessage((m) => messages.push(m)); + + const turn = backend.sendPrompt('/work', 'hi'); + stdout.emit('data', `{"event":"ignored","payload":"${'x'.repeat(16 * 1024 * 1024)}secret`); + stdout.emit('data', `\n${JSON.stringify({ event: 'result', result: { status: 'SUCCESS', response: 'recovered' } })}\n`); + child.emit('close', 0); + await turn; + + expect(messages.filter((m) => m.type === 'model-output')).toEqual([ + { type: 'model-output', textDelta: 'recovered' }, + ]); + expect(log).toHaveBeenCalledWith('ignored oversized agy stream-json record (>16777216 bytes)'); expect(log.mock.calls.flat().join(' ')).not.toContain('secret'); }); + it('uses a fixed failure message instead of forwarding result status', async () => { + const { child, stdout } = makeFakeChild(); + const spawnFn = vi.fn(() => child) as unknown as SpawnFn; + const backend = new AgyBackend({ cwd: '/work', permissionMode: 'default', spawnFn, resolveConversationId: () => null }); + const messages: AgentMessage[] = []; + backend.onMessage((m) => messages.push(m)); + + const turn = backend.sendPrompt('/work', 'hi'); + stdout.emit('data', `${JSON.stringify({ event: 'result', result: { status: 'secret-status' } })}\n`); + child.emit('close', 0); + + await expect(turn).rejects.toThrow('agy stream reported failure'); + expect(JSON.stringify(messages)).not.toContain('secret-status'); + }); + it('emits an error status and rejects on non-zero exit', async () => { const { child } = makeFakeChild(); const spawnFn = vi.fn(() => child) as unknown as SpawnFn; diff --git a/packages/happy-cli/src/agy/AgyBackend.ts b/packages/happy-cli/src/agy/AgyBackend.ts index eff2222fdb..b4de08e9c0 100644 --- a/packages/happy-cli/src/agy/AgyBackend.ts +++ b/packages/happy-cli/src/agy/AgyBackend.ts @@ -1,18 +1,9 @@ /** * Agy AgentBackend Implementation * - * Custom AgentBackend that drives the agy (Antigravity) CLI. Unlike the ACP-based - * backends, agy has no streaming-event protocol — its only non-interactive surface - * is `agy --print ""`, which streams the final answer as plain text and - * exits. So this backend spawns one `agy --print` process per turn and maps: - * - * spawn → { type: 'status', status: 'running' } - * stdout chunk → { type: 'model-output', textDelta } - * exit code 0 → { type: 'status', status: 'idle' } - * exit code != 0 → { type: 'status', status: 'error', detail } - * - * There are no tool-call or permission events: print mode is one-shot and governed - * by the CLI flags (see cliArgs.ts) plus agy's own settings.json. + * Agy's print-mode stream contains response fragments, a complete terminal + * response, and tool lifecycle records. Happy needs the complete response as + * one message and tool calls translated to its native card schemas. */ import { spawn, type ChildProcess } from 'node:child_process'; @@ -48,16 +39,116 @@ export interface AgyBackendOptions { resolveConversationId?: (cwd: string) => string | null; } -type AgyStepUpdate = { - conversation_id?: unknown; - step_index?: unknown; - state?: unknown; - step_type?: unknown; - text_delta?: unknown; - tool_name?: unknown; - tool_info?: { parameters?: unknown }; +type DisplayTool = { + toolName: string; + args: Record; }; +const MAX_STREAM_RECORD_BYTES = 16 * 1024 * 1024; +const MAX_TOOL_CALLS_PER_TURN = 512; +const MAX_TOOL_NAME_BYTES = 128; +const MAX_TOOL_STEP_INDEX = 1_000_000; +const MAX_TOOL_ARG_BYTES = 4096; + +const CREDENTIAL_CARRIER_PATTERNS = [ + /\b(?:Bearer|Basic)\s+\S+/i, + /\b[a-z0-9_-]*(?:authorization|cookie|credential|password|private[-_]?key|secret|token|api[-_]?key)[a-z0-9_-]*["']?\s*(?:=|:)\s*\S+/i, + /(?:^|\s)--?[a-z0-9_-]*(?:authorization|cookie|credential|password|private[-_]?key|secret|token|api[-_]?key)[a-z0-9_-]*\s+\S+/i, + /(?:^|\s)(?:-u|--user)(?:=|\s)\s*\S+/i, + /[a-z][a-z0-9+.-]*:\/\/[^/\s@]+@/i, +]; + +function isPlainRecord(value: unknown): value is Record { + if (!value || typeof value !== 'object' || Array.isArray(value)) return false; + const prototype = Object.getPrototypeOf(value); + return prototype === Object.prototype || prototype === null; +} + +function truncateUtf8(value: string): string { + if (Buffer.byteLength(value, 'utf8') <= MAX_TOOL_ARG_BYTES) return value; + const prefix = Buffer.from(value) + .subarray(0, MAX_TOOL_ARG_BYTES - 3) + .toString('utf8') + .replace(/\uFFFD$/, ''); + return `${prefix}...`; +} + +function safeString(value: unknown, redacted = '[redacted]'): string | undefined { + if (typeof value !== 'string' || value.length === 0) return undefined; + const truncated = truncateUtf8(value); + return CREDENTIAL_CARRIER_PATTERNS.some((pattern) => pattern.test(truncated)) + ? redacted + : truncated; +} + +function safeLine(value: unknown): number | undefined { + return typeof value === 'number' && Number.isSafeInteger(value) && value >= 0 + ? value + : undefined; +} + +function safeToolName(value: string): string { + if (Buffer.byteLength(value, 'utf8') > MAX_TOOL_NAME_BYTES) return 'agy_tool'; + return /^[a-zA-Z0-9_.:-]+$/.test(value) ? value : 'agy_tool'; +} + +/** Copy only known scalar inputs and translate them to Happy's native cards. */ +function displayTool(toolName: string, parameters: unknown): DisplayTool { + const values = isPlainRecord(parameters) ? parameters : {}; + const fallback = { toolName: safeToolName(toolName), args: {} }; + + switch (toolName.toLowerCase()) { + case 'list_dir': { + const path = safeString(values.DirectoryPath); + return { toolName: 'LS', args: path ? { path } : {} }; + } + case 'view_file': + case 'sed_file': { + const filePath = safeString(values.AbsolutePath); + const offset = safeLine(values.StartLine); + const endLine = safeLine(values.EndLine); + const limit = offset !== undefined && endLine !== undefined && endLine >= offset + ? endLine - offset + 1 + : undefined; + return { + toolName: 'Read', + args: { + ...(filePath ? { file_path: filePath } : {}), + ...(offset !== undefined ? { offset } : {}), + ...(limit !== undefined ? { limit } : {}), + }, + }; + } + case 'run_command': { + const command = safeString(values.CommandLine, '[redacted command]'); + return { toolName: 'Bash', args: command ? { command } : {} }; + } + case 'grep_search': { + const pattern = safeString(values.Query); + const path = safeString(values.SearchPath); + return { + toolName: 'Grep', + args: { ...(pattern ? { pattern } : {}), ...(path ? { path } : {}) }, + }; + } + case 'find_by_name': { + const pattern = safeString(values.Pattern); + const path = safeString(values.SearchDirectory); + return { + toolName: 'Glob', + args: { ...(pattern ? { pattern } : {}), ...(path ? { path } : {}) }, + }; + } + case 'open_browser_url': + case 'read_url_content': { + const url = safeString(values.Url); + return { toolName: 'WebFetch', args: url ? { url } : {} }; + } + default: + return fallback; + } +} + /** Parse an agy duration string ("10m", "30s", "1h") to milliseconds; defaults to 10m. */ function parsePrintTimeoutMs(value: string): number { const m = /^(\d+)\s*(s|m|h)$/.exec(value.trim()); @@ -78,6 +169,7 @@ export class AgyBackend implements AgentBackend { private model?: string; private conversationId: string | null = null; private child: ChildProcess | null = null; + private turnSequence = 0; constructor(opts: AgyBackendOptions) { this.cwd = opts.cwd; @@ -109,6 +201,7 @@ export class AgyBackend implements AgentBackend { } async sendPrompt(_sessionId: SessionId, prompt: string): Promise { + const turnNumber = ++this.turnSequence; const args = buildAgyArgs({ prompt, model: this.model, @@ -137,38 +230,117 @@ export class AgyBackend implements AgentBackend { stdio: ['ignore', 'pipe', 'pipe'], }); this.child = child; + let streamBuffer = ''; - let streamConversationId: string | null = null; + let discardingOversizedRecord = false; let streamFailure: string | null = null; + let resultSeen = false; + let stderrBytes = 0; + const startedToolCalls = new Set(); + const endedToolCalls = new Set(); + const handleRecord = (line: string) => { - let record: { event?: unknown; conversation_id?: unknown; step_update?: AgyStepUpdate; result?: { status?: unknown } }; + let record: unknown; try { - record = JSON.parse(line) as typeof record; + record = JSON.parse(line) as unknown; } catch { this.log(`ignored malformed agy stream-json record (${Buffer.byteLength(line, 'utf8')} bytes)`); return; } - if (record.event === 'init' && typeof record.conversation_id === 'string') { - streamConversationId = record.conversation_id; + if (!isPlainRecord(record)) { + this.log(`ignored non-object agy stream-json record (${Buffer.byteLength(line, 'utf8')} bytes)`); return; } - if (record.event === 'result' && record.result?.status !== 'SUCCESS') { - streamFailure = `agy stream result status ${String(record.result?.status)}`; + + if (record.event === 'init') return; + if (record.event === 'result') { + if (resultSeen) return; + resultSeen = true; + const result = isPlainRecord(record.result) ? record.result : {}; + if (result.status !== 'SUCCESS') { + streamFailure = 'agy stream reported failure'; + } else if (typeof result.response === 'string' && result.response) { + this.emit({ type: 'model-output', textDelta: result.response }); + } return; } - if (record.event !== 'step_update' || !record.step_update) return; + if (record.event !== 'step_update' || !isPlainRecord(record.step_update)) return; + const update = record.step_update; - if (typeof update.conversation_id === 'string') streamConversationId = update.conversation_id; - if (update.step_type === 'agent_response' && typeof update.text_delta === 'string' && update.text_delta) { - this.emit({ type: 'model-output', textDelta: update.text_delta, flush: true }); + // The terminal result contains the same final answer intact. + if (update.step_type === 'agent_response') return; + if ( + update.step_type !== 'tool' + || typeof update.step_index !== 'number' + || !Number.isSafeInteger(update.step_index) + || update.step_index < 0 + || update.step_index > MAX_TOOL_STEP_INDEX + || typeof update.tool_name !== 'string' + ) return; + + const callId = `agy:${turnNumber}:${update.step_index}`; + const toolInfo = isPlainRecord(update.tool_info) ? update.tool_info : {}; + const display = displayTool(update.tool_name, toolInfo.parameters); + + if (update.state === 'ACTIVE') { + if (startedToolCalls.has(callId) || startedToolCalls.size >= MAX_TOOL_CALLS_PER_TURN) return; + startedToolCalls.add(callId); + this.emit({ type: 'tool-call', ...display, callId }); return; } - if (update.step_type !== 'tool' || typeof update.step_index !== 'number' || typeof update.tool_name !== 'string') return; - const callId = `${streamConversationId ?? 'agy'}:${update.step_index}`; - if (update.state === 'ACTIVE') { - this.emit({ type: 'tool-call', toolName: update.tool_name, callId, args: {} }); - } else if (update.state === 'DONE' || update.state === 'ERROR') { - this.emit({ type: 'tool-result', toolName: update.tool_name, callId, result: { status: update.state } }); + + if (update.state !== 'DONE' && update.state !== 'ERROR') return; + if (endedToolCalls.has(callId)) return; + if (!startedToolCalls.has(callId)) { + if (startedToolCalls.size >= MAX_TOOL_CALLS_PER_TURN) return; + startedToolCalls.add(callId); + this.emit({ type: 'tool-call', ...display, callId }); + } + endedToolCalls.add(callId); + this.emit({ + type: 'tool-result', + toolName: display.toolName, + callId, + result: { status: update.state }, + }); + }; + + const handleStdoutChunk = (chunk: string) => { + let remaining = chunk; + while (remaining) { + if (discardingOversizedRecord) { + const newline = remaining.indexOf('\n'); + if (newline === -1) return; + discardingOversizedRecord = false; + remaining = remaining.slice(newline + 1); + continue; + } + + const newline = remaining.indexOf('\n'); + if (newline === -1) { + const combinedBytes = Buffer.byteLength(streamBuffer, 'utf8') + + Buffer.byteLength(remaining, 'utf8'); + if (combinedBytes > MAX_STREAM_RECORD_BYTES) { + this.log(`ignored oversized agy stream-json record (>${MAX_STREAM_RECORD_BYTES} bytes)`); + streamBuffer = ''; + discardingOversizedRecord = true; + } else { + streamBuffer += remaining; + } + return; + } + + const prefix = remaining.slice(0, newline); + const recordBytes = Buffer.byteLength(streamBuffer, 'utf8') + + Buffer.byteLength(prefix, 'utf8'); + if (recordBytes > MAX_STREAM_RECORD_BYTES) { + this.log(`ignored oversized agy stream-json record (${recordBytes} bytes)`); + } else { + const record = streamBuffer + prefix; + if (record.trim()) handleRecord(record); + } + streamBuffer = ''; + remaining = remaining.slice(newline + 1); } }; @@ -186,25 +358,18 @@ export class AgyBackend implements AgentBackend { }; child.stdout?.setEncoding('utf8'); - child.stdout?.on('data', (chunk: string) => { - streamBuffer += chunk; - const records = streamBuffer.split('\n'); - streamBuffer = records.pop() ?? ''; - for (const record of records) if (record.trim()) handleRecord(record); - }); + child.stdout?.on('data', handleStdoutChunk); child.stderr?.setEncoding('utf8'); child.stderr?.on('data', (chunk: string) => { - const text = chunk.trimEnd(); - if (text) { - this.log(`stderr: ${text}`); - } + stderrBytes += Buffer.byteLength(chunk, 'utf8'); }); child.on('error', (err: Error) => { if (settled) return; settled = true; cleanup(); + if (stderrBytes > 0) this.log(`agy stderr suppressed (${stderrBytes} bytes)`); const detail = (err as NodeJS.ErrnoException).code === 'ENOENT' ? `agy executable not found. Install the Antigravity CLI, or set HAPPY_AGY_PATH to its absolute path (tried 'agy' on PATH and ~/.local/bin/agy).` : err.message; @@ -216,7 +381,11 @@ export class AgyBackend implements AgentBackend { if (settled) return; settled = true; cleanup(); - if (streamBuffer.trim()) this.log(`ignored unterminated agy stream-json record (${Buffer.byteLength(streamBuffer, 'utf8')} bytes)`); + if (stderrBytes > 0) this.log(`agy stderr suppressed (${stderrBytes} bytes)`); + if (streamBuffer.trim()) { + this.log(`ignored unterminated agy stream-json record (${Buffer.byteLength(streamBuffer, 'utf8')} bytes)`); + } + // Pin the conversation our first turn created so later turns resume it. // Once pinned, never re-read the cache: another session in the same cwd // may have updated it since, and adopting that id would cross-resume. @@ -236,11 +405,12 @@ export class AgyBackend implements AgentBackend { } } - if (code === 0 && !streamFailure) { + const detail = streamFailure + ?? (code === 0 ? null : `agy exited with code ${code ?? 'null'}`); + if (!detail) { this.emit({ type: 'status', status: 'idle' }); resolve(); } else { - const detail = streamFailure ?? `agy exited with code ${code ?? 'null'}`; this.emit({ type: 'status', status: 'error', detail }); reject(new Error(detail)); } From f0af1849cbdd3013ed61fbd32bd3974b04ad3e9a Mon Sep 17 00:00:00 2001 From: Timo Date: Wed, 2 Sep 2026 06:35:27 +0300 Subject: [PATCH 4/5] fix(agy): render structured tools and live progress --- packages/happy-cli/src/agy/AgyBackend.test.ts | 189 ++++++++++++++++-- packages/happy-cli/src/agy/AgyBackend.ts | 115 ++++++++++- 2 files changed, 280 insertions(+), 24 deletions(-) diff --git a/packages/happy-cli/src/agy/AgyBackend.test.ts b/packages/happy-cli/src/agy/AgyBackend.test.ts index c9af14026a..31658ab7a8 100644 --- a/packages/happy-cli/src/agy/AgyBackend.test.ts +++ b/packages/happy-cli/src/agy/AgyBackend.test.ts @@ -61,7 +61,7 @@ describe('AgyBackend', () => { expect(messages.at(-1)).toMatchObject({ type: 'status', status: 'idle' }); }); - it('ignores response fragments and emits result.response exactly once', async () => { + it('buffers response fragments and does not duplicate the terminal aggregate', async () => { const { child, stdout } = makeFakeChild(); const spawnFn = vi.fn(() => child) as unknown as SpawnFn; const backend = new AgyBackend({ cwd: '/work', permissionMode: 'default', spawnFn, resolveConversationId: () => null }); @@ -69,8 +69,8 @@ describe('AgyBackend', () => { backend.onMessage((m) => messages.push(m)); const turn = backend.sendPrompt('/work', 'hi'); - stdout.emit('data', `${JSON.stringify({ event: 'step_update', step_update: { step_type: 'agent_response', text_delta: 'split frag' } })}\n`); - stdout.emit('data', `${JSON.stringify({ event: 'step_update', step_update: { step_type: 'agent_response', text_delta: 'ment' } })}\n`); + stdout.emit('data', `${JSON.stringify({ event: 'step_update', step_update: { step_index: 1, state: 'ACTIVE', step_type: 'agent_response', text_delta: 'split frag' } })}\n`); + stdout.emit('data', `${JSON.stringify({ event: 'step_update', step_update: { step_index: 1, state: 'DONE', step_type: 'agent_response', text_delta: 'ment' } })}\n`); const result = `${JSON.stringify({ event: 'result', result: { status: 'SUCCESS', response: 'split fragment' } })}\n`; stdout.emit('data', result); stdout.emit('data', result); @@ -82,7 +82,7 @@ describe('AgyBackend', () => { ]); }); - it('produces one intact Happy text envelope after fragmented agy output', async () => { + it('emits intact progress before tools and final text after them', async () => { const { child, stdout } = makeFakeChild(); const spawnFn = vi.fn(() => child) as unknown as SpawnFn; const backend = new AgyBackend({ cwd: '/work', permissionMode: 'default', spawnFn, resolveConversationId: () => null }); @@ -91,19 +91,32 @@ describe('AgyBackend', () => { backend.onMessage((message) => envelopes.push(...mapper.mapMessage(message))); const turn = backend.sendPrompt('/work', 'hi'); - const deltas = ['# Head', 'ing\n', '* first', ' item\n', '* second item']; - for (const text_delta of deltas) { - stdout.emit('data', `${JSON.stringify({ event: 'step_update', step_update: { step_type: 'agent_response', text_delta } })}\n`); - } - const response = deltas.join(''); + stdout.emit('data', `${JSON.stringify({ event: 'step_update', step_update: { step_index: 1, state: 'ACTIVE', step_type: 'agent_response', text_delta: '# Head' } })}\n`); + stdout.emit('data', `${JSON.stringify({ event: 'step_update', step_update: { step_index: 1, state: 'DONE', step_type: 'agent_response', text_delta: 'ing\n' } })}\n`); + stdout.emit('data', `${JSON.stringify({ event: 'step_update', step_update: { step_index: 2, state: 'ACTIVE', step_type: 'tool', tool_name: 'list_dir', tool_info: { parameters: { DirectoryPath: '/work' } } } })}\n`); + stdout.emit('data', `${JSON.stringify({ event: 'step_update', step_update: { step_index: 2, state: 'DONE', step_type: 'tool', tool_name: 'list_dir' } })}\n`); + stdout.emit('data', `${JSON.stringify({ event: 'step_update', step_update: { step_index: 3, state: 'ACTIVE', step_type: 'agent_response', text_delta: '* first' } })}\n`); + stdout.emit('data', `${JSON.stringify({ event: 'step_update', step_update: { step_index: 3, state: 'DONE', step_type: 'agent_response', text_delta: ' item' } })}\n`); + const response = '# Heading\n* first item'; stdout.emit('data', `${JSON.stringify({ event: 'result', result: { status: 'SUCCESS', response } })}\n`); child.emit('close', 0); await turn; envelopes.push(...mapper.endTurn('completed')); const textEnvelopes = envelopes.filter((envelope) => envelope.ev.t === 'text'); - expect(textEnvelopes).toHaveLength(1); - expect(textEnvelopes[0].ev).toEqual({ t: 'text', text: response }); + expect(textEnvelopes).toHaveLength(2); + expect(textEnvelopes.map((envelope) => envelope.ev)).toEqual([ + { t: 'text', text: '# Heading' }, + { t: 'text', text: '* first item' }, + ]); + expect(envelopes.map((envelope) => envelope.ev.t)).toEqual([ + 'turn-start', + 'text', + 'tool-call-start', + 'tool-call-end', + 'text', + 'turn-end', + ]); expect(envelopes.at(-1)?.ev).toMatchObject({ t: 'turn-end', status: 'completed' }); }); @@ -135,15 +148,146 @@ describe('AgyBackend', () => { tool_info: { parameters: { DirectoryPath: '/work/tasks' }, output: 'private listing' }, }, })}\n`); + for (const step_update of [ + { + step_index: 3, + state: 'ACTIVE', + step_type: 'tool', + tool_name: 'replace_file_content', + tool_info: { + parameters: { + TargetFile: '/work/TODO.md', + TargetContent: 'private old content', + ReplacementContent: 'private replacement content', + }, + }, + }, + { step_index: 3, state: 'DONE', step_type: 'tool', tool_name: 'replace_file_content' }, + { + step_index: 4, + state: 'ACTIVE', + step_type: 'tool', + tool_name: 'write_to_file', + tool_info: { parameters: { TargetFile: '/work/new.md', CodeContent: 'private new content' } }, + }, + { step_index: 4, state: 'DONE', step_type: 'tool', tool_name: 'write_to_file' }, + { + step_index: 5, + state: 'ACTIVE', + step_type: 'tool', + tool_name: 'manage_task', + tool_info: { + parameters: { + Action: 'status', + TaskId: 'private-conversation-id/task-20', + Input: 'private task input', + toolSummary: 'private task summary', + }, + }, + }, + { + step_index: 5, + state: 'DONE', + step_type: 'tool', + tool_name: 'manage_task', + tool_info: { output: 'Task is still running' }, + }, + { + step_index: 6, + state: 'ACTIVE', + step_type: 'tool', + tool_name: 'manage_task', + tool_info: { parameters: { Action: 'kill', TaskId: 'private-conversation-id/task-30' } }, + }, + { + step_index: 6, + state: 'DONE', + step_type: 'tool', + tool_name: 'manage_task', + tool_info: { output: 'Task killed' }, + }, + { + step_index: 7, + state: 'ACTIVE', + step_type: 'tool', + tool_name: 'call_mcp_tool', + tool_info: { + parameters: { + ServerName: 'happy', + ToolName: 'change_title', + Arguments: { title: 'private title' }, + }, + }, + }, + { + step_index: 7, + state: 'DONE', + step_type: 'tool', + tool_name: 'call_mcp_tool', + tool_info: { output: 'Title changed' }, + }, + { + step_index: 8, + state: 'ACTIVE', + step_type: 'tool', + tool_name: 'search_web', + tool_info: { parameters: { query: 'Agy documentation', toolSummary: 'private search summary' } }, + }, + { + step_index: 8, + state: 'DONE', + step_type: 'tool', + tool_name: 'search_web', + tool_info: { output: 'Search result' }, + }, + ]) { + stdout.emit('data', `${JSON.stringify({ event: 'step_update', step_update })}\n`); + } child.emit('close', 0); await turn; expect(messages.filter((m) => m.type === 'tool-call' || m.type === 'tool-result')).toEqual([ { type: 'tool-call', toolName: 'LS', callId: 'agy:1:2', args: { path: '/work/tasks' } }, { type: 'tool-result', toolName: 'LS', callId: 'agy:1:2', result: { status: 'DONE' } }, + { type: 'tool-call', toolName: 'Edit', callId: 'agy:1:3', args: { file_path: '/work/TODO.md' } }, + { type: 'tool-result', toolName: 'Edit', callId: 'agy:1:3', result: { status: 'DONE' } }, + { type: 'tool-call', toolName: 'Write', callId: 'agy:1:4', args: { file_path: '/work/new.md' } }, + { type: 'tool-result', toolName: 'Write', callId: 'agy:1:4', result: { status: 'DONE' } }, + { type: 'tool-call', toolName: 'Bash', callId: 'agy:1:5', args: { command: 'manage_task status task-20' } }, + { type: 'tool-result', toolName: 'Bash', callId: 'agy:1:5', result: { status: 'DONE', stdout: 'Task is still running' } }, + { type: 'tool-call', toolName: 'Bash', callId: 'agy:1:6', args: { command: 'manage_task kill task-30' } }, + { type: 'tool-result', toolName: 'Bash', callId: 'agy:1:6', result: { status: 'DONE', stdout: 'Task killed' } }, + { type: 'tool-call', toolName: 'Bash', callId: 'agy:1:7', args: { command: 'mcp happy.change_title' } }, + { type: 'tool-result', toolName: 'Bash', callId: 'agy:1:7', result: { status: 'DONE', stdout: 'Title changed' } }, + { type: 'tool-call', toolName: 'WebSearch', callId: 'agy:1:8', args: { query: 'Agy documentation' } }, + { type: 'tool-result', toolName: 'WebSearch', callId: 'agy:1:8', result: { status: 'DONE', stdout: 'Search result' } }, + ]); + expect(messages.filter((m) => m.type === 'event')).toEqual([ + { type: 'event', name: 'thinking', payload: { text: 'Task is still running', streaming: false } }, + { type: 'event', name: 'thinking', payload: { text: 'Task killed', streaming: false } }, + { type: 'event', name: 'thinking', payload: { text: 'Title changed', streaming: false } }, + { type: 'event', name: 'thinking', payload: { text: 'Search result', streaming: false } }, ]); - expect(JSON.stringify(messages)).not.toContain('private listing'); - expect(JSON.stringify(messages)).not.toContain('nope'); + const mapper = new AcpSessionManager(); + const envelopes = mapper.startTurn(); + for (const message of messages) envelopes.push(...mapper.mapMessage(message)); + envelopes.push(...mapper.endTurn('completed')); + expect( + envelopes + .filter((envelope) => envelope.ev.t === 'text' && envelope.ev.thinking) + .map((envelope) => envelope.ev.t === 'text' ? envelope.ev.text : ''), + ).toEqual(['Task is still running', 'Task killed', 'Title changed', 'Search result']); + const serialized = JSON.stringify(messages); + expect(serialized).not.toContain('private listing'); + expect(serialized).not.toContain('private old content'); + expect(serialized).not.toContain('private replacement content'); + expect(serialized).not.toContain('private new content'); + expect(serialized).not.toContain('private-conversation-id'); + expect(serialized).not.toContain('private task input'); + expect(serialized).not.toContain('private task summary'); + expect(serialized).not.toContain('private title'); + expect(serialized).not.toContain('private search summary'); + expect(serialized).not.toContain('nope'); }); it('maps file and command cards while hiding credential-bearing commands', async () => { @@ -177,10 +321,16 @@ describe('AgyBackend', () => { { step_index: 3, state: 'ACTIVE', step_type: 'tool', tool_name: 'view_file', tool_info: { parameters: { AbsolutePath: '/work/package.json', StartLine: 1, EndLine: 20 } } }, { step_index: 3, state: 'DONE', step_type: 'tool', tool_name: 'view_file', tool_info: { output: secrets[0] } }, { step_index: 4, state: 'ACTIVE', step_type: 'tool', tool_name: 'run_command', tool_info: { parameters: { CommandLine: 'pwd' } } }, - { step_index: 4, state: 'DONE', step_type: 'tool', tool_name: 'run_command' }, + { step_index: 4, state: 'DONE', step_type: 'tool', tool_name: 'run_command', tool_info: { output: 'command output\n' } }, ...commands.flatMap((CommandLine, index) => [ { step_index: index + 5, state: 'ACTIVE', step_type: 'tool', tool_name: 'run_command', tool_info: { parameters: { CommandLine } } }, - { step_index: index + 5, state: 'ERROR', step_type: 'tool', tool_name: 'run_command', tool_info: { error: { message: secrets[index] } } }, + { + step_index: index + 5, + state: 'ERROR', + step_type: 'tool', + tool_name: 'run_command', + tool_info: { output: `token=${secrets[index]}`, error: { message: secrets[index] } }, + }, ]), ]; const turn = backend.sendPrompt('/work', 'hi'); @@ -203,11 +353,20 @@ describe('AgyBackend', () => { callId: 'agy:1:4', args: { command: 'pwd' }, }); + expect(messages).toContainEqual({ + type: 'tool-result', + toolName: 'Bash', + callId: 'agy:1:4', + result: { status: 'DONE', stdout: 'command output\n' }, + }); for (const [index, call] of calls.slice(2).entries()) { expect(call).toMatchObject({ type: 'tool-call', toolName: 'Bash', callId: `agy:1:${index + 5}` }); if (call?.type !== 'tool-call') throw new Error('expected run_command call'); expect(call.args.command, `command ${index}`).toBe('[redacted command]'); } + for (const result of messages.filter((m) => m.type === 'tool-result').slice(2)) { + expect(result).toMatchObject({ result: { status: 'ERROR', stderr: '[redacted output]' } }); + } for (const secret of secrets) expect(JSON.stringify(messages)).not.toContain(secret); }); diff --git a/packages/happy-cli/src/agy/AgyBackend.ts b/packages/happy-cli/src/agy/AgyBackend.ts index b4de08e9c0..d09fea2e4d 100644 --- a/packages/happy-cli/src/agy/AgyBackend.ts +++ b/packages/happy-cli/src/agy/AgyBackend.ts @@ -2,8 +2,9 @@ * Agy AgentBackend Implementation * * Agy's print-mode stream contains response fragments, a complete terminal - * response, and tool lifecycle records. Happy needs the complete response as - * one message and tool calls translated to its native card schemas. + * response, and tool lifecycle records. Happy needs fragments buffered into + * complete progress messages, without replaying the terminal aggregate, and + * tool calls translated to its native card schemas. */ import { spawn, type ChildProcess } from 'node:child_process'; @@ -46,9 +47,17 @@ type DisplayTool = { const MAX_STREAM_RECORD_BYTES = 16 * 1024 * 1024; const MAX_TOOL_CALLS_PER_TURN = 512; +const MAX_AGENT_RESPONSE_STEPS = 512; const MAX_TOOL_NAME_BYTES = 128; const MAX_TOOL_STEP_INDEX = 1_000_000; const MAX_TOOL_ARG_BYTES = 4096; +const MAX_TOOL_OUTPUT_BYTES = 64 * 1024; +const VISIBLE_OUTPUT_TOOLS = new Set([ + 'call_mcp_tool', + 'manage_task', + 'run_command', + 'search_web', +]); const CREDENTIAL_CARRIER_PATTERNS = [ /\b(?:Bearer|Basic)\s+\S+/i, @@ -64,10 +73,10 @@ function isPlainRecord(value: unknown): value is Record { return prototype === Object.prototype || prototype === null; } -function truncateUtf8(value: string): string { - if (Buffer.byteLength(value, 'utf8') <= MAX_TOOL_ARG_BYTES) return value; +function truncateUtf8(value: string, maxBytes = MAX_TOOL_ARG_BYTES): string { + if (Buffer.byteLength(value, 'utf8') <= maxBytes) return value; const prefix = Buffer.from(value) - .subarray(0, MAX_TOOL_ARG_BYTES - 3) + .subarray(0, maxBytes - 3) .toString('utf8') .replace(/\uFFFD$/, ''); return `${prefix}...`; @@ -81,6 +90,14 @@ function safeString(value: unknown, redacted = '[redacted]'): string | undefined : truncated; } +function safeToolOutput(value: unknown): string | undefined { + if (typeof value !== 'string' || value.length === 0) return undefined; + const truncated = truncateUtf8(value, MAX_TOOL_OUTPUT_BYTES); + return CREDENTIAL_CARRIER_PATTERNS.some((pattern) => pattern.test(truncated)) + ? '[redacted output]' + : truncated; +} + function safeLine(value: unknown): number | undefined { return typeof value === 'number' && Number.isSafeInteger(value) && value >= 0 ? value @@ -123,6 +140,30 @@ function displayTool(toolName: string, parameters: unknown): DisplayTool { const command = safeString(values.CommandLine, '[redacted command]'); return { toolName: 'Bash', args: command ? { command } : {} }; } + case 'replace_file_content': { + const filePath = safeString(values.TargetFile); + return { toolName: 'Edit', args: filePath ? { file_path: filePath } : {} }; + } + case 'write_to_file': { + const filePath = safeString(values.TargetFile); + return { toolName: 'Write', args: filePath ? { file_path: filePath } : {} }; + } + case 'manage_task': { + const action = values.Action === 'status' || values.Action === 'kill' ? values.Action : undefined; + const taskId = safeString(values.TaskId)?.split('/').filter(Boolean).at(-1); + const command = ['manage_task', action, taskId].filter(Boolean).join(' '); + return { toolName: 'Bash', args: { command } }; + } + case 'call_mcp_tool': { + const server = safeString(values.ServerName); + const tool = safeString(values.ToolName); + const target = [server, tool].filter(Boolean).join('.'); + return { toolName: 'Bash', args: { command: target ? `mcp ${target}` : 'mcp tool' } }; + } + case 'search_web': { + const query = safeString(values.query); + return { toolName: 'WebSearch', args: query ? { query } : {} }; + } case 'grep_search': { const pattern = safeString(values.Query); const path = safeString(values.SearchPath); @@ -238,6 +279,17 @@ export class AgyBackend implements AgentBackend { let stderrBytes = 0; const startedToolCalls = new Set(); const endedToolCalls = new Set(); + const agentResponseBuffers = new Map(); + let streamedResponseText = ''; + let agentResponseBytes = 0; + + const emitAgentResponse = (stepIndex: number) => { + const text = agentResponseBuffers.get(stepIndex); + agentResponseBuffers.delete(stepIndex); + if (!text) return; + streamedResponseText += text; + this.emit({ type: 'model-output', textDelta: text }); + }; const handleRecord = (line: string) => { let record: unknown; @@ -256,19 +308,46 @@ export class AgyBackend implements AgentBackend { if (record.event === 'result') { if (resultSeen) return; resultSeen = true; + for (const stepIndex of [...agentResponseBuffers.keys()].sort((a, b) => a - b)) { + emitAgentResponse(stepIndex); + } const result = isPlainRecord(record.result) ? record.result : {}; if (result.status !== 'SUCCESS') { streamFailure = 'agy stream reported failure'; } else if (typeof result.response === 'string' && result.response) { - this.emit({ type: 'model-output', textDelta: result.response }); + const remaining = result.response.startsWith(streamedResponseText) + ? result.response.slice(streamedResponseText.length) + : streamedResponseText + ? '' + : result.response; + if (remaining) this.emit({ type: 'model-output', textDelta: remaining }); } return; } if (record.event !== 'step_update' || !isPlainRecord(record.step_update)) return; const update = record.step_update; - // The terminal result contains the same final answer intact. - if (update.step_type === 'agent_response') return; + if (update.step_type === 'agent_response') { + if ( + typeof update.step_index !== 'number' + || !Number.isSafeInteger(update.step_index) + || update.step_index < 0 + || update.step_index > MAX_TOOL_STEP_INDEX + ) return; + if (typeof update.text_delta === 'string' && update.text_delta) { + if ( + !agentResponseBuffers.has(update.step_index) + && agentResponseBuffers.size >= MAX_AGENT_RESPONSE_STEPS + ) return; + const deltaBytes = Buffer.byteLength(update.text_delta, 'utf8'); + if (agentResponseBytes + deltaBytes > MAX_STREAM_RECORD_BYTES) return; + const current = agentResponseBuffers.get(update.step_index) ?? ''; + agentResponseBuffers.set(update.step_index, current + update.text_delta); + agentResponseBytes += deltaBytes; + } + if (update.state === 'DONE') emitAgentResponse(update.step_index); + return; + } if ( update.step_type !== 'tool' || typeof update.step_index !== 'number' @@ -297,11 +376,24 @@ export class AgyBackend implements AgentBackend { this.emit({ type: 'tool-call', ...display, callId }); } endedToolCalls.add(callId); + const normalizedToolName = update.tool_name.toLowerCase(); + const toolOutput = VISIBLE_OUTPUT_TOOLS.has(normalizedToolName) + ? safeToolOutput(toolInfo.output) + : undefined; + const toolResult: Record = { status: update.state }; + if (toolOutput) { + toolResult[update.state === 'ERROR' ? 'stderr' : 'stdout'] = toolOutput; + this.emit({ + type: 'event', + name: 'thinking', + payload: { text: toolOutput, streaming: false }, + }); + } this.emit({ type: 'tool-result', toolName: display.toolName, callId, - result: { status: update.state }, + result: toolResult, }); }; @@ -385,6 +477,11 @@ export class AgyBackend implements AgentBackend { if (streamBuffer.trim()) { this.log(`ignored unterminated agy stream-json record (${Buffer.byteLength(streamBuffer, 'utf8')} bytes)`); } + if (!resultSeen) { + for (const stepIndex of [...agentResponseBuffers.keys()].sort((a, b) => a - b)) { + emitAgentResponse(stepIndex); + } + } // Pin the conversation our first turn created so later turns resume it. // Once pinned, never re-read the cache: another session in the same cwd From fde629789630d10ba1b774e029c33bd1c4b10de3 Mon Sep 17 00:00:00 2001 From: Timo Date: Wed, 2 Sep 2026 19:49:45 +0300 Subject: [PATCH 5/5] fix(agy): set Happy session titles --- packages/happy-cli/src/agy/AgyBackend.test.ts | 115 ++++++++++++++++++ packages/happy-cli/src/agy/AgyBackend.ts | 39 +++++- packages/happy-cli/src/agy/cliArgs.ts | 6 +- packages/happy-cli/src/agy/runAgy.ts | 14 +++ 4 files changed, 170 insertions(+), 4 deletions(-) diff --git a/packages/happy-cli/src/agy/AgyBackend.test.ts b/packages/happy-cli/src/agy/AgyBackend.test.ts index 31658ab7a8..aed06aeb00 100644 --- a/packages/happy-cli/src/agy/AgyBackend.test.ts +++ b/packages/happy-cli/src/agy/AgyBackend.test.ts @@ -1,6 +1,7 @@ import { EventEmitter } from 'node:events'; import { describe, expect, it, vi } from 'vitest'; +import { CHANGE_TITLE_INSTRUCTION } from '@/gemini/constants'; import { AgyBackend, type SpawnFn } from './AgyBackend'; import type { AgentMessage } from '@/agent/core/AgentBackend'; import { AcpSessionManager } from '@/agent/acp/AcpSessionManager'; @@ -61,6 +62,120 @@ describe('AgyBackend', () => { expect(messages.at(-1)).toMatchObject({ type: 'status', status: 'idle' }); }); + it('appends the Happy title instruction only to the first turn', async () => { + const spawnCalls: string[][] = []; + let current = makeFakeChild(); + const spawnFn = vi.fn((_bin: string, args: string[]) => { + spawnCalls.push(args); + return current.child; + }) as unknown as SpawnFn; + + const backend = new AgyBackend({ + cwd: '/work', + permissionMode: 'yolo', + spawnFn, + resolveConversationId: () => null, + }); + + const firstTurn = backend.sendPrompt('/work', 'first task'); + current.child.emit('close', 0); + await firstTurn; + + current = makeFakeChild(); + const secondTurn = backend.sendPrompt('/work', 'continue'); + current.child.emit('close', 0); + await secondTurn; + + const promptFromArgs = (args: string[]) => { + const printIndex = args.indexOf('--print'); + expect(printIndex).toBeGreaterThanOrEqual(0); + return args[printIndex + 1]; + }; + + expect(promptFromArgs(spawnCalls[0])).toBe( + `first task\n\n${CHANGE_TITLE_INSTRUCTION}`, + ); + expect(promptFromArgs(spawnCalls[1])).toBe('continue'); + }); + + it('sets a safe fallback title without invoking the MCP tool in sandbox mode', async () => { + const { child } = makeFakeChild(); + const spawnFn = vi.fn(() => child) as unknown as SpawnFn; + const onTitle = vi.fn(); + const backend = new AgyBackend({ + cwd: '/work', + permissionMode: 'default', + spawnFn, + resolveConversationId: () => null, + onTitle, + }); + + const turn = backend.sendPrompt( + '/work', + 'Summarize the infrastructure repository purpose.', + ); + child.emit('close', 0); + await turn; + + expect(onTitle).toHaveBeenCalledWith( + 'Summarize the infrastructure repository purpose', + ); + const args = (spawnFn as unknown as ReturnType).mock.calls[0][1] as string[]; + const prompt = args[args.indexOf('--print') + 1]; + expect(prompt).toContain('Happy has already set the session title'); + expect(prompt).not.toContain('functions.happy__change_title'); + }); + + it('bounds long fallback titles at a word boundary', async () => { + const { child } = makeFakeChild(); + const spawnFn = vi.fn(() => child) as unknown as SpawnFn; + const onTitle = vi.fn(); + const backend = new AgyBackend({ + cwd: '/work', + permissionMode: 'default', + spawnFn, + resolveConversationId: () => null, + onTitle, + }); + + const turn = backend.sendPrompt( + '/work', + 'Investigate why the Happy mobile application does not receive the final Agy response after setting its title', + ); + child.emit('close', 0); + await turn; + + const title = onTitle.mock.calls[0][0] as string; + expect([...title].length).toBeLessThanOrEqual(60); + expect(title).toMatch(/…$/); + }); + + it('passes the session-scoped Happy MCP URL to agy', async () => { + const { child } = makeFakeChild(); + const spawnFn = vi.fn(() => child) as unknown as SpawnFn; + const env = { + PATH: '/test/bin', + HAPPY_HTTP_MCP_URL: 'http://127.0.0.1:43210', + }; + const backend = new AgyBackend({ + cwd: '/work', + permissionMode: 'default', + spawnFn, + resolveConversationId: () => null, + env, + }); + + const turn = backend.sendPrompt('/work', 'hi'); + child.emit('close', 0); + await turn; + + expect(spawnFn).toHaveBeenCalledWith( + expect.any(String), + expect.any(Array), + expect.objectContaining({ env }), + ); + }); + it('buffers response fragments and does not duplicate the terminal aggregate', async () => { const { child, stdout } = makeFakeChild(); const spawnFn = vi.fn(() => child) as unknown as SpawnFn; diff --git a/packages/happy-cli/src/agy/AgyBackend.ts b/packages/happy-cli/src/agy/AgyBackend.ts index d09fea2e4d..a43f79e23f 100644 --- a/packages/happy-cli/src/agy/AgyBackend.ts +++ b/packages/happy-cli/src/agy/AgyBackend.ts @@ -16,8 +16,9 @@ import type { SessionId, StartSessionResult, } from '@/agent/core/AgentBackend'; +import { CHANGE_TITLE_INSTRUCTION } from '@/gemini/constants'; import { resolveAgyBin, AGY_PRINT_TIMEOUT } from './constants'; -import { buildAgyArgs } from './cliArgs'; +import { agySkipsPermissions, buildAgyArgs } from './cliArgs'; import { readAgyConversationId } from './conversationStore'; /** Signature of node's `spawn`, injectable so tests can supply a fake process. */ @@ -38,6 +39,10 @@ export interface AgyBackendOptions { spawnFn?: SpawnFn; /** Optional override for resolving the resume conversation id (tests). */ resolveConversationId?: (cwd: string) => string | null; + /** Environment inherited by each agy child process. */ + env?: NodeJS.ProcessEnv; + /** Sets a safe fallback title when sandbox mode cannot approve the MCP call. */ + onTitle?: (title: string) => void; } type DisplayTool = { @@ -58,6 +63,20 @@ const VISIBLE_OUTPUT_TOOLS = new Set([ 'run_command', 'search_web', ]); +const TITLE_ALREADY_SET_INSTRUCTION = + 'Happy has already set the session title for this turn. Do not call happy.change_title or any other title tool.'; +const MAX_FALLBACK_TITLE_CHARS = 60; + +function fallbackTitle(prompt: string): string | null { + const compact = prompt.replace(/\s+/g, ' ').trim(); + if (!compact) return null; + const chars = [...compact]; + if (chars.length <= MAX_FALLBACK_TITLE_CHARS) return compact.replace(/[.!?]+$/, ''); + const prefix = chars.slice(0, MAX_FALLBACK_TITLE_CHARS - 1).join(''); + const wordBoundary = prefix.lastIndexOf(' '); + const shortened = wordBoundary >= 30 ? prefix.slice(0, wordBoundary) : prefix; + return `${shortened.replace(/[\s.!?,;:]+$/, '')}…`; +} const CREDENTIAL_CARRIER_PATTERNS = [ /\b(?:Bearer|Basic)\s+\S+/i, @@ -205,6 +224,8 @@ export class AgyBackend implements AgentBackend { private readonly log: (msg: string) => void; private readonly spawnFn: SpawnFn; private readonly resolveConversationId: (cwd: string) => string | null; + private readonly env: NodeJS.ProcessEnv; + private readonly onTitle?: (title: string) => void; private permissionMode: PermissionMode; private model?: string; @@ -220,6 +241,8 @@ export class AgyBackend implements AgentBackend { this.log = opts.log ?? (() => {}); this.spawnFn = opts.spawnFn ?? spawn; this.resolveConversationId = opts.resolveConversationId ?? readAgyConversationId; + this.env = opts.env ?? process.env; + this.onTitle = opts.onTitle; } /** Update the permission mode applied to subsequent turns. */ @@ -243,8 +266,18 @@ export class AgyBackend implements AgentBackend { async sendPrompt(_sessionId: SessionId, prompt: string): Promise { const turnNumber = ++this.turnSequence; + let turnPrompt = prompt; + if (turnNumber === 1) { + if (agySkipsPermissions(this.permissionMode)) { + turnPrompt = `${prompt}\n\n${CHANGE_TITLE_INSTRUCTION}`; + } else { + const title = fallbackTitle(prompt); + if (title) this.onTitle?.(title); + turnPrompt = `${prompt}\n\n${TITLE_ALREADY_SET_INSTRUCTION}`; + } + } const args = buildAgyArgs({ - prompt, + prompt: turnPrompt, model: this.model, conversationId: this.conversationId, permissionMode: this.permissionMode, @@ -263,7 +296,7 @@ export class AgyBackend implements AgentBackend { await new Promise((resolve, reject) => { const child = this.spawnFn(resolveAgyBin(), args, { cwd: this.cwd, - env: process.env, + env: this.env, windowsHide: true, // agy --print blocks until stdin reaches EOF. We never write stdin, so // give the child an empty stdin (immediate EOF) instead of an open pipe; diff --git a/packages/happy-cli/src/agy/cliArgs.ts b/packages/happy-cli/src/agy/cliArgs.ts index 5ae40e3020..59b44839a7 100644 --- a/packages/happy-cli/src/agy/cliArgs.ts +++ b/packages/happy-cli/src/agy/cliArgs.ts @@ -21,6 +21,10 @@ const SKIP_PERMISSION_MODES: ReadonlySet = new Set { let thinking = false; let displayedModel = DEFAULT_AGY_MODEL; + const happyServer = await startHappyServer(session); const backend = new AgyBackend({ cwd: process.cwd(), permissionMode: 'default', model: DEFAULT_AGY_MODEL, log, + env: { + ...process.env, + HAPPY_HTTP_MCP_URL: happyServer.url, + }, + onTitle: (title) => { + session.sendClaudeSessionMessage({ + type: 'summary', + summary: title, + leafUuid: randomUUID(), + }); + }, }); // Terminal UI (only with a real TTY; the daemon runs headless). @@ -254,6 +267,7 @@ export async function runAgy(opts: RunAgyOptions): Promise { backend.offMessage(onBackendMessage); await backend.dispose(); + happyServer.stop(); inkInstance?.unmount(); try {