From 1bc3267c6c7acdcf4bc1366bbe2658734a703649 Mon Sep 17 00:00:00 2001 From: Ame <123734885+luokerenx4@users.noreply.github.com> Date: Sat, 11 Jul 2026 15:33:15 +0800 Subject: [PATCH 1/5] feat(automation): normalize headless runtime output --- .../src/workspace-acceptance-smoke.spec.ts | 2 + .../desktop/src/workspace-acceptance-smoke.ts | 12 + docs/workspace-issues-and-scheduling.md | 42 +- src/webui/routes/headless.spec.ts | 44 +- src/webui/routes/headless.ts | 54 ++- src/webui/routes/workspaces.ts | 13 +- src/workspaces/adapters/claude.ts | 56 +++ src/workspaces/adapters/codex.ts | 61 +++ src/workspaces/adapters/opencode.ts | 65 +++ src/workspaces/adapters/pi.ts | 63 +++ .../agent-runtime-readiness.spec.ts | 7 + src/workspaces/cli-adapter.ts | 4 + src/workspaces/headless-output.spec.ts | 119 ++++++ src/workspaces/headless-output.ts | 182 +++++++++ src/workspaces/headless-task-registry.spec.ts | 21 + src/workspaces/headless-task-registry.ts | 27 +- src/workspaces/headless-task.spec.ts | 80 +++- src/workspaces/headless-task.ts | 140 ++++++- src/workspaces/service.ts | 43 +- ui/src/api/headless.ts | 45 +- ui/src/demo/handlers/headless.ts | 12 +- ui/src/pages/AutomationRunsSection.tsx | 384 ++++++++++++------ 22 files changed, 1283 insertions(+), 193 deletions(-) create mode 100644 src/workspaces/headless-output.spec.ts create mode 100644 src/workspaces/headless-output.ts diff --git a/apps/desktop/src/workspace-acceptance-smoke.spec.ts b/apps/desktop/src/workspace-acceptance-smoke.spec.ts index d5ed89164..a8b5a4a9c 100644 --- a/apps/desktop/src/workspace-acceptance-smoke.spec.ts +++ b/apps/desktop/src/workspace-acceptance-smoke.spec.ts @@ -14,6 +14,8 @@ describe('Workspace acceptance renderer source', () => { expect(source).toContain(`"printf '__OPENALICE_%s_OK__\\\\n' '${marker}'"`) } expect(source).toContain('__OPENALICE_WORKSPACE_%s_FAILED__ %s %s\\\\n%s\\\\n') + expect(source).toContain('managedPiStructuredOutput') + expect(source).toContain("block?.type === 'tool' && block?.status === 'completed'") expect(() => new Function(`return ${source}`)).not.toThrow() }) }) diff --git a/apps/desktop/src/workspace-acceptance-smoke.ts b/apps/desktop/src/workspace-acceptance-smoke.ts index 347d879b6..3fe2c596b 100644 --- a/apps/desktop/src/workspace-acceptance-smoke.ts +++ b/apps/desktop/src/workspace-acceptance-smoke.ts @@ -16,6 +16,7 @@ export interface WorkspaceAcceptanceReceipt { readonly allCliManifestsLoaded: boolean readonly shellCliRoundTrip: boolean readonly managedPiAssistantReply: boolean + readonly managedPiStructuredOutput: boolean readonly managedPiCliSideEffect: boolean readonly cleanupComplete: boolean } @@ -52,6 +53,7 @@ export async function runRendererWorkspaceAcceptanceSmoke( allCliManifestsLoaded: false, shellCliRoundTrip: false, managedPiAssistantReply: false, + managedPiStructuredOutput: false, managedPiCliSideEffect: false, cleanupComplete: false, } @@ -222,6 +224,16 @@ export async function runRendererWorkspaceAcceptanceSmoke( throw new Error('managed Pi assistant reply was not decoded: ' + JSON.stringify(headless.assistantText)) } checks.managedPiAssistantReply = true + if ( + headless.structured?.schemaVersion !== 1 || + headless.structured?.assistantText?.trim() !== '${ASSISTANT_TEXT}' || + typeof headless.structured?.metrics?.toolCalls !== 'number' || + headless.structured.metrics.toolCalls < 1 || + !headless.structured?.blocks?.some((block) => block?.type === 'tool' && block?.status === 'completed') + ) { + throw new Error('managed Pi structured output was not decoded: ' + JSON.stringify(headless.structured)) + } + checks.managedPiStructuredOutput = true const agentIssues = await json(await fetch('/api/issues')) if (!issueExists(agentIssues, workspaceId, agentIssueId)) { diff --git a/docs/workspace-issues-and-scheduling.md b/docs/workspace-issues-and-scheduling.md index a51a2c5d4..bad78919e 100644 --- a/docs/workspace-issues-and-scheduling.md +++ b/docs/workspace-issues-and-scheduling.md @@ -89,6 +89,7 @@ an attended, human-approved path and a commit in the peer repository. -> due calculation from `when` + last-fired marker -> headless run of the owning Workspace -> native agent CLI + -> normalized reply + message/tool blocks -> inbox_push when there is a user-visible result -> Inbox item linked to the run and issue ``` @@ -102,13 +103,36 @@ Schedule semantics remain in the issue file. Markers are written after a successful dispatch; capacity/transient rejection stays due for retry. Headless runs may overlap with interactive sessions or other runs in the same -checkout. Agents must tolerate concurrent edits. Global headless capacity is -bounded, but there is no per-Workspace exclusive lock. +checkout. Agents must tolerate concurrent edits. The launcher currently admits +at most eight headless processes globally and serializes registry persistence, +but there is no per-Workspace exclusive lock. + +## Structured Runtime Output + +Claude Code, Codex, opencode, and Pi all emit different JSON event streams. +Adapters translate those streams into one launcher-owned contract while the run +is active: + +- `assistantText` — the latest completed assistant reply; +- ordered `text`, `tool`, and `error` message blocks; +- tool name, input, output, and `running | completed | failed` status; +- compact metrics for reply presence, tool count, and tool failures. + +Automation reads a debounced `.structured.json` snapshot instead of replaying +an entire vendor log. This makes live polling cheap and gives future workbench +orchestration a stable contract independent of CLI versions. Runs created before +this contract are parsed best-effort from the last 2 MB of stdout when opened. + +Raw stdout/stderr remain available only as a diagnostic fallback. Each stream +is capped at 16 MB because Pi's JSON mode can emit cumulative `message_update` +frames; appending every cumulative frame without a cap can otherwise turn one +normal conversation into a multi-gigabyte log. Normalized output is separately +bounded to 300 blocks, 64 KB per text reply, and 8 KB per tool input/output. ## Delivery and Trading Safety -Headless stdout is diagnostic, not the user delivery channel. A run with a -meaningful result calls: +Structured headless output is the live control-plane result, while Inbox is the +durable user-delivery channel. A run with a meaningful report or artifact calls: ```bash alice-workspace inbox push --doc --comments "" @@ -140,6 +164,12 @@ human approval boundaries. | `src/workspaces/schedule/scanner.ts` | Workspace scan, due calculation, dispatch | | `src/workspaces/schedule/marker-store.ts` | Atomic last-fired persistence | | `src/workspaces/service.ts` | Scanner composition, agent resolution, headless registry | +| `src/workspaces/headless-task.ts` | Process lifecycle, bounded logs, live structured snapshots | +| `src/workspaces/headless-task-registry.ts` | Concurrent run records, capacity projection, and log pruning | +| `src/workspaces/headless-output.ts` | Vendor-neutral reply/tool block contract and accumulator | +| `src/workspaces/adapters/{claude,codex,opencode,pi}.ts` | Runtime-specific JSON event translation | +| `src/webui/routes/headless.ts` | Cross-workspace capacity, task, normalized output, and raw-tail API | +| `ui/src/pages/AutomationRunsSection.tsx` | Run list, final reply, tool activity, and diagnostics UI | | `src/tool/issue-tools.ts` | Workspace-scoped issue CLI/MCP tools | | `src/tool/inbox-push.ts` | Headless/interactive delivery to Inbox | | `src/workspaces/session-registry.ts` | Durable Session identity and run → Session source index | @@ -157,6 +187,10 @@ central schedule store or revive the legacy cron/AgentWork path. ```bash npx tsc --noEmit pnpm vitest run \ + src/workspaces/headless-output.spec.ts \ + src/workspaces/headless-task.spec.ts \ + src/workspaces/headless-task-registry.spec.ts \ + src/webui/routes/headless.spec.ts \ src/workspaces/issues/declaration.spec.ts \ src/workspaces/issues/mutate.spec.ts \ src/workspaces/issues/board.spec.ts \ diff --git a/src/webui/routes/headless.spec.ts b/src/webui/routes/headless.spec.ts index eb7c66c30..b1a8e35bb 100644 --- a/src/webui/routes/headless.spec.ts +++ b/src/webui/routes/headless.spec.ts @@ -1,6 +1,11 @@ +import { mkdtemp, rm, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' + import { describe, expect, it, vi } from 'vitest' import { createHeadlessRoutes } from './headless.js' +import { headlessLogPaths } from '../../workspaces/headless-task-registry.js' import type { WorkspaceService } from '../../workspaces/service.js' /* eslint-disable @typescript-eslint/no-explicit-any */ @@ -10,14 +15,20 @@ const TASKS = [ { taskId: 't2', wsId: 'w2', agent: 'pi', status: 'running', startedAt: 2 }, ] -function build() { +function build(logsDir = '/tmp/openalice-headless-route-test') { const list = vi.fn((opts: any = {}) => TASKS.filter( (t) => (!opts.wsId || t.wsId === opts.wsId) && (!opts.status || t.status === opts.status), ), ) const get = vi.fn((id: string) => TASKS.find((t) => t.taskId === id) ?? null) - const svc = { headlessTasks: { list, get } } as unknown as WorkspaceService + const runningCount = vi.fn(() => TASKS.filter((task) => task.status === 'running').length) + const svc = { + headlessTasks: { list, get, runningCount }, + headlessCapacity: 8, + headlessLogsDir: logsDir, + adapters: { get: vi.fn(() => null) }, + } as unknown as WorkspaceService return { app: createHeadlessRoutes(svc), list, get } } @@ -26,7 +37,9 @@ describe('GET /api/headless', () => { const { app } = build() const r = await app.request('/') expect(r.status).toBe(200) - expect(((await r.json()) as any).tasks.length).toBe(2) + const body = (await r.json()) as any + expect(body.tasks.length).toBe(2) + expect(body.capacity).toEqual({ running: 1, limit: 8 }) }) it('passes wsId/status/limit filters through to the registry', async () => { @@ -52,4 +65,29 @@ describe('GET /api/headless', () => { const { app } = build() expect((await app.request('/nope')).status).toBe(404) }) + + it('GET /:taskId/output returns the persisted normalized snapshot', async () => { + const dir = await mkdtemp(join(tmpdir(), 'headless-route-')) + try { + const paths = headlessLogPaths(dir, 't1') + await writeFile(paths.stdout, '{"vendor":"event"}\n') + await writeFile(paths.stderr, '') + await writeFile(paths.structured, JSON.stringify({ + schemaVersion: 1, + assistantText: 'Ready.', + blocks: [{ type: 'tool', id: 'tool-1', name: 'bash', status: 'completed', output: 'ok' }], + metrics: { textBlocks: 0, toolCalls: 1, toolFailures: 0 }, + truncated: false, + })) + const { app } = build(dir) + const response = await app.request('/t1/output') + expect(response.status).toBe(200) + const body = (await response.json()) as any + expect(body.structured.assistantText).toBe('Ready.') + expect(body.structured.metrics.toolCalls).toBe(1) + expect(body.stdout.text).toContain('vendor') + } finally { + await rm(dir, { recursive: true, force: true }) + } + }) }) diff --git a/src/webui/routes/headless.ts b/src/webui/routes/headless.ts index b7e2d2272..20be271fd 100644 --- a/src/webui/routes/headless.ts +++ b/src/webui/routes/headless.ts @@ -4,19 +4,24 @@ * Read-only view over `WorkspaceService.headlessTasks`: "what are the workers * doing" across every workspace. Dispatch lives at POST /api/workspaces/:id/ * headless (it's per-workspace); this surface is the panel + per-task status - * + the task's full output log (the run's own stdout/stderr on disk). + * + its normalized reply/tool timeline and size-capped raw diagnostic logs. */ -import { open, stat } from 'node:fs/promises' +import { open, readFile, stat } from 'node:fs/promises' import { Hono } from 'hono' import { headlessLogPaths, type HeadlessTaskStatus } from '../../workspaces/headless-task-registry.js' +import { + parseHeadlessOutputText, + type HeadlessStructuredOutput, +} from '../../workspaces/headless-output.js' import type { WorkspaceService } from '../../workspaces/service.js' const STATUSES = new Set(['running', 'done', 'failed', 'interrupted']) const DEFAULT_TAIL_BYTES = 64 * 1024 const MAX_TAIL_BYTES = 1024 * 1024 +const STRUCTURED_TAIL_BYTES = 2 * 1024 * 1024 /** Read the last `tailBytes` of a file; null when the file doesn't exist. */ async function readTail( @@ -40,6 +45,16 @@ async function readTail( } } +async function readStructured(path: string): Promise { + try { + const parsed = JSON.parse(await readFile(path, 'utf8')) as HeadlessStructuredOutput + if (parsed.schemaVersion !== 1 || !Array.isArray(parsed.blocks)) return null + return parsed + } catch { + return null + } +} + export function createHeadlessRoutes(svc: WorkspaceService): Hono { const app = new Hono() @@ -53,7 +68,13 @@ export function createHeadlessRoutes(svc: WorkspaceService): Hono { : undefined const limitRaw = Number(c.req.query('limit')) const limit = Number.isFinite(limitRaw) && limitRaw > 0 ? Math.min(limitRaw, 500) : 100 - return c.json({ tasks: svc.headlessTasks.list({ wsId, status, limit }) }) + return c.json({ + tasks: svc.headlessTasks.list({ wsId, status, limit }), + capacity: { + running: svc.headlessTasks.runningCount(), + limit: svc.headlessCapacity, + }, + }) }) // GET /api/headless/:taskId → one task's record. @@ -63,11 +84,10 @@ export function createHeadlessRoutes(svc: WorkspaceService): Hono { return c.json(rec) }) - // GET /api/headless/:taskId/output?tailBytes= → the task's on-disk log - // tails (stdout = the agent's structured event stream, stderr = CLI - // diagnostics). Tail-bounded so a chatty run can't flood the panel; the - // viewer polls this while the task runs. Streams are null when the log file - // doesn't exist (task predates log capture, or spawn failed before output). + // GET /api/headless/:taskId/output?tailBytes= → compact normalized output + + // raw diagnostic tails. New runs read their live structured snapshot; + // historical runs are parsed from a bounded stdout tail. Streams are null + // when the log file doesn't exist (old task, pruned log, or spawn failure). app.get('/:taskId/output', async (c) => { const taskId = c.req.param('taskId') const rec = svc.headlessTasks.get(taskId) @@ -76,11 +96,25 @@ export function createHeadlessRoutes(svc: WorkspaceService): Hono { const tailBytes = Number.isFinite(tailRaw) && tailRaw > 0 ? Math.min(tailRaw, MAX_TAIL_BYTES) : DEFAULT_TAIL_BYTES const paths = headlessLogPaths(svc.headlessLogsDir, taskId) - const [stdout, stderr] = await Promise.all([ + const [stdout, stderr, storedStructured] = await Promise.all([ readTail(paths.stdout, tailBytes), readTail(paths.stderr, tailBytes), + readStructured(paths.structured), ]) - return c.json({ taskId, status: rec.status, stdout, stderr }) + const adapter = svc.adapters.get(rec.agent) + const structuredSource = storedStructured ? null : await readTail(paths.stdout, STRUCTURED_TAIL_BYTES) + const structured = storedStructured ?? parseHeadlessOutputText({ + text: structuredSource?.text ?? '', + ...(adapter?.extractHeadlessOutputEvents + ? { extractEvents: adapter.extractHeadlessOutputEvents.bind(adapter) } + : {}), + ...(adapter?.extractHeadlessAssistantText + ? { extractAssistantText: adapter.extractHeadlessAssistantText.bind(adapter) } + : {}), + sourceTruncated: structuredSource?.truncated ?? false, + runStillActive: rec.status === 'running', + }) + return c.json({ taskId, status: rec.status, structured, stdout, stderr }) }) return app diff --git a/src/webui/routes/workspaces.ts b/src/webui/routes/workspaces.ts index 8797e9ae6..2b1cafd38 100644 --- a/src/webui/routes/workspaces.ts +++ b/src/webui/routes/workspaces.ts @@ -1255,10 +1255,11 @@ export function createWorkspaceRoutes( // Headless task dispatch — the standard automation API. Spawns the // workspace's agent CLI in one-shot headless mode with a positional prompt, - // runs to natural exit, returns exit/duration + bounded output tails. The - // agent reports its actual result via `inbox_push`; this endpoint just waits - // on the process exit (the turn boundary). No session/PTY — a fresh one-shot - // clone each call (no respawn, not pooled). Synchronous: the request stays + // runs to natural exit, returns exit/duration, a normalized reply/tool + // timeline, and bounded output tails. `inbox_push` remains the durable + // user-delivery channel; structured output powers readiness, Automation, and + // orchestration. No session/PTY — a fresh one-shot clone each call (no + // respawn, not pooled). Synchronous: the request stays // open until the task exits (the cron/automation trigger calls // `svc.runHeadlessTask` directly instead). Body: { prompt, agent?, timeoutMs? }. // curl -XPOST .../:id/headless -d '{"prompt":"...","agent":"claude"}' @@ -1341,8 +1342,8 @@ export function createWorkspaceRoutes( } } // Default → async: record + spawn in the background, return the taskId. The - // run's status is queryable at GET /api/headless/:taskId; the agent reports - // its actual result via the Inbox. + // run's status and normalized output are queryable under /api/headless; + // the agent can additionally publish durable user-facing work to Inbox. try { const { taskId } = await svc.dispatchHeadlessTask(meta, adapter, prompt, timeoutMs); return c.json({ taskId, status: 'running' }, 202); diff --git a/src/workspaces/adapters/claude.ts b/src/workspaces/adapters/claude.ts index ee5c46616..099d6d1ae 100644 --- a/src/workspaces/adapters/claude.ts +++ b/src/workspaces/adapters/claude.ts @@ -4,6 +4,7 @@ import { join, resolve } from 'node:path'; import type { CliAdapter, SpawnContext, WorkspaceAiCred } from '../cli-adapter.js'; import { readWorkspaceFile, writeWorkspaceFile } from '../file-service.js'; +import type { HeadlessOutputEvent } from '../headless-output.js'; const SESSION_FILE_RE = /^([0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})\.jsonl$/i; @@ -135,6 +136,61 @@ export const claudeAdapter: CliAdapter = { } }, + extractHeadlessOutputEvents(line: string): readonly HeadlessOutputEvent[] { + try { + const evt = JSON.parse(line) as Record; + const message = evt['message']; + if (message && typeof message === 'object') { + const record = message as Record; + const content = record['content']; + if (Array.isArray(content)) { + if (evt['type'] === 'assistant' && record['role'] === 'assistant') { + return content.flatMap((part): HeadlessOutputEvent[] => { + if (!part || typeof part !== 'object') return []; + const block = part as Record; + if (block['type'] === 'text' && typeof block['text'] === 'string') { + return [{ type: 'text', text: block['text'] }]; + } + if ( + block['type'] === 'tool_use' && + typeof block['id'] === 'string' && + typeof block['name'] === 'string' + ) { + return [{ + type: 'tool-start', + id: block['id'], + name: block['name'], + ...(block['input'] !== undefined ? { input: block['input'] } : {}), + }]; + } + return []; + }); + } + if (evt['type'] === 'user' && record['role'] === 'user') { + return content.flatMap((part): HeadlessOutputEvent[] => { + if (!part || typeof part !== 'object') return []; + const block = part as Record; + if (block['type'] !== 'tool_result' || typeof block['tool_use_id'] !== 'string') return []; + return [{ + type: 'tool-finish', + id: block['tool_use_id'], + ...(block['content'] !== undefined ? { output: block['content'] } : {}), + ...(block['is_error'] === true ? { isError: true } : {}), + }]; + }); + } + } + } + if (evt['type'] === 'result' && evt['is_error'] === true) { + const result = evt['result']; + return [{ type: 'error', message: typeof result === 'string' ? result : 'Claude run failed' }]; + } + return []; + } catch { + return []; + } + }, + async writeAiConfig(cwd: string, cred: WorkspaceAiCred): Promise { const hasAny = cred.baseUrl || cred.apiKey || cred.model; if (!hasAny) { diff --git a/src/workspaces/adapters/codex.ts b/src/workspaces/adapters/codex.ts index d0de8639d..35f271a41 100644 --- a/src/workspaces/adapters/codex.ts +++ b/src/workspaces/adapters/codex.ts @@ -6,6 +6,7 @@ import { createInterface } from 'node:readline'; import type { BootstrapContext, CliAdapter, OnDiskSession, SpawnContext, WorkspaceAiCred } from '../cli-adapter.js'; import { readWorkspaceFile, writeWorkspaceFile } from '../file-service.js'; +import type { HeadlessOutputEvent } from '../headless-output.js'; const CODEX_CONFIG_PATH = '.codex/config.toml'; const CODEX_ENV_PATH = '.codex/env.json'; @@ -147,6 +148,66 @@ export const codexAdapter: CliAdapter = { } }, + extractHeadlessOutputEvents(line: string): readonly HeadlessOutputEvent[] { + try { + const evt = JSON.parse(line) as Record; + if (evt['type'] === 'turn.failed') { + const error = evt['error']; + const message = error && typeof error === 'object' && typeof (error as Record)['message'] === 'string' + ? (error as Record)['message'] as string + : typeof error === 'string' + ? error + : 'Codex turn failed'; + return [{ type: 'error', message }]; + } + if (evt['type'] !== 'item.started' && evt['type'] !== 'item.completed') return []; + const item = evt['item']; + if (!item || typeof item !== 'object') return []; + const record = item as Record; + const id = typeof record['id'] === 'string' ? record['id'] : `codex-${record['type'] ?? 'item'}`; + if (evt['type'] === 'item.completed' && record['type'] === 'agent_message' && typeof record['text'] === 'string') { + return [{ type: 'text', text: record['text'] }]; + } + if (record['type'] === 'command_execution') { + const input = typeof record['command'] === 'string' ? { command: record['command'] } : record['command']; + if (evt['type'] === 'item.started') return [{ type: 'tool-start', id, name: 'Shell', input }]; + return [{ + type: 'tool-finish', + id, + name: 'Shell', + ...(record['aggregated_output'] !== undefined ? { output: record['aggregated_output'] } : {}), + ...(typeof record['exit_code'] === 'number' && record['exit_code'] !== 0 ? { isError: true } : {}), + }]; + } + if (record['type'] === 'file_change') { + if (evt['type'] === 'item.started') { + return [{ type: 'tool-start', id, name: 'File changes', input: record['changes'] }]; + } + return [{ type: 'tool-finish', id, name: 'File changes', output: record['changes'] }]; + } + if (record['type'] === 'mcp_tool_call' || record['type'] === 'tool_call') { + const name = typeof record['tool'] === 'string' + ? record['tool'] + : typeof record['name'] === 'string' + ? record['name'] + : 'Tool'; + if (evt['type'] === 'item.started') { + return [{ type: 'tool-start', id, name, input: record['arguments'] ?? record['input'] }]; + } + return [{ + type: 'tool-finish', + id, + name, + output: record['result'] ?? record['output'], + ...(record['status'] === 'failed' ? { isError: true } : {}), + }]; + } + return []; + } catch { + return []; + } + }, + async writeAiConfig(cwd: string, cred: WorkspaceAiCred): Promise { const hasProvider = !!(cred.baseUrl || cred.model); diff --git a/src/workspaces/adapters/opencode.ts b/src/workspaces/adapters/opencode.ts index 51e55a32f..b41fdba65 100644 --- a/src/workspaces/adapters/opencode.ts +++ b/src/workspaces/adapters/opencode.ts @@ -5,6 +5,7 @@ import { promisify } from 'node:util'; import type { CliAdapter, OnDiskSession, SpawnContext, WorkspaceAiCred } from '../cli-adapter.js'; import { readWorkspaceFile, writeWorkspaceFile } from '../file-service.js'; +import type { HeadlessOutputEvent } from '../headless-output.js'; const execFileAsync = promisify(execFile); @@ -137,6 +138,70 @@ export const opencodeAdapter: CliAdapter = { } }, + extractHeadlessOutputEvents(line: string): readonly HeadlessOutputEvent[] { + try { + const evt = JSON.parse(line) as Record; + if (evt['type'] === 'error') { + const error = evt['error']; + const record = error && typeof error === 'object' ? error as Record : null; + const data = record?.['data'] && typeof record['data'] === 'object' + ? record['data'] as Record + : null; + const message = typeof data?.['message'] === 'string' + ? data['message'] + : typeof record?.['message'] === 'string' + ? record['message'] + : typeof record?.['name'] === 'string' + ? record['name'] + : typeof error === 'string' + ? error + : 'OpenCode session failed'; + return [{ type: 'error', message }]; + } + const part = evt['part']; + if (!part || typeof part !== 'object') return []; + const record = part as Record; + if (evt['type'] === 'text' && record['type'] === 'text' && typeof record['text'] === 'string') { + return [{ type: 'text', text: record['text'] }]; + } + if (evt['type'] !== 'tool_use' && record['type'] !== 'tool') return []; + const state = record['state'] && typeof record['state'] === 'object' + ? record['state'] as Record + : {}; + const id = typeof record['callID'] === 'string' + ? record['callID'] + : typeof record['id'] === 'string' + ? record['id'] + : `opencode-${record['tool'] ?? 'tool'}`; + const name = typeof record['tool'] === 'string' + ? record['tool'] + : typeof record['name'] === 'string' + ? record['name'] + : 'Tool'; + const start: HeadlessOutputEvent = { + type: 'tool-start', + id, + name, + ...(state['input'] !== undefined || record['input'] !== undefined + ? { input: state['input'] ?? record['input'] } + : {}), + }; + const status = state['status']; + if (status !== 'completed' && status !== 'error' && status !== 'failed') return [start]; + return [start, { + type: 'tool-finish', + id, + name, + ...(state['output'] !== undefined || state['error'] !== undefined + ? { output: state['output'] ?? state['error'] } + : {}), + ...(status === 'error' || status === 'failed' ? { isError: true } : {}), + }]; + } catch { + return []; + } + }, + composeEnv(ctx: SpawnContext): Record { const env: Record = { OPENCODE_DISABLE_MODELS_FETCH: '1', diff --git a/src/workspaces/adapters/pi.ts b/src/workspaces/adapters/pi.ts index 802f750f3..871ac0396 100644 --- a/src/workspaces/adapters/pi.ts +++ b/src/workspaces/adapters/pi.ts @@ -7,6 +7,7 @@ import { resolveBashPath } from '@/core/shell-resolver.js'; import type { CliAdapter, SpawnContext, WorkspaceAiCred } from '../cli-adapter.js'; import { readWorkspaceFile, writeWorkspaceFile } from '../file-service.js'; +import type { HeadlessOutputEvent } from '../headless-output.js'; // Pi's per-workspace provider override. `models.json` is read from Pi's AGENT // DIR, which has NO project-local layer — so we redirect the whole agent dir to @@ -186,6 +187,9 @@ export const piAdapter: CliAdapter = { }, extractHeadlessAssistantText(line: string): string | null { + // Pi's message_update frames contain cumulative content and dominate large + // runs. JSON mode uses JSON.stringify, so cheaply reject them before parse. + if (!line.startsWith('{"type":"message_end"')) return null; try { const evt = JSON.parse(line) as Record; if (evt['type'] !== 'message_end') return null; @@ -208,6 +212,65 @@ export const piAdapter: CliAdapter = { } }, + extractHeadlessOutputEvents(line: string): readonly HeadlessOutputEvent[] { + if ( + !line.startsWith('{"type":"tool_execution_start"') && + !line.startsWith('{"type":"tool_execution_end"') && + !line.startsWith('{"type":"message_end"') + ) return []; + try { + const evt = JSON.parse(line) as Record; + if ( + evt['type'] === 'tool_execution_start' && + typeof evt['toolCallId'] === 'string' && + typeof evt['toolName'] === 'string' + ) { + return [{ + type: 'tool-start', + id: evt['toolCallId'], + name: evt['toolName'], + ...(evt['args'] !== undefined ? { input: evt['args'] } : {}), + }]; + } + if ( + evt['type'] === 'tool_execution_end' && + typeof evt['toolCallId'] === 'string' + ) { + return [{ + type: 'tool-finish', + id: evt['toolCallId'], + ...(typeof evt['toolName'] === 'string' ? { name: evt['toolName'] } : {}), + ...(evt['result'] !== undefined ? { output: evt['result'] } : {}), + ...(evt['isError'] === true ? { isError: true } : {}), + }]; + } + if (evt['type'] !== 'message_end') return []; + const message = evt['message']; + if (!message || typeof message !== 'object') return []; + const record = message as Record; + if (record['role'] !== 'assistant' || !Array.isArray(record['content'])) return []; + const events: HeadlessOutputEvent[] = []; + if (record['stopReason'] === 'error' || record['stopReason'] === 'aborted') { + events.push({ + type: 'error', + message: typeof record['errorMessage'] === 'string' + ? record['errorMessage'] + : `Pi request ${record['stopReason']}`, + }); + } + events.push(...record['content'].flatMap((part): HeadlessOutputEvent[] => { + if (!part || typeof part !== 'object') return []; + const content = part as Record; + return content['type'] === 'text' && typeof content['text'] === 'string' + ? [{ type: 'text', text: content['text'] }] + : []; + })); + return events; + } catch { + return []; + } + }, + composeEnv(ctx: SpawnContext): Record { // Do not force PI_OFFLINE. OpenAlice is a networked product and Pi may // download missing runtime tools during startup. A user or launcher can diff --git a/src/workspaces/agent-runtime-readiness.spec.ts b/src/workspaces/agent-runtime-readiness.spec.ts index e0ea7b7cf..dbd1bb969 100644 --- a/src/workspaces/agent-runtime-readiness.spec.ts +++ b/src/workspaces/agent-runtime-readiness.spec.ts @@ -37,6 +37,13 @@ function result(overrides: Partial): HeadlessTaskResult { stderrTail: '', agentSessionId: null, assistantText: null, + structured: { + schemaVersion: 1, + assistantText: null, + blocks: [], + metrics: { textBlocks: 0, toolCalls: 0, toolFailures: 0 }, + truncated: false, + }, ...overrides, }; } diff --git a/src/workspaces/cli-adapter.ts b/src/workspaces/cli-adapter.ts index 58652d168..83f446ead 100644 --- a/src/workspaces/cli-adapter.ts +++ b/src/workspaces/cli-adapter.ts @@ -12,6 +12,7 @@ */ import type { WireShape } from '../ai-providers/preset-catalog.js'; +import type { HeadlessOutputEvent } from './headless-output.js'; export interface OnDiskSession { readonly sessionId: string; @@ -210,6 +211,9 @@ export interface CliAdapter { */ extractHeadlessAssistantText?(line: string): string | null; + /** Translate one native JSONL line into vendor-neutral response/tool events. */ + extractHeadlessOutputEvents?(line: string): readonly HeadlessOutputEvent[]; + /** Optional per-CLI env adjustments on top of `spawn-env.ts`'s baseline. */ envOverrides?(parent: NodeJS.ProcessEnv): EnvOverrides; diff --git a/src/workspaces/headless-output.spec.ts b/src/workspaces/headless-output.spec.ts new file mode 100644 index 000000000..406aedc94 --- /dev/null +++ b/src/workspaces/headless-output.spec.ts @@ -0,0 +1,119 @@ +import { describe, expect, it } from 'vitest' + +import { claudeAdapter } from './adapters/claude.js' +import { codexAdapter } from './adapters/codex.js' +import { opencodeAdapter } from './adapters/opencode.js' +import { piAdapter } from './adapters/pi.js' +import type { CliAdapter } from './cli-adapter.js' +import { parseHeadlessOutputText } from './headless-output.js' + +function parse( + adapter: CliAdapter, + events: readonly Record[], + sourceTruncated = false, +) { + return parseHeadlessOutputText({ + text: events.map((event) => JSON.stringify(event)).join('\n'), + extractEvents: adapter.extractHeadlessOutputEvents?.bind(adapter), + extractAssistantText: adapter.extractHeadlessAssistantText?.bind(adapter), + sourceTruncated, + }) +} + +describe('headless structured output', () => { + it('pairs Claude tool_use/tool_result and keeps the final reply', () => { + const output = parse(claudeAdapter, [ + { + type: 'assistant', + message: { role: 'assistant', content: [{ type: 'tool_use', id: 't1', name: 'Bash', input: { command: 'alice status' } }] }, + }, + { + type: 'user', + message: { role: 'user', content: [{ type: 'tool_result', tool_use_id: 't1', content: 'ok' }] }, + }, + { type: 'assistant', message: { role: 'assistant', content: [{ type: 'text', text: 'Everything is ready.' }] } }, + { type: 'result', subtype: 'success', result: 'Everything is ready.' }, + ]) + expect(output.assistantText).toBe('Everything is ready.') + expect(output.blocks).toContainEqual({ + type: 'tool', id: 't1', name: 'Bash', status: 'completed', input: { command: 'alice status' }, output: 'ok', + }) + expect(output.metrics).toEqual({ textBlocks: 1, toolCalls: 1, toolFailures: 0 }) + }) + + it('normalizes Codex command execution and file-change items', () => { + const output = parse(codexAdapter, [ + { type: 'item.started', item: { id: 'c1', type: 'command_execution', command: 'alice status', status: 'in_progress' } }, + { type: 'item.completed', item: { id: 'c1', type: 'command_execution', command: 'alice status', aggregated_output: 'ok', exit_code: 0, status: 'completed' } }, + { type: 'item.started', item: { id: 'f1', type: 'file_change', changes: [{ path: 'report.md' }], status: 'in_progress' } }, + { type: 'item.completed', item: { id: 'f1', type: 'file_change', changes: [{ path: 'report.md' }], status: 'completed' } }, + { type: 'item.completed', item: { id: 'm1', type: 'agent_message', text: 'Report written.' } }, + ]) + expect(output.assistantText).toBe('Report written.') + expect(output.metrics.toolCalls).toBe(2) + expect(output.blocks).toContainEqual(expect.objectContaining({ type: 'tool', id: 'c1', name: 'Shell', status: 'completed' })) + expect(output.blocks).toContainEqual(expect.objectContaining({ type: 'tool', id: 'f1', name: 'File changes', status: 'completed' })) + }) + + it('keeps runtime-level failures as error blocks', () => { + const codex = parse(codexAdapter, [ + { type: 'turn.failed', error: { message: 'Codex provider unavailable' } }, + ]) + const opencode = parse(opencodeAdapter, [ + { type: 'error', error: { name: 'APIError', data: { message: 'OpenCode provider unavailable' } } }, + ]) + const pi = parse(piAdapter, [ + { + type: 'message_end', + message: { + role: 'assistant', stopReason: 'error', errorMessage: 'Pi provider unavailable', content: [], + }, + }, + ]) + expect(codex.blocks).toContainEqual({ type: 'error', message: 'Codex provider unavailable' }) + expect(opencode.blocks).toContainEqual({ type: 'error', message: 'OpenCode provider unavailable' }) + expect(pi.blocks).toContainEqual({ type: 'error', message: 'Pi provider unavailable' }) + }) + + it('normalizes Pi execution events and failed tools', () => { + const output = parse(piAdapter, [ + { type: 'tool_execution_start', toolCallId: 'p1', toolName: 'bash', args: { command: 'false' } }, + { type: 'tool_execution_end', toolCallId: 'p1', toolName: 'bash', result: { content: 'failed' }, isError: true }, + { type: 'message_end', message: { role: 'assistant', content: [{ type: 'text', text: 'The command failed.' }] } }, + ]) + expect(output.assistantText).toBe('The command failed.') + expect(output.metrics).toEqual({ textBlocks: 1, toolCalls: 1, toolFailures: 1 }) + }) + + it('skips Pi cumulative message_update frames before JSON parsing', () => { + const cumulativeFrame = '{"type":"message_update", deliberately-not-valid-json' + expect(piAdapter.extractHeadlessOutputEvents?.(cumulativeFrame)).toEqual([]) + expect(piAdapter.extractHeadlessAssistantText?.(cumulativeFrame)).toBeNull() + }) + + it('accepts OpenCode part.state tool snapshots', () => { + const output = parse(opencodeAdapter, [ + { + type: 'tool_use', + sessionID: 'ses_1', + part: { + type: 'tool', callID: 'o1', tool: 'bash', + state: { status: 'completed', input: { command: 'alice status' }, output: 'ok' }, + }, + }, + { type: 'text', sessionID: 'ses_1', part: { type: 'text', text: 'Done.' } }, + ]) + expect(output.assistantText).toBe('Done.') + expect(output.blocks).toContainEqual({ + type: 'tool', id: 'o1', name: 'bash', status: 'completed', input: { command: 'alice status' }, output: 'ok', + }) + }) + + it('marks unfinished tools failed after a finished/truncated run', () => { + const output = parse(piAdapter, [ + { type: 'tool_execution_start', toolCallId: 'p1', toolName: 'bash', args: {} }, + ], true) + expect(output.blocks).toContainEqual(expect.objectContaining({ id: 'p1', status: 'failed' })) + expect(output.truncated).toBe(true) + }) +}) diff --git a/src/workspaces/headless-output.ts b/src/workspaces/headless-output.ts new file mode 100644 index 000000000..5f4e399ae --- /dev/null +++ b/src/workspaces/headless-output.ts @@ -0,0 +1,182 @@ +/** Vendor-neutral structured output for one-shot agent runs. */ + +export type HeadlessOutputEvent = + | { readonly type: 'text'; readonly text: string } + | { readonly type: 'tool-start'; readonly id: string; readonly name: string; readonly input?: unknown } + | { + readonly type: 'tool-finish' + readonly id: string + readonly name?: string + readonly output?: unknown + readonly isError?: boolean + } + | { readonly type: 'error'; readonly message: string } + +export type HeadlessToolStatus = 'running' | 'completed' | 'failed' + +export type HeadlessMessageBlock = + | { readonly type: 'text'; readonly text: string } + | { + readonly type: 'tool' + readonly id: string + readonly name: string + readonly status: HeadlessToolStatus + readonly input?: unknown + readonly output?: unknown + } + | { readonly type: 'error'; readonly message: string } + +export interface HeadlessStructuredOutput { + readonly schemaVersion: 1 + readonly assistantText: string | null + readonly blocks: readonly HeadlessMessageBlock[] + readonly metrics: { + readonly textBlocks: number + readonly toolCalls: number + readonly toolFailures: number + } + /** True when the source log or normalized block budget was truncated. */ + readonly truncated: boolean +} + +const MAX_BLOCKS = 300 +const MAX_TEXT_CHARS = 64 * 1024 +const MAX_TOOL_VALUE_CHARS = 8 * 1024 + +function clipText(value: string, max = MAX_TEXT_CHARS): string { + return value.length <= max ? value : `${value.slice(0, max)}\n… (truncated)` +} + +function boundedValue(value: unknown): unknown { + if (value === undefined) return undefined + if (typeof value === 'string') return clipText(value, MAX_TOOL_VALUE_CHARS) + try { + const json = JSON.stringify(value) + if (json.length <= MAX_TOOL_VALUE_CHARS) return value + return `${json.slice(0, MAX_TOOL_VALUE_CHARS)}… (truncated)` + } catch { + return clipText(String(value), MAX_TOOL_VALUE_CHARS) + } +} + +/** Pairs tool start/end events and keeps one final assistant reply projection. */ +export class HeadlessOutputAccumulator { + private readonly blocks: HeadlessMessageBlock[] = [] + private readonly toolIndexes = new Map() + private assistantText: string | null = null + private sourceTruncated = false + private blockTruncated = false + + add(events: readonly HeadlessOutputEvent[]): void { + for (const event of events) { + if (event.type === 'text') { + const text = clipText(event.text.trim()) + if (!text) continue + this.assistantText = text + const previous = this.blocks[this.blocks.length - 1] + if (previous?.type === 'text' && previous.text === text) continue + this.push({ type: 'text', text }) + continue + } + + if (event.type === 'error') { + const message = clipText(event.message.trim(), MAX_TOOL_VALUE_CHARS) + if (message) this.push({ type: 'error', message }) + continue + } + + if (event.type === 'tool-start') { + const index = this.toolIndexes.get(event.id) + const next: Extract = { + type: 'tool', + id: event.id, + name: event.name || 'Tool', + status: 'running', + ...(event.input !== undefined ? { input: boundedValue(event.input) } : {}), + } + if (index === undefined) { + if (this.push(next)) this.toolIndexes.set(event.id, this.blocks.length - 1) + } else { + const current = this.blocks[index] + this.blocks[index] = current?.type === 'tool' ? { ...current, ...next } : next + } + continue + } + + const index = this.toolIndexes.get(event.id) + const current = index === undefined ? null : this.blocks[index] + const finished: Extract = { + type: 'tool', + id: event.id, + name: event.name || (current?.type === 'tool' ? current.name : 'Tool'), + status: event.isError ? 'failed' : 'completed', + ...(current?.type === 'tool' && current.input !== undefined ? { input: current.input } : {}), + ...(event.output !== undefined ? { output: boundedValue(event.output) } : {}), + } + if (index === undefined) { + if (this.push(finished)) this.toolIndexes.set(event.id, this.blocks.length - 1) + } else { + this.blocks[index] = finished + } + } + } + + /** Adapter text decoders remain a compatibility/final-result fallback. */ + setAssistantText(text: string): void { + const normalized = clipText(text.trim()) + if (normalized) this.assistantText = normalized + } + + markSourceTruncated(): void { + this.sourceTruncated = true + } + + snapshot(runStillActive = false): HeadlessStructuredOutput { + const blocks = this.blocks.map((block) => { + if (block.type !== 'tool' || block.status !== 'running' || runStillActive) return block + return { ...block, status: 'failed' as const } + }) + const tools = blocks.filter( + (block): block is Extract => block.type === 'tool', + ) + return { + schemaVersion: 1, + assistantText: this.assistantText, + blocks, + metrics: { + textBlocks: blocks.reduce((count, block) => count + (block.type === 'text' ? 1 : 0), 0), + toolCalls: tools.length, + toolFailures: tools.reduce((count, block) => count + (block.status === 'failed' ? 1 : 0), 0), + }, + truncated: this.sourceTruncated || this.blockTruncated, + } + } + + private push(block: HeadlessMessageBlock): boolean { + if (this.blocks.length >= MAX_BLOCKS) { + this.blockTruncated = true + return false + } + this.blocks.push(block) + return true + } +} + +export function parseHeadlessOutputText(opts: { + readonly text: string + readonly extractEvents?: (line: string) => readonly HeadlessOutputEvent[] + readonly extractAssistantText?: (line: string) => string | null + readonly sourceTruncated?: boolean + readonly runStillActive?: boolean +}): HeadlessStructuredOutput { + const accumulator = new HeadlessOutputAccumulator() + if (opts.sourceTruncated) accumulator.markSourceTruncated() + for (const raw of opts.text.split(/\r?\n/)) { + const line = raw.trim() + if (!line) continue + accumulator.add(opts.extractEvents?.(line) ?? []) + const assistant = opts.extractAssistantText?.(line)?.trim() + if (assistant) accumulator.setAssistantText(assistant) + } + return accumulator.snapshot(opts.runStillActive ?? false) +} diff --git a/src/workspaces/headless-task-registry.spec.ts b/src/workspaces/headless-task-registry.spec.ts index fd661fe53..5858e9d11 100644 --- a/src/workspaces/headless-task-registry.spec.ts +++ b/src/workspaces/headless-task-registry.spec.ts @@ -93,6 +93,25 @@ describe('HeadlessTaskRegistry', () => { expect(reg2.get(a.taskId)?.status).toBe('done') }) + it('serializes concurrent registry writes without losing records', async () => { + const reg = await HeadlessTaskRegistry.load(path, noopLogger) + const created = await Promise.all( + Array.from({ length: 24 }, (_, index) => reg.create({ + wsId: `w${index % 3}`, + agent: index % 2 ? 'pi' : 'codex', + prompt: `task ${index}`, + startedAt: index, + })), + ) + await Promise.all(created.map((task, index) => reg.complete(task.taskId, { + status: 'done', + output: { hasAssistantReply: true, assistantPreview: `reply ${index}`, blockCount: 2, toolCalls: 1, toolFailures: 0 }, + }))) + const reloaded = await HeadlessTaskRegistry.load(path, noopLogger) + expect(reloaded.list()).toHaveLength(24) + expect(reloaded.list().every((task) => task.status === 'done' && task.output?.toolCalls === 1)).toBe(true) + }) + it('reconcile-on-boot flips a leftover running task → interrupted', async () => { const reg = await HeadlessTaskRegistry.load(path, noopLogger) await reg.create({ wsId: 'w1', agent: 'codex', prompt: 'x', startedAt: 1 }) // stays running @@ -119,6 +138,7 @@ describe('HeadlessTaskRegistry', () => { const firstLogs = headlessLogPaths(logsDir, first.taskId) await writeFile(firstLogs.stdout, 'old stdout') await writeFile(firstLogs.stderr, 'old stderr') + await writeFile(firstLogs.structured, '{}') // Fill past MAX_RECORDS (200) so `first` (oldest finished) gets pruned. for (let i = 0; i < 200; i++) { const t = await reg.create({ wsId: 'w1', agent: 'codex', prompt: `t${i}`, startedAt: 2 + i }) @@ -129,5 +149,6 @@ describe('HeadlessTaskRegistry', () => { await new Promise((r) => setTimeout(r, 50)) expect(existsSync(firstLogs.stdout)).toBe(false) expect(existsSync(firstLogs.stderr)).toBe(false) + expect(existsSync(firstLogs.structured)).toBe(false) }) }) diff --git a/src/workspaces/headless-task-registry.ts b/src/workspaces/headless-task-registry.ts index 80e966652..22278d625 100644 --- a/src/workspaces/headless-task-registry.ts +++ b/src/workspaces/headless-task-registry.ts @@ -21,6 +21,14 @@ import type { Logger } from './logger.js' export type HeadlessTaskStatus = 'running' | 'done' | 'failed' | 'interrupted' +export interface HeadlessTaskOutputSummary { + readonly hasAssistantReply: boolean + readonly assistantPreview?: string + readonly blockCount: number + readonly toolCalls: number + readonly toolFailures: number +} + export interface HeadlessTaskRecord { readonly taskId: string readonly wsId: string @@ -51,13 +59,19 @@ export interface HeadlessTaskRecord { * announcing (spawn failure) or predate the field. */ agentSessionId?: string + /** Compact list-view projection; full normalized blocks stay in the log API. */ + output?: HeadlessTaskOutputSummary } /** Task-log file paths — shared by the writer (service) and reader (route). */ -export function headlessLogPaths(logsDir: string, taskId: string): { stdout: string; stderr: string } { +export function headlessLogPaths( + logsDir: string, + taskId: string, +): { stdout: string; stderr: string; structured: string } { return { stdout: join(logsDir, `${taskId}.stdout.log`), stderr: join(logsDir, `${taskId}.stderr.log`), + structured: join(logsDir, `${taskId}.structured.json`), } } @@ -65,6 +79,8 @@ const MAX_RECORDS = 200 // prune oldest FINISHED records past this (bounds the f export class HeadlessTaskRegistry { private tasks: HeadlessTaskRecord[] = [] // newest-last in memory + /** Mutations may finish concurrently; serialize tmp→rename writes. */ + private flushChain: Promise = Promise.resolve() private constructor( private readonly path: string, @@ -136,7 +152,7 @@ export class HeadlessTaskRegistry { patch: Partial< Pick< HeadlessTaskRecord, - 'status' | 'finishedAt' | 'durationMs' | 'exitCode' | 'signal' | 'killed' | 'error' + 'status' | 'finishedAt' | 'durationMs' | 'exitCode' | 'signal' | 'killed' | 'error' | 'output' > >, ): Promise { @@ -177,6 +193,12 @@ export class HeadlessTaskRegistry { } private async flush(): Promise { + const next = this.flushChain.then(() => this.flushNow()) + this.flushChain = next.catch(() => undefined) + await next + } + + private async flushNow(): Promise { if (this.tasks.length > MAX_RECORDS) { // Drop the OLDEST finished records; never drop a `running` one. const dropCount = this.tasks.length - MAX_RECORDS @@ -194,6 +216,7 @@ export class HeadlessTaskRegistry { const paths = headlessLogPaths(this.logsDir, taskId) void rm(paths.stdout, { force: true }).catch(() => undefined) void rm(paths.stderr, { force: true }).catch(() => undefined) + void rm(paths.structured, { force: true }).catch(() => undefined) } } } diff --git a/src/workspaces/headless-task.spec.ts b/src/workspaces/headless-task.spec.ts index 3bb8e4399..a5c08667e 100644 --- a/src/workspaces/headless-task.spec.ts +++ b/src/workspaces/headless-task.spec.ts @@ -1,3 +1,7 @@ +import { mkdtemp, readFile, rm, stat } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; + import { describe, expect, it } from 'vitest'; import { runHeadlessTask } from './headless-task.js'; @@ -120,23 +124,25 @@ describe('runHeadlessTask', () => { timeoutMs: 5_000, logger: noopLogger, extractAssistantText: (line) => { + throw new Error(`structured translator should own text parsing: ${line}`); + }, + extractOutputEvents: (line) => { try { const evt = JSON.parse(line) as Record; return evt['type'] === 'assistant' && typeof evt['text'] === 'string' - ? evt['text'] - : null; + ? [{ type: 'text' as const, text: evt['text'] }] + : []; } catch { - return null; + return []; } }, }); expect(r.assistantText).toBe('Hello 👋'); + expect(r.structured.assistantText).toBe('Hello 👋'); + expect(r.structured.blocks).toHaveLength(2); }); - it('streams the FULL stdout/stderr to log files (beyond the 16KB tails)', async () => { - const { mkdtemp, readFile, rm } = await import('node:fs/promises'); - const { tmpdir } = await import('node:os'); - const { join } = await import('node:path'); + it('streams raw stdout/stderr to log files beyond the 16KB in-memory tails', async () => { const dir = await mkdtemp(join(tmpdir(), 'headless-log-')); try { // 64KB of stdout — far past the 16KB tail budget. @@ -155,7 +161,8 @@ describe('runHeadlessTask', () => { }); expect(r.exitCode).toBe(0); expect(r.stdoutTail.length).toBeLessThanOrEqual(16 * 1024); // tail stays bounded - // The log file has everything. The write stream is end()ed at exit but + // This output is below the 16MB raw cap, so the log file has everything. + // The write stream is end()ed at exit but // not awaited; poll briefly for the flush. let full = ''; for (let i = 0; i < 40 && full.length < 64 * 1024; i++) { @@ -168,4 +175,61 @@ describe('runHeadlessTask', () => { await rm(dir, { recursive: true, force: true }); } }); + + it('writes a compact structured snapshot for live Automation polling', async () => { + const dir = await mkdtemp(join(tmpdir(), 'headless-structured-')); + try { + const structuredFile = join(dir, 't1.structured.json'); + const event = JSON.stringify({ type: 'assistant', text: 'Snapshot reply' }); + const result = await runHeadlessTask({ + command: ['node', '-e', `process.stdout.write(${JSON.stringify(`${event}\n`)})`], + cwd: process.cwd(), + env: baseEnv, + timeoutMs: 5_000, + logger: noopLogger, + structuredFile, + extractAssistantText: (line) => { + const parsed = JSON.parse(line) as { type?: string; text?: string }; + return parsed.type === 'assistant' ? parsed.text ?? null : null; + }, + extractOutputEvents: (line) => { + const parsed = JSON.parse(line) as { type?: string; text?: string }; + return parsed.type === 'assistant' && parsed.text + ? [{ type: 'text' as const, text: parsed.text }] + : []; + }, + }); + const stored = JSON.parse(await readFile(structuredFile, 'utf8')) as typeof result.structured; + expect(stored).toEqual(result.structured); + expect(stored.assistantText).toBe('Snapshot reply'); + } finally { + await rm(dir, { recursive: true, force: true }); + } + }); + + it('caps each raw diagnostic stream at 16MB', async () => { + const dir = await mkdtemp(join(tmpdir(), 'headless-log-cap-')); + try { + const stdoutFile = join(dir, 't1.stdout.log'); + const result = await runHeadlessTask({ + command: ['node', '-e', 'process.stdout.write(Buffer.alloc(17 * 1024 * 1024, 83))'], + cwd: process.cwd(), + env: baseEnv, + timeoutMs: 10_000, + logger: noopLogger, + stdoutFile, + }); + expect(result.exitCode).toBe(0); + let size = 0; + for (let i = 0; i < 80 && size < 16 * 1024 * 1024; i++) { + await new Promise((resolve) => setTimeout(resolve, 25)); + size = (await stat(stdoutFile).catch(() => ({ size: 0 }))).size; + } + const stored = await readFile(stdoutFile, 'utf8'); + expect(size).toBeLessThan(16 * 1024 * 1024 + 256); + expect(stored).toContain('raw log capped at 16MB'); + } finally { + await rm(dir, { recursive: true, force: true }); + } + }); }); diff --git a/src/workspaces/headless-task.ts b/src/workspaces/headless-task.ts index 340d04259..0aa5cbc43 100644 --- a/src/workspaces/headless-task.ts +++ b/src/workspaces/headless-task.ts @@ -17,21 +17,29 @@ * - **NOT routed through SessionPool/PersistentSession**, whose respawn-on-exit * circuit is anti-semantic for a one-shot task (exit == completion). * - * The launcher does NOT parse the output: the agent reports via `inbox_push`. - * We only need the exit signal + a bounded output tail for diagnostics. + * The launcher normalizes each runtime's JSON event stream into a compact, + * vendor-neutral reply/tool timeline while preserving size-capped raw logs for + * diagnostics. `inbox_push` remains the durable user-delivery channel. */ -import { createWriteStream, type WriteStream } from 'node:fs'; -import { mkdir } from 'node:fs/promises'; +import { createWriteStream } from 'node:fs'; +import { mkdir, rename, writeFile } from 'node:fs/promises'; import { dirname } from 'node:path'; import { spawn } from 'node:child_process'; import { StringDecoder } from 'node:string_decoder'; import type { Logger } from './logger.js'; +import { + HeadlessOutputAccumulator, + type HeadlessOutputEvent, + type HeadlessStructuredOutput, +} from './headless-output.js'; import { resolveLaunchCommand } from './win-command.js'; const KILL_GRACE_MS = 5_000; const OUTPUT_TAIL_BYTES = 16 * 1024; const ASSISTANT_TEXT_MAX_CHARS = 64 * 1024; +const RAW_LOG_MAX_BYTES = 16 * 1024 * 1024; +const STRUCTURED_WRITE_DEBOUNCE_MS = 100; /** Scanner line buffer cap — a "line" past this without \n is not the id announcement. */ const SCAN_LINE_MAX_BYTES = 256 * 1024; @@ -44,13 +52,14 @@ export interface HeadlessTaskArgs { readonly timeoutMs: number; readonly logger: Logger; /** - * Stream the FULL stdout/stderr to these files (the task log an operator can - * open later — the in-memory tails only keep the last 16KB). Parent dir is - * created; write failure degrades to tail-only with a warn, never kills the - * run. + * Stream stdout/stderr to bounded operator logs (16MB per stream; the + * in-memory tails keep the last 16KB). Pi emits cumulative message_update + * frames, so unbounded raw logs can otherwise grow into gigabytes. */ readonly stdoutFile?: string; readonly stderrFile?: string; + /** Debounced compact structured snapshot used by the Automation UI. */ + readonly structuredFile?: string; /** * Adapter hook (`extractHeadlessSessionId`): called once per complete stdout * line until it returns a non-null agent session id; `onSessionId` then fires @@ -59,8 +68,10 @@ export interface HeadlessTaskArgs { */ readonly extractSessionId?: (line: string) => string | null; readonly onSessionId?: (id: string) => void; - /** Adapter-owned JSONL decoder for completed assistant messages. */ + /** Legacy/fallback JSONL decoder when no structured event translator exists. */ readonly extractAssistantText?: (line: string) => string | null; + /** Adapter-owned JSONL translator for response and tool blocks. */ + readonly extractOutputEvents?: (line: string) => readonly HeadlessOutputEvent[]; /** * Default false. The normal automation path refuses Windows npm .cmd shims * because the task prompt is user-controlled. Runtime readiness probes pass a @@ -77,13 +88,15 @@ export interface HeadlessTaskResult { /** True if the watchdog had to kill the process (timeout, not natural exit). */ readonly killed: boolean; readonly durationMs: number; - /** Last bytes of stdout/stderr — diagnostics only; not parsed for control flow. */ + /** Last bytes of stdout/stderr — diagnostics only; never parsed for control flow. */ readonly stdoutTail: string; readonly stderrTail: string; /** The agent's own session id, if `extractSessionId` found one in stdout. */ readonly agentSessionId: string | null; /** Latest completed assistant reply decoded from structured stdout. */ readonly assistantText: string | null; + /** Vendor-neutral reply + message/tool block timeline. */ + readonly structured: HeadlessStructuredOutput; } /** @@ -123,6 +136,8 @@ function makeStructuredOutputScanner(opts: { readonly onSessionId: (id: string) => void; readonly extractAssistantText?: (line: string) => string | null; readonly onAssistantText: (text: string) => void; + readonly extractOutputEvents?: (line: string) => readonly HeadlessOutputEvent[]; + readonly onOutputEvents: (events: readonly HeadlessOutputEvent[]) => void; }): { push(c: Buffer): void; finish(): void } { let buf = ''; let sessionDone = false; @@ -138,7 +153,12 @@ function makeStructuredOutputScanner(opts: { opts.onSessionId(id); } } - if (opts.extractAssistantText) { + // A structured translator owns assistant text as well as tool/error events. + // Do not parse the same vendor JSON twice; the old text hook remains a + // compatibility fallback for adapters without a translator. + if (opts.extractOutputEvents) { + opts.onOutputEvents(opts.extractOutputEvents(line)); + } else if (opts.extractAssistantText) { const text = opts.extractAssistantText(line)?.trim(); if (text) opts.onAssistantText(text); } @@ -167,19 +187,89 @@ function makeStructuredOutputScanner(opts: { }; } -/** Open a log write-stream, creating the parent dir; null (+ warn) on failure. */ -async function openLogStream(path: string, logger: Logger, name: string): Promise { +interface BoundedLogStream { + write(chunk: Buffer): void + end(): void +} + +/** Open a size-capped raw log stream; null (+ warn) on failure. */ +async function openLogStream(path: string, logger: Logger, name: string): Promise { try { await mkdir(dirname(path), { recursive: true }); const ws = createWriteStream(path); ws.on('error', (err) => logger.warn('headless.log_write_failed', { name, path, err })); - return ws; + let written = 0; + let capped = false; + return { + write(chunk) { + if (capped) return; + const remaining = RAW_LOG_MAX_BYTES - written; + if (remaining > 0) { + const slice = chunk.length <= remaining ? chunk : chunk.subarray(0, remaining); + ws.write(slice); + written += slice.length; + } + if (chunk.length > remaining) { + capped = true; + ws.write(Buffer.from('\n… raw log capped at 16MB; use structured output …\n')); + } + }, + end() { + ws.end(); + }, + }; } catch (err) { logger.warn('headless.log_open_failed', { name, path, err }); return null; } } +function createStructuredSnapshotWriter(path: string | undefined, logger: Logger): { + schedule(snapshot: HeadlessStructuredOutput): void + finish(snapshot: HeadlessStructuredOutput): Promise +} | null { + if (!path) return null + let latest: HeadlessStructuredOutput | null = null + let timer: NodeJS.Timeout | null = null + let chain = Promise.resolve() + + const enqueue = () => { + const snapshot = latest + if (!snapshot) return + chain = chain.then(async () => { + try { + await mkdir(dirname(path), { recursive: true }) + const tmp = `${path}.tmp` + await writeFile(tmp, JSON.stringify(snapshot), 'utf8') + await rename(tmp, path) + } catch (err) { + logger.warn('headless.structured_write_failed', { path, err }) + } + }) + } + + return { + schedule(snapshot) { + latest = snapshot + if (timer) return + timer = setTimeout(() => { + timer = null + enqueue() + }, STRUCTURED_WRITE_DEBOUNCE_MS) + timer.unref() + }, + async finish(snapshot) { + latest = snapshot + if (timer) { + clearTimeout(timer) + timer = null + } + enqueue() + await chain + }, + } +} + export async function runHeadlessTask(args: HeadlessTaskArgs): Promise { const { command, cwd, env, timeoutMs, logger } = args; const [argv0] = command; @@ -191,12 +281,15 @@ export async function runHeadlessTask(args: HeadlessTaskArgs): Promise { agentSessionId = id; logger.info('headless.session_id_captured', { agentSessionId: id }); @@ -204,6 +297,16 @@ export async function runHeadlessTask(args: HeadlessTaskArgs): Promise { assistantText = text.slice(-ASSISTANT_TEXT_MAX_CHARS); + structuredOutput.setAssistantText(assistantText); + structuredWriter?.schedule(structuredOutput.snapshot(true)); + }, + onOutputEvents: (events) => { + structuredOutput.add(events); + if (events.length > 0) { + const snapshot = structuredOutput.snapshot(true); + assistantText = snapshot.assistantText; + structuredWriter?.schedule(snapshot); + } }, }) : null; @@ -218,6 +321,8 @@ export async function runHeadlessTask(args: HeadlessTaskArgs): Promise; /** The headless-task management plane (cross-workspace; powers GET /api/headless). */ headlessTasks: HeadlessTaskRegistry; + /** Global in-flight capacity exposed to the Automation control plane. */ + headlessCapacity: number; /** Where dispatched tasks' full stdout/stderr logs land (read by the output route). */ headlessLogsDir: string; isShuttingDown(): boolean; @@ -618,6 +619,9 @@ export async function createWorkspaceService(opts: CreateWorkspaceServiceOptions ...(adapter.extractHeadlessAssistantText ? { extractAssistantText: adapter.extractHeadlessAssistantText.bind(adapter) } : {}), + ...(adapter.extractHeadlessOutputEvents + ? { extractOutputEvents: adapter.extractHeadlessOutputEvents.bind(adapter) } + : {}), }); return { result, source: effectiveSource }; }; @@ -633,6 +637,13 @@ export async function createWorkspaceService(opts: CreateWorkspaceServiceOptions stderrTail: message, agentSessionId: null, assistantText: null, + structured: { + schemaVersion: 1, + assistantText: null, + blocks: [], + metrics: { textBlocks: 0, toolCalls: 0, toolFailures: 0 }, + truncated: false, + }, }); const runtimeReadinessProbeInFlight = new Map>(); @@ -847,13 +858,18 @@ export async function createWorkspaceService(opts: CreateWorkspaceServiceOptions env, timeoutMs, logger: launcherLogger.child({ scope: 'headless', wsId: ws.id, agent: adapter.id }), - ...(logPaths ? { stdoutFile: logPaths.stdout, stderrFile: logPaths.stderr } : {}), + ...(logPaths + ? { stdoutFile: logPaths.stdout, stderrFile: logPaths.stderr, structuredFile: logPaths.structured } + : {}), ...(adapter.extractHeadlessSessionId ? { extractSessionId: adapter.extractHeadlessSessionId.bind(adapter) } : {}), ...(adapter.extractHeadlessAssistantText ? { extractAssistantText: adapter.extractHeadlessAssistantText.bind(adapter) } : {}), + ...(adapter.extractHeadlessOutputEvents + ? { extractOutputEvents: adapter.extractHeadlessOutputEvents.bind(adapter) } + : {}), ...(opts.onSessionId ? { onSessionId: opts.onSessionId } : {}), }); }; @@ -896,7 +912,7 @@ export async function createWorkspaceService(opts: CreateWorkspaceServiceOptions // Fire-and-forget: run to natural exit, then fill the record. NOTE: status // is judged by exit code — pi can exit 0 on an in-band model error, so // "done" means "process exited cleanly", not "the agent succeeded"; the - // operator confirms via the Inbox / the task's tail. + // operator confirms via the normalized output / Inbox. void runHeadlessTaskMethod(ws, adapter, prompt, timeoutMs, { taskId: rec.taskId, onSessionId: (id) => @@ -907,7 +923,8 @@ export async function createWorkspaceService(opts: CreateWorkspaceServiceOptions ), }) .then(async (r) => { - const status = r.killed ? 'failed' : r.exitCode === 0 ? 'done' : 'failed'; + const runtimeReportedError = r.structured.blocks.some((block) => block.type === 'error'); + const status = r.killed || runtimeReportedError || r.exitCode !== 0 ? 'failed' : 'done'; await headlessTasks.complete(rec.taskId, { status, finishedAt: Date.now(), @@ -915,6 +932,15 @@ export async function createWorkspaceService(opts: CreateWorkspaceServiceOptions exitCode: r.exitCode, signal: r.signal, killed: r.killed, + output: { + hasAssistantReply: r.structured.assistantText !== null, + ...(r.structured.assistantText + ? { assistantPreview: r.structured.assistantText.slice(0, 1000) } + : {}), + blockCount: r.structured.blocks.length, + toolCalls: r.structured.metrics.toolCalls, + toolFailures: r.structured.metrics.toolFailures, + }, }); // Scheduled one-shot issues are the only board items whose lifecycle can // be closed mechanically from a run exit. Repeating schedules keep their @@ -1349,6 +1375,7 @@ export async function createWorkspaceService(opts: CreateWorkspaceServiceOptions issueDetail, resolveIssuesByName, headlessTasks, + headlessCapacity: MAX_CONCURRENT_HEADLESS, headlessLogsDir, isShuttingDown: () => shuttingDown, dispose, diff --git a/ui/src/api/headless.ts b/ui/src/api/headless.ts index 49782fa78..8d175e45b 100644 --- a/ui/src/api/headless.ts +++ b/ui/src/api/headless.ts @@ -24,6 +24,33 @@ export interface HeadlessTaskRecord { * present (and the run is finished) the run can be reopened as a normal * interactive session via spawn { resume: agentSessionId }. */ agentSessionId?: string + output?: { + hasAssistantReply: boolean + assistantPreview?: string + blockCount: number + toolCalls: number + toolFailures: number + } +} + +export type HeadlessToolStatus = 'running' | 'completed' | 'failed' + +export type HeadlessMessageBlock = + | { type: 'text'; text: string } + | { type: 'tool'; id: string; name: string; status: HeadlessToolStatus; input?: unknown; output?: unknown } + | { type: 'error'; message: string } + +export interface HeadlessStructuredOutput { + schemaVersion: 1 + assistantText: string | null + blocks: HeadlessMessageBlock[] + metrics: { textBlocks: number; toolCalls: number; toolFailures: number } + truncated: boolean +} + +export interface HeadlessListSnapshot { + tasks: HeadlessTaskRecord[] + capacity: { running: number; limit: number } } /** One stream's tail from GET /api/headless/:taskId/output. */ @@ -36,24 +63,28 @@ export interface HeadlessOutputStream { export interface HeadlessOutput { taskId: string status: HeadlessTaskStatus + structured: HeadlessStructuredOutput stdout: HeadlessOutputStream | null stderr: HeadlessOutputStream | null } export const headlessApi = { - /** List headless runs across all workspaces, newest-first. */ - async list( + async snapshot( opts: { wsId?: string; status?: HeadlessTaskStatus; limit?: number } = {}, - ): Promise { + ): Promise { const q = new URLSearchParams() if (opts.wsId) q.set('wsId', opts.wsId) if (opts.status) q.set('status', opts.status) if (opts.limit) q.set('limit', String(opts.limit)) const qs = q.toString() - const { tasks } = await fetchJson<{ tasks: HeadlessTaskRecord[] }>( - `/api/headless${qs ? `?${qs}` : ''}`, - ) - return tasks + return fetchJson(`/api/headless${qs ? `?${qs}` : ''}`) + }, + + /** List headless runs across all workspaces, newest-first. */ + async list( + opts: { wsId?: string; status?: HeadlessTaskStatus; limit?: number } = {}, + ): Promise { + return (await this.snapshot(opts)).tasks }, /** Tail of a run's on-disk stdout/stderr log (poll while running). */ diff --git a/ui/src/demo/handlers/headless.ts b/ui/src/demo/handlers/headless.ts index 0f606ab6b..36b00546d 100644 --- a/ui/src/demo/handlers/headless.ts +++ b/ui/src/demo/handlers/headless.ts @@ -48,6 +48,16 @@ const demoOutput = (taskId: string): HeadlessOutput | null => { return { taskId, status: t.status, + structured: { + schemaVersion: 1, + assistantText: 'Report pushed to the inbox.', + blocks: [ + { type: 'tool', id: 'tool-1', name: 'alice analysis', status: 'completed', input: { symbol: 'NVDA' }, output: 'snapshot ready' }, + { type: 'text', text: 'Report pushed to the inbox.' }, + ], + metrics: { textBlocks: 1, toolCalls: 1, toolFailures: 0 }, + truncated: false, + }, stdout: { text, sizeBytes: text.length, truncated: false }, stderr: null, } @@ -57,7 +67,7 @@ export const headlessHandlers = [ http.get('/api/headless', ({ request }) => { const wsId = new URL(request.url).searchParams.get('wsId') const tasks = wsId ? demoHeadlessTasks.filter((t) => t.wsId === wsId) : demoHeadlessTasks - return HttpResponse.json({ tasks }) + return HttpResponse.json({ tasks, capacity: { running: tasks.filter((task) => task.status === 'running').length, limit: 8 } }) }), // Path-specific route BEFORE the :taskId catch-all (msw matches in order). http.get('/api/headless/:taskId/output', ({ params }) => { diff --git a/ui/src/pages/AutomationRunsSection.tsx b/ui/src/pages/AutomationRunsSection.tsx index e12cabfda..04e4f6402 100644 --- a/ui/src/pages/AutomationRunsSection.tsx +++ b/ui/src/pages/AutomationRunsSection.tsx @@ -1,7 +1,24 @@ -import { useCallback, useEffect, useState } from 'react' +import { useCallback, useEffect, useMemo, useState } from 'react' +import { + Bot, + ChevronDown, + ChevronRight, + CircleAlert, + ExternalLink, + MessageSquareText, + TerminalSquare, + Wrench, +} from 'lucide-react' import { api } from '../api' -import type { HeadlessOutput, HeadlessTaskRecord, HeadlessTaskStatus } from '../api/headless' +import type { + HeadlessListSnapshot, + HeadlessMessageBlock, + HeadlessOutput, + HeadlessTaskRecord, + HeadlessTaskStatus, +} from '../api/headless' +import { MarkdownContent } from '../components/MarkdownContent' import { Skeleton } from '../components/StateViews' import { useWorkspaces } from '../contexts/workspaces-context' import { formatRelativeTime } from '../lib/intl' @@ -21,12 +38,56 @@ function fmtDuration(ms?: number): string { return `${Math.floor(s / 60)}m ${s % 60}s` } -/** - * The expanded row's output log: tail of the run's on-disk stdout/stderr, - * polled while the run is still going. stdout is the agent's structured - * event stream (JSONL); shown raw — this is an operator surface. - */ -function OutputLog({ task }: { task: HeadlessTaskRecord }) { +function formatValue(value: unknown): string { + if (typeof value === 'string') return value + try { + return JSON.stringify(value, null, 2) + } catch { + return String(value) + } +} + +function ToolBlock({ block }: { block: Extract }) { + const hasDetails = block.input !== undefined || block.output !== undefined + const statusClass = block.status === 'failed' + ? 'text-red-400' + : block.status === 'completed' + ? 'text-emerald-400' + : 'text-blue-400' + return ( +
+ + + {block.name} + {block.status} + {hasDetails && } + + {hasDetails && ( +
+ {block.input !== undefined && ( +
+
Input
+ 
+                {formatValue(block.input)}
+
+
+ )} + {block.output !== undefined && ( +
+
Output
+ 
+                {formatValue(block.output)}
+
+
+ )} +
+ )} +
+ ) +} + +/** Parsed response/tool timeline with raw logs retained as a diagnostic layer. */ +function RunOutput({ task }: { task: HeadlessTaskRecord }) { const [output, setOutput] = useState(null) const [error, setError] = useState(null) const running = task.status === 'running' @@ -53,54 +114,105 @@ function OutputLog({ task }: { task: HeadlessTaskRecord }) { } }, [task.taskId, running]) - if (error) return
output unavailable: {error}
- if (!output) return
loading output…
- if (!output.stdout && !output.stderr) { - return
no output log for this run
- } + if (error) return
Output unavailable: {error}
+ if (!output) return
Loading structured output…
+ + const tools = output.structured.blocks.filter( + (block): block is Extract => block.type === 'tool', + ) + const errors = output.structured.blocks.filter( + (block): block is Extract => block.type === 'error', + ) + return ( -
- {output.stdout && ( - 
-          {output.stdout.truncated ? '… (tail)\n' : ''}
-          {output.stdout.text || '(empty)'}
-
- )} - {output.stderr && output.stderr.text.length > 0 && ( - 
-          {output.stderr.truncated ? '… (tail)\n' : ''}
-          {output.stderr.text}
-
+
+
+
+ + Reply +
+ {output.structured.assistantText ? ( + + ) : ( +

+ {running ? 'Waiting for an assistant reply…' : 'This run produced no assistant reply.'} +

+ )} +
+ + {(tools.length > 0 || errors.length > 0) && ( +
+
+ + Activity · {tools.length} tool{tools.length === 1 ? '' : 's'} +
+
+ {tools.map((block) => )} + {errors.map((block, index) => ( +
+ + {block.message} +
+ ))} +
+ {output.structured.truncated && ( +

Earlier activity was truncated; raw logs remain available below.

+ )} +
)} + +
+ + + Raw runtime logs + +
+ {output.stdout && ( + 
+              {output.stdout.truncated ? '… (tail)\n' : ''}
+              {output.stdout.text || '(empty)'}
+
+ )} + {output.stderr && output.stderr.text.length > 0 && ( + 
+              {output.stderr.truncated ? '… (tail)\n' : ''}
+              {output.stderr.text}
+
+ )} + {!output.stdout && !output.stderr &&
No raw log for this run.
} +
+
) } -/** - * Headless runs — the management panel over GET /api/headless. Every headless - * (automation) dispatch across workspaces: who's running what, status, how - * long. Expanding a row shows the full prompt + the run's output log; a - * finished run with a captured agent session id can be reopened as a normal - * interactive session for inspection/takeover. Low-frequency passive surface - * → simple polling. - */ +function SummaryCard({ label, value, detail }: { label: string; value: string; detail: string }) { + return ( +
+
{label}
+
{value}
+
{detail}
+
+ ) +} + +/** Cross-workspace control plane for concurrent native-agent runs. */ export function AutomationRunsSection() { - const [tasks, setTasks] = useState(null) + const [snapshot, setSnapshot] = useState(null) const [error, setError] = useState(null) const [expanded, setExpanded] = useState>(new Set()) const { openHeadlessRun } = useWorkspaces() - const toggle = (id: string) => - setExpanded((prev) => { - const next = new Set(prev) - if (next.has(id)) next.delete(id) - else next.add(id) - return next - }) + const toggle = (id: string) => setExpanded((prev) => { + const next = new Set(prev) + if (next.has(id)) next.delete(id) + else next.add(id) + return next + }) const load = useCallback(async () => { try { - setTasks(await api.headless.list({ limit: 100 })) + setSnapshot(await api.headless.snapshot({ limit: 100 })) setError(null) } catch (e) { setError(e instanceof Error ? e.message : String(e)) @@ -113,106 +225,120 @@ export function AutomationRunsSection() { return () => clearInterval(id) }, [load]) + const counts = useMemo(() => { + const tasks = snapshot?.tasks ?? [] + return { + done: tasks.filter((task) => task.status === 'done').length, + failed: tasks.filter((task) => task.status === 'failed' || task.status === 'interrupted').length, + } + }, [snapshot]) + if (error) return
Failed to load runs: {error}
- if (!tasks) + if (!snapshot) { return ( - - ) - if (tasks.length === 0) { - return ( -
- No headless runs yet. Dispatch one with{' '} - POST /api/workspaces/:id/headless. + ) } return ( -
- - - - - - - - - - - - - {tasks.map((t) => { - const isExpanded = expanded.has(t.taskId) - const openable = t.status !== 'running' && !!t.agentSessionId +
+
+ + + +
+ + {snapshot.tasks.length === 0 ? ( +
+ No headless runs yet. Dispatch one with POST /api/workspaces/:id/headless. +
+ ) : ( +
+ {snapshot.tasks.map((task) => { + const isExpanded = expanded.has(task.taskId) + const openable = task.status !== 'running' && !!task.agentSessionId + const toolSummary = task.output?.toolCalls + ? `${task.output.toolCalls} tool${task.output.toolCalls === 1 ? '' : 's'}` + : task.output + ? 'No tools used' + : 'Parse on open' return ( -
- - - - - - - + + {isExpanded ? : } + + + {isExpanded && ( +
+
+ + Task instructions + + 
+                        {task.prompt}
+
+
+ {task.error && ( +
+ + {task.error} +
+ )} + {openable && ( + + )} + +
+ )} + ) })} - -
StatusAgentTaskWorkspaceStartedDuration
- - {t.status} +
+
{t.agent} - - {t.error ?
{t.error}
: null} - {isExpanded && ( - <> - {openable && ( - - )} - - - )} - 		- {t.wsId.slice(0, 8)} - - {formatRelativeTime(t.startedAt)} - {fmtDuration(t.durationMs)}
+
+ )}
) } From c4d1c75d2d114f58ec306efa859b824e9877b1ec Mon Sep 17 00:00:00 2001 From: Ame <123734885+luokerenx4@users.noreply.github.com> Date: Sat, 11 Jul 2026 16:17:33 +0800 Subject: [PATCH 2/5] fix(automation): compact transient runtime events --- .../src/workspace-acceptance-smoke.spec.ts | 2 + .../desktop/src/workspace-acceptance-smoke.ts | 47 +++++++++++++----- docs/workspace-issues-and-scheduling.md | 23 ++++++--- src/webui/routes/headless.ts | 4 +- src/workspaces/adapters/codex.ts | 42 ++++++++++++++-- src/workspaces/adapters/pi.ts | 10 ++++ src/workspaces/cli-adapter.ts | 9 ++++ src/workspaces/headless-output.spec.ts | 34 ++++++++++++- src/workspaces/headless-output.ts | 5 +- src/workspaces/headless-task.spec.ts | 49 +++++++++++++++++-- src/workspaces/headless-task.ts | 46 ++++++++++++----- src/workspaces/service.ts | 6 +++ ui/src/pages/AutomationRunsSection.tsx | 8 +-- 13 files changed, 241 insertions(+), 44 deletions(-) diff --git a/apps/desktop/src/workspace-acceptance-smoke.spec.ts b/apps/desktop/src/workspace-acceptance-smoke.spec.ts index a8b5a4a9c..348fa8d1f 100644 --- a/apps/desktop/src/workspace-acceptance-smoke.spec.ts +++ b/apps/desktop/src/workspace-acceptance-smoke.spec.ts @@ -15,7 +15,9 @@ describe('Workspace acceptance renderer source', () => { } expect(source).toContain('__OPENALICE_WORKSPACE_%s_FAILED__ %s %s\\\\n%s\\\\n') expect(source).toContain('managedPiStructuredOutput') + expect(source).toContain('managedPiDiagnosticCompaction') expect(source).toContain("block?.type === 'tool' && block?.status === 'completed'") + expect(source).toContain("diagnosticText.includes('\"type\":\"message_update\"')") expect(() => new Function(`return ${source}`)).not.toThrow() }) }) diff --git a/apps/desktop/src/workspace-acceptance-smoke.ts b/apps/desktop/src/workspace-acceptance-smoke.ts index 3fe2c596b..4374fc92e 100644 --- a/apps/desktop/src/workspace-acceptance-smoke.ts +++ b/apps/desktop/src/workspace-acceptance-smoke.ts @@ -17,6 +17,7 @@ export interface WorkspaceAcceptanceReceipt { readonly shellCliRoundTrip: boolean readonly managedPiAssistantReply: boolean readonly managedPiStructuredOutput: boolean + readonly managedPiDiagnosticCompaction: boolean readonly managedPiCliSideEffect: boolean readonly cleanupComplete: boolean } @@ -54,6 +55,7 @@ export async function runRendererWorkspaceAcceptanceSmoke( shellCliRoundTrip: false, managedPiAssistantReply: false, managedPiStructuredOutput: false, + managedPiDiagnosticCompaction: false, managedPiCliSideEffect: false, cleanupComplete: false, } @@ -68,6 +70,15 @@ export async function runRendererWorkspaceAcceptanceSmoke( const owner = snapshot?.workspaces?.find((row) => row.wsId === workspaceId) return Boolean(owner?.issues?.some((issue) => issue.id === issueId)) } + const waitForHeadlessRun = async (taskId) => { + const deadline = Date.now() + 120000 + while (Date.now() < deadline) { + const record = await json(await fetch('/api/headless/' + encodeURIComponent(taskId))) + if (record.status !== 'running') return record + await new Promise((resolve) => setTimeout(resolve, 100)) + } + throw new Error('managed Pi Automation run timed out: ' + taskId) + } const decode = (value) => { if (typeof value === 'string') return value if (value instanceof Uint8Array) return new TextDecoder().decode(value) @@ -208,32 +219,42 @@ export async function runRendererWorkspaceAcceptanceSmoke( headers: { 'content-type': 'application/json' }, body: JSON.stringify({ agent: 'pi', - wait: true, timeoutMs: 120000, prompt: '${ACCEPTANCE_MARKER}: execute the requested Workspace CLI acceptance action, then report completion.', }), })) - if (headless.killed || headless.exitCode !== 0) { + const headlessRecord = await waitForHeadlessRun(headless.taskId) + const headlessOutput = await json(await fetch('/api/headless/' + encodeURIComponent(headless.taskId) + '/output')) + if (headlessRecord.status !== 'done' || headlessRecord.killed || headlessRecord.exitCode !== 0) { throw new Error('managed Pi headless run failed: ' + JSON.stringify({ - exitCode: headless.exitCode, - killed: headless.killed, - stderrTail: headless.stderrTail, + status: headlessRecord.status, + exitCode: headlessRecord.exitCode, + killed: headlessRecord.killed, + error: headlessRecord.error, + stderr: headlessOutput.stderr, })) } - if (headless.assistantText?.trim() !== '${ASSISTANT_TEXT}') { - throw new Error('managed Pi assistant reply was not decoded: ' + JSON.stringify(headless.assistantText)) + if (headlessOutput.structured?.assistantText?.trim() !== '${ASSISTANT_TEXT}') { + throw new Error('managed Pi assistant reply was not decoded: ' + JSON.stringify(headlessOutput.structured?.assistantText)) } checks.managedPiAssistantReply = true if ( - headless.structured?.schemaVersion !== 1 || - headless.structured?.assistantText?.trim() !== '${ASSISTANT_TEXT}' || - typeof headless.structured?.metrics?.toolCalls !== 'number' || - headless.structured.metrics.toolCalls < 1 || - !headless.structured?.blocks?.some((block) => block?.type === 'tool' && block?.status === 'completed') + headlessOutput.structured?.schemaVersion !== 1 || + typeof headlessOutput.structured?.metrics?.toolCalls !== 'number' || + headlessOutput.structured.metrics.toolCalls < 1 || + !headlessOutput.structured?.blocks?.some((block) => block?.type === 'tool' && block?.status === 'completed') ) { - throw new Error('managed Pi structured output was not decoded: ' + JSON.stringify(headless.structured)) + throw new Error('managed Pi structured output was not decoded: ' + JSON.stringify(headlessOutput.structured)) } checks.managedPiStructuredOutput = true + const diagnosticText = headlessOutput.stdout?.text || '' + if ( + diagnosticText.includes('"type":"message_update"') || + diagnosticText.includes('"type":"tool_execution_update"') + ) { + throw new Error('managed Pi diagnostic log retained transient updates') + } + checks.managedPiDiagnosticCompaction = true const agentIssues = await json(await fetch('/api/issues')) if (!issueExists(agentIssues, workspaceId, agentIssueId)) { diff --git a/docs/workspace-issues-and-scheduling.md b/docs/workspace-issues-and-scheduling.md index bad78919e..23ddb70e2 100644 --- a/docs/workspace-issues-and-scheduling.md +++ b/docs/workspace-issues-and-scheduling.md @@ -118,16 +118,27 @@ is active: - tool name, input, output, and `running | completed | failed` status; - compact metrics for reply presence, tool count, and tool failures. +The native stream contracts differ materially: + +| Runtime | Native one-shot stream | Normalization posture | +|---|---|---| +| Claude Code | completed assistant/tool messages plus result | pair `tool_use` / `tool_result`; keep the latest assistant result | +| Codex | thread/turn lifecycle and started/updated/completed items | commands, file changes, MCP, web search, and collaboration become tools; stream/turn/error items become errors | +| opencode | completed text/tool parts plus step boundaries | terminal tool snapshots become one completed/failed tool block; no token-delta persistence | +| Pi | every session event, including cumulative message/tool updates | parse final messages and tool boundaries; discard transient updates from diagnostics before disk | + Automation reads a debounced `.structured.json` snapshot instead of replaying an entire vendor log. This makes live polling cheap and gives future workbench orchestration a stable contract independent of CLI versions. Runs created before this contract are parsed best-effort from the last 2 MB of stdout when opened. -Raw stdout/stderr remain available only as a diagnostic fallback. Each stream -is capped at 16 MB because Pi's JSON mode can emit cumulative `message_update` -frames; appending every cumulative frame without a cap can otherwise turn one -normal conversation into a multi-gigabyte log. Normalized output is separately -bounded to 300 blocks, 64 KB per text reply, and 8 KB per tool input/output. +Bounded stdout/stderr diagnostics remain as a fallback. Adapters may discard +documented high-frequency transient events before persistence: Pi drops +`message_update` (which repeats both the cumulative partial and current message) +and `tool_execution_update`, while retaining final messages, tool boundaries, +errors, and lifecycle events. Each diagnostic stream is still capped at 16 MB +as a second guard. Normalized output is separately bounded to 300 blocks, 64 KB +per text reply, and 8 KB per tool input/output. ## Delivery and Trading Safety @@ -168,7 +179,7 @@ human approval boundaries. | `src/workspaces/headless-task-registry.ts` | Concurrent run records, capacity projection, and log pruning | | `src/workspaces/headless-output.ts` | Vendor-neutral reply/tool block contract and accumulator | | `src/workspaces/adapters/{claude,codex,opencode,pi}.ts` | Runtime-specific JSON event translation | -| `src/webui/routes/headless.ts` | Cross-workspace capacity, task, normalized output, and raw-tail API | +| `src/webui/routes/headless.ts` | Cross-workspace capacity, task, normalized output, and diagnostic-tail API | | `ui/src/pages/AutomationRunsSection.tsx` | Run list, final reply, tool activity, and diagnostics UI | | `src/tool/issue-tools.ts` | Workspace-scoped issue CLI/MCP tools | | `src/tool/inbox-push.ts` | Headless/interactive delivery to Inbox | diff --git a/src/webui/routes/headless.ts b/src/webui/routes/headless.ts index 20be271fd..686072dfc 100644 --- a/src/webui/routes/headless.ts +++ b/src/webui/routes/headless.ts @@ -4,7 +4,7 @@ * Read-only view over `WorkspaceService.headlessTasks`: "what are the workers * doing" across every workspace. Dispatch lives at POST /api/workspaces/:id/ * headless (it's per-workspace); this surface is the panel + per-task status - * + its normalized reply/tool timeline and size-capped raw diagnostic logs. + * + its normalized reply/tool timeline and size-capped runtime diagnostics. */ import { open, readFile, stat } from 'node:fs/promises' @@ -85,7 +85,7 @@ export function createHeadlessRoutes(svc: WorkspaceService): Hono { }) // GET /api/headless/:taskId/output?tailBytes= → compact normalized output + - // raw diagnostic tails. New runs read their live structured snapshot; + // bounded diagnostic tails. New runs read their live structured snapshot; // historical runs are parsed from a bounded stdout tail. Streams are null // when the log file doesn't exist (old task, pruned log, or spawn failure). app.get('/:taskId/output', async (c) => { diff --git a/src/workspaces/adapters/codex.ts b/src/workspaces/adapters/codex.ts index 35f271a41..02e78e9a5 100644 --- a/src/workspaces/adapters/codex.ts +++ b/src/workspaces/adapters/codex.ts @@ -151,6 +151,9 @@ export const codexAdapter: CliAdapter = { extractHeadlessOutputEvents(line: string): readonly HeadlessOutputEvent[] { try { const evt = JSON.parse(line) as Record; + if (evt['type'] === 'error' && typeof evt['message'] === 'string') { + return [{ type: 'error', message: evt['message'] }]; + } if (evt['type'] === 'turn.failed') { const error = evt['error']; const message = error && typeof error === 'object' && typeof (error as Record)['message'] === 'string' @@ -165,25 +168,37 @@ export const codexAdapter: CliAdapter = { if (!item || typeof item !== 'object') return []; const record = item as Record; const id = typeof record['id'] === 'string' ? record['id'] : `codex-${record['type'] ?? 'item'}`; + if (evt['type'] === 'item.completed' && record['type'] === 'error' && typeof record['message'] === 'string') { + return [{ type: 'error', message: record['message'] }]; + } if (evt['type'] === 'item.completed' && record['type'] === 'agent_message' && typeof record['text'] === 'string') { return [{ type: 'text', text: record['text'] }]; } if (record['type'] === 'command_execution') { const input = typeof record['command'] === 'string' ? { command: record['command'] } : record['command']; if (evt['type'] === 'item.started') return [{ type: 'tool-start', id, name: 'Shell', input }]; + const failed = record['status'] === 'failed' || + record['status'] === 'declined' || + (typeof record['exit_code'] === 'number' && record['exit_code'] !== 0); return [{ type: 'tool-finish', id, name: 'Shell', ...(record['aggregated_output'] !== undefined ? { output: record['aggregated_output'] } : {}), - ...(typeof record['exit_code'] === 'number' && record['exit_code'] !== 0 ? { isError: true } : {}), + ...(failed ? { isError: true } : {}), }]; } if (record['type'] === 'file_change') { if (evt['type'] === 'item.started') { return [{ type: 'tool-start', id, name: 'File changes', input: record['changes'] }]; } - return [{ type: 'tool-finish', id, name: 'File changes', output: record['changes'] }]; + return [{ + type: 'tool-finish', + id, + name: 'File changes', + output: record['changes'], + ...(record['status'] === 'failed' ? { isError: true } : {}), + }]; } if (record['type'] === 'mcp_tool_call' || record['type'] === 'tool_call') { const name = typeof record['tool'] === 'string' @@ -198,7 +213,28 @@ export const codexAdapter: CliAdapter = { type: 'tool-finish', id, name, - output: record['result'] ?? record['output'], + output: record['result'] ?? record['output'] ?? record['error'], + ...(record['status'] === 'failed' ? { isError: true } : {}), + }]; + } + if (record['type'] === 'web_search') { + const input = { query: record['query'], action: record['action'] }; + if (evt['type'] === 'item.started') return [{ type: 'tool-start', id, name: 'Web search', input }]; + return [{ type: 'tool-finish', id, name: 'Web search', output: input }]; + } + if (record['type'] === 'collab_tool_call') { + const rawTool = typeof record['tool'] === 'string' ? record['tool'] : 'collaboration'; + const name = `Collaboration · ${rawTool.replaceAll('_', ' ')}`; + const input = { + ...(record['receiver_thread_ids'] !== undefined ? { receiverThreadIds: record['receiver_thread_ids'] } : {}), + ...(record['prompt'] !== undefined ? { prompt: record['prompt'] } : {}), + }; + if (evt['type'] === 'item.started') return [{ type: 'tool-start', id, name, input }]; + return [{ + type: 'tool-finish', + id, + name, + output: record['agents_states'], ...(record['status'] === 'failed' ? { isError: true } : {}), }]; } diff --git a/src/workspaces/adapters/pi.ts b/src/workspaces/adapters/pi.ts index 871ac0396..0aecbd2b8 100644 --- a/src/workspaces/adapters/pi.ts +++ b/src/workspaces/adapters/pi.ts @@ -271,6 +271,16 @@ export const piAdapter: CliAdapter = { } }, + // JSON mode intentionally emits every streaming event. Its documented + // message_update payload contains both a cumulative partial message and the + // current message snapshot; tool_execution_update likewise carries partial + // progress. They are useful for live rendering, not durable one-shot + // diagnostics. The structured parser still sees the full stream. + keepHeadlessDiagnosticLine(line: string): boolean { + return !line.startsWith('{"type":"message_update"') && + !line.startsWith('{"type":"tool_execution_update"'); + }, + composeEnv(ctx: SpawnContext): Record { // Do not force PI_OFFLINE. OpenAlice is a networked product and Pi may // download missing runtime tools during startup. A user or launcher can diff --git a/src/workspaces/cli-adapter.ts b/src/workspaces/cli-adapter.ts index 83f446ead..c71668850 100644 --- a/src/workspaces/cli-adapter.ts +++ b/src/workspaces/cli-adapter.ts @@ -214,6 +214,15 @@ export interface CliAdapter { /** Translate one native JSONL line into vendor-neutral response/tool events. */ extractHeadlessOutputEvents?(line: string): readonly HeadlessOutputEvent[]; + /** + * Decide whether one complete stdout line belongs in the bounded diagnostic + * log/tail. Structured parsers still see every line. Use this for documented + * high-frequency transient events (Pi's cumulative `message_update` and + * `tool_execution_update`) that are useful to a live TUI but pathological in + * a persisted one-shot run log. Omit to preserve stdout byte-for-byte. + */ + keepHeadlessDiagnosticLine?(line: string): boolean; + /** Optional per-CLI env adjustments on top of `spawn-env.ts`'s baseline. */ envOverrides?(parent: NodeJS.ProcessEnv): EnvOverrides; diff --git a/src/workspaces/headless-output.spec.ts b/src/workspaces/headless-output.spec.ts index 406aedc94..09eb85862 100644 --- a/src/workspaces/headless-output.spec.ts +++ b/src/workspaces/headless-output.spec.ts @@ -57,6 +57,8 @@ describe('headless structured output', () => { it('keeps runtime-level failures as error blocks', () => { const codex = parse(codexAdapter, [ + { type: 'item.completed', item: { id: 'e1', type: 'error', message: 'Codex reconnecting' } }, + { type: 'error', message: 'Codex stream failed' }, { type: 'turn.failed', error: { message: 'Codex provider unavailable' } }, ]) const opencode = parse(opencodeAdapter, [ @@ -70,11 +72,38 @@ describe('headless structured output', () => { }, }, ]) - expect(codex.blocks).toContainEqual({ type: 'error', message: 'Codex provider unavailable' }) + expect(codex.blocks).toEqual([ + { type: 'error', message: 'Codex reconnecting' }, + { type: 'error', message: 'Codex stream failed' }, + { type: 'error', message: 'Codex provider unavailable' }, + ]) expect(opencode.blocks).toContainEqual({ type: 'error', message: 'OpenCode provider unavailable' }) expect(pi.blocks).toContainEqual({ type: 'error', message: 'Pi provider unavailable' }) }) + it('deduplicates repeated terminal runtime errors', () => { + const output = parse(codexAdapter, [ + { type: 'error', message: 'provider unavailable' }, + { type: 'turn.failed', error: { message: 'provider unavailable' } }, + ]) + expect(output.blocks).toEqual([{ type: 'error', message: 'provider unavailable' }]) + }) + + it('normalizes current Codex web-search, MCP, and collaboration items', () => { + const output = parse(codexAdapter, [ + { type: 'item.started', item: { id: 'w1', type: 'web_search', query: 'OpenAlice', action: { type: 'search' } } }, + { type: 'item.completed', item: { id: 'w1', type: 'web_search', query: 'OpenAlice', action: { type: 'search' } } }, + { type: 'item.started', item: { id: 'm1', type: 'mcp_tool_call', tool: 'lookup', arguments: { q: 'x' }, status: 'in_progress' } }, + { type: 'item.completed', item: { id: 'm1', type: 'mcp_tool_call', tool: 'lookup', error: { message: 'offline' }, status: 'failed' } }, + { type: 'item.started', item: { id: 'c1', type: 'collab_tool_call', tool: 'spawn_agent', receiver_thread_ids: [], prompt: 'inspect', status: 'in_progress' } }, + { type: 'item.completed', item: { id: 'c1', type: 'collab_tool_call', tool: 'spawn_agent', receiver_thread_ids: ['child'], agents_states: { child: { status: 'completed' } }, status: 'completed' } }, + ]) + expect(output.metrics).toEqual({ textBlocks: 0, toolCalls: 3, toolFailures: 1 }) + expect(output.blocks).toContainEqual(expect.objectContaining({ id: 'w1', name: 'Web search', status: 'completed' })) + expect(output.blocks).toContainEqual(expect.objectContaining({ id: 'm1', name: 'lookup', status: 'failed', output: { message: 'offline' } })) + expect(output.blocks).toContainEqual(expect.objectContaining({ id: 'c1', name: 'Collaboration · spawn agent', status: 'completed' })) + }) + it('normalizes Pi execution events and failed tools', () => { const output = parse(piAdapter, [ { type: 'tool_execution_start', toolCallId: 'p1', toolName: 'bash', args: { command: 'false' } }, @@ -89,6 +118,9 @@ describe('headless structured output', () => { const cumulativeFrame = '{"type":"message_update", deliberately-not-valid-json' expect(piAdapter.extractHeadlessOutputEvents?.(cumulativeFrame)).toEqual([]) expect(piAdapter.extractHeadlessAssistantText?.(cumulativeFrame)).toBeNull() + expect(piAdapter.keepHeadlessDiagnosticLine?.(cumulativeFrame)).toBe(false) + expect(piAdapter.keepHeadlessDiagnosticLine?.('{"type":"tool_execution_update"}')).toBe(false) + expect(piAdapter.keepHeadlessDiagnosticLine?.('{"type":"message_end"}')).toBe(true) }) it('accepts OpenCode part.state tool snapshots', () => { diff --git a/src/workspaces/headless-output.ts b/src/workspaces/headless-output.ts index 5f4e399ae..d96b1e3b3 100644 --- a/src/workspaces/headless-output.ts +++ b/src/workspaces/headless-output.ts @@ -81,7 +81,10 @@ export class HeadlessOutputAccumulator { if (event.type === 'error') { const message = clipText(event.message.trim(), MAX_TOOL_VALUE_CHARS) - if (message) this.push({ type: 'error', message }) + const previous = this.blocks[this.blocks.length - 1] + if (message && !(previous?.type === 'error' && previous.message === message)) { + this.push({ type: 'error', message }) + } continue } diff --git a/src/workspaces/headless-task.spec.ts b/src/workspaces/headless-task.spec.ts index a5c08667e..264aeba07 100644 --- a/src/workspaces/headless-task.spec.ts +++ b/src/workspaces/headless-task.spec.ts @@ -142,7 +142,7 @@ describe('runHeadlessTask', () => { expect(r.structured.blocks).toHaveLength(2); }); - it('streams raw stdout/stderr to log files beyond the 16KB in-memory tails', async () => { + it('streams stdout/stderr diagnostics beyond the 16KB in-memory tails', async () => { const dir = await mkdtemp(join(tmpdir(), 'headless-log-')); try { // 64KB of stdout — far past the 16KB tail budget. @@ -176,6 +176,49 @@ describe('runHeadlessTask', () => { } }); + it('compacts line-oriented diagnostics without hiding lines from structured parsing', async () => { + const dir = await mkdtemp(join(tmpdir(), 'headless-filtered-log-')); + try { + const stdoutFile = join(dir, 't1.stdout.log'); + const lines = [ + JSON.stringify({ type: 'message_update', text: 'cumulative snapshot' }), + JSON.stringify({ type: 'tool_execution_update', text: 'partial tool output' }), + JSON.stringify({ type: 'message_end', text: 'Final reply' }), + ]; + const seen: string[] = []; + const result = await runHeadlessTask({ + command: ['node', '-e', `process.stdout.write(${JSON.stringify(`${lines.join('\n')}\n`)})`], + cwd: process.cwd(), + env: baseEnv, + timeoutMs: 5_000, + logger: noopLogger, + stdoutFile, + keepDiagnosticLine: (line) => !line.includes('_update"'), + extractOutputEvents: (line) => { + seen.push(line); + const event = JSON.parse(line) as { type?: string; text?: string }; + return event.type === 'message_end' && event.text + ? [{ type: 'text' as const, text: event.text }] + : []; + }, + }); + let diagnostic = ''; + for (let i = 0; i < 40 && !diagnostic.includes('message_end'); i++) { + await new Promise((resolve) => setTimeout(resolve, 25)); + diagnostic = await readFile(stdoutFile, 'utf8').catch(() => ''); + } + expect(seen).toHaveLength(3); + expect(result.structured.assistantText).toBe('Final reply'); + expect(result.stdoutTail).toContain('message_end'); + expect(result.stdoutTail).not.toContain('message_update'); + expect(diagnostic).toContain('message_end'); + expect(diagnostic).not.toContain('message_update'); + expect(diagnostic).not.toContain('tool_execution_update'); + } finally { + await rm(dir, { recursive: true, force: true }); + } + }); + it('writes a compact structured snapshot for live Automation polling', async () => { const dir = await mkdtemp(join(tmpdir(), 'headless-structured-')); try { @@ -207,7 +250,7 @@ describe('runHeadlessTask', () => { } }); - it('caps each raw diagnostic stream at 16MB', async () => { + it('caps each diagnostic stream at 16MB', async () => { const dir = await mkdtemp(join(tmpdir(), 'headless-log-cap-')); try { const stdoutFile = join(dir, 't1.stdout.log'); @@ -227,7 +270,7 @@ describe('runHeadlessTask', () => { } const stored = await readFile(stdoutFile, 'utf8'); expect(size).toBeLessThan(16 * 1024 * 1024 + 256); - expect(stored).toContain('raw log capped at 16MB'); + expect(stored).toContain('diagnostic log capped at 16MB'); } finally { await rm(dir, { recursive: true, force: true }); } diff --git a/src/workspaces/headless-task.ts b/src/workspaces/headless-task.ts index 0aa5cbc43..98cb60361 100644 --- a/src/workspaces/headless-task.ts +++ b/src/workspaces/headless-task.ts @@ -18,7 +18,7 @@ * circuit is anti-semantic for a one-shot task (exit == completion). * * The launcher normalizes each runtime's JSON event stream into a compact, - * vendor-neutral reply/tool timeline while preserving size-capped raw logs for + * vendor-neutral reply/tool timeline while preserving size-capped operator * diagnostics. `inbox_push` remains the durable user-delivery channel. */ import { createWriteStream } from 'node:fs'; @@ -54,7 +54,7 @@ export interface HeadlessTaskArgs { /** * Stream stdout/stderr to bounded operator logs (16MB per stream; the * in-memory tails keep the last 16KB). Pi emits cumulative message_update - * frames, so unbounded raw logs can otherwise grow into gigabytes. + * frames, so unbounded diagnostics can otherwise grow into gigabytes. */ readonly stdoutFile?: string; readonly stderrFile?: string; @@ -72,6 +72,8 @@ export interface HeadlessTaskArgs { readonly extractAssistantText?: (line: string) => string | null; /** Adapter-owned JSONL translator for response and tool blocks. */ readonly extractOutputEvents?: (line: string) => readonly HeadlessOutputEvent[]; + /** Adapter-owned compaction filter for persisted stdout diagnostics. */ + readonly keepDiagnosticLine?: (line: string) => boolean; /** * Default false. The normal automation path refuses Windows npm .cmd shims * because the task prompt is user-controlled. Runtime readiness probes pass a @@ -138,12 +140,14 @@ function makeStructuredOutputScanner(opts: { readonly onAssistantText: (text: string) => void; readonly extractOutputEvents?: (line: string) => readonly HeadlessOutputEvent[]; readonly onOutputEvents: (events: readonly HeadlessOutputEvent[]) => void; + readonly onDiagnosticLine?: (rawLine: string, terminated: boolean) => void; }): { push(c: Buffer): void; finish(): void } { let buf = ''; let sessionDone = false; const decoder = new StringDecoder('utf8'); - const inspect = (raw: string) => { + const inspect = (raw: string, terminated: boolean) => { + opts.onDiagnosticLine?.(raw, terminated); const line = raw.trim(); if (!line) return; if (!sessionDone && opts.extractSessionId) { @@ -167,7 +171,7 @@ function makeStructuredOutputScanner(opts: { const drain = () => { let nl: number; while ((nl = buf.indexOf('\n')) !== -1) { - inspect(buf.slice(0, nl)); + inspect(buf.slice(0, nl), true); buf = buf.slice(nl + 1); } if (buf.length > SCAN_LINE_MAX_BYTES) buf = ''; @@ -181,7 +185,7 @@ function makeStructuredOutputScanner(opts: { finish() { buf += decoder.end(); drain(); - if (buf.trim()) inspect(buf); + if (buf.length > 0) inspect(buf, false); buf = ''; }, }; @@ -192,7 +196,7 @@ interface BoundedLogStream { end(): void } -/** Open a size-capped raw log stream; null (+ warn) on failure. */ +/** Open a size-capped diagnostic stream; null (+ warn) on failure. */ async function openLogStream(path: string, logger: Logger, name: string): Promise { try { await mkdir(dirname(path), { recursive: true }); @@ -211,7 +215,7 @@ async function openLogStream(path: string, logger: Logger, name: string): Promis } if (chunk.length > remaining) { capped = true; - ws.write(Buffer.from('\n… raw log capped at 16MB; use structured output …\n')); + ws.write(Buffer.from('\n… diagnostic log capped at 16MB; use structured output …\n')); } }, end() { @@ -285,7 +289,8 @@ export async function runHeadlessTask(args: HeadlessTaskArgs): Promise { + const line = rawLine.trim(); + let keep = true; + try { + keep = !line || args.keepDiagnosticLine?.(line) !== false; + } catch (err) { + logger.warn('headless.diagnostic_filter_failed', { err }); + } + if (!keep) return; + const chunk = Buffer.from(terminated ? `${rawLine}\n` : rawLine); + outSink.push(chunk); + outFile?.write(chunk); + }, + } + : {}), }) : null; // win32: resolve the bare CLI name against PATH × PATHEXT. Native-exe agents @@ -343,7 +365,7 @@ export async function runHeadlessTask(args: HeadlessTaskArgs): Promise { - outSink.push(d); + if (!args.keepDiagnosticLine) { + outSink.push(d); + outFile?.write(d); + } scanner?.push(d); - outFile?.write(d); }); child.stderr?.on('data', (d: Buffer) => { errSink.push(d); diff --git a/src/workspaces/service.ts b/src/workspaces/service.ts index 2a7b8d2ef..9180c288a 100644 --- a/src/workspaces/service.ts +++ b/src/workspaces/service.ts @@ -622,6 +622,9 @@ export async function createWorkspaceService(opts: CreateWorkspaceServiceOptions ...(adapter.extractHeadlessOutputEvents ? { extractOutputEvents: adapter.extractHeadlessOutputEvents.bind(adapter) } : {}), + ...(adapter.keepHeadlessDiagnosticLine + ? { keepDiagnosticLine: adapter.keepHeadlessDiagnosticLine.bind(adapter) } + : {}), }); return { result, source: effectiveSource }; }; @@ -870,6 +873,9 @@ export async function createWorkspaceService(opts: CreateWorkspaceServiceOptions ...(adapter.extractHeadlessOutputEvents ? { extractOutputEvents: adapter.extractHeadlessOutputEvents.bind(adapter) } : {}), + ...(adapter.keepHeadlessDiagnosticLine + ? { keepDiagnosticLine: adapter.keepHeadlessDiagnosticLine.bind(adapter) } + : {}), ...(opts.onSessionId ? { onSessionId: opts.onSessionId } : {}), }); }; diff --git a/ui/src/pages/AutomationRunsSection.tsx b/ui/src/pages/AutomationRunsSection.tsx index 04e4f6402..fd47dc1cc 100644 --- a/ui/src/pages/AutomationRunsSection.tsx +++ b/ui/src/pages/AutomationRunsSection.tsx @@ -86,7 +86,7 @@ function ToolBlock({ block }: { block: Extract(null) const [error, setError] = useState(null) @@ -156,7 +156,7 @@ function RunOutput({ task }: { task: HeadlessTaskRecord }) { ))}
{output.structured.truncated && ( -

Earlier activity was truncated; raw logs remain available below.

+

Earlier activity was truncated; runtime diagnostics remain available below.

)} )} @@ -164,7 +164,7 @@ function RunOutput({ task }: { task: HeadlessTaskRecord }) {
- Raw runtime logs + Runtime diagnostics
{output.stdout && ( @@ -179,7 +179,7 @@ function RunOutput({ task }: { task: HeadlessTaskRecord }) { {output.stderr.text} )} - {!output.stdout && !output.stderr &&
No raw log for this run.
} + {!output.stdout && !output.stderr &&
No runtime diagnostics for this run.
}
From 6c2f117c9b01d426387507f9da25f5d33f0dcfeb Mon Sep 17 00:00:00 2001 From: Ame <123734885+luokerenx4@users.noreply.github.com> Date: Sat, 11 Jul 2026 16:20:36 +0800 Subject: [PATCH 3/5] fix(automation): make run history scrollable --- ui/src/pages/AutomationPage.tsx | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/ui/src/pages/AutomationPage.tsx b/ui/src/pages/AutomationPage.tsx index 545fed0cb..f254e4aee 100644 --- a/ui/src/pages/AutomationPage.tsx +++ b/ui/src/pages/AutomationPage.tsx @@ -31,7 +31,10 @@ export function AutomationPage({ spec }: AutomationPageProps) { return (
-
+
{section === 'api' ? ( From 5b6922d1d0f90a2fa6e8dd3b6650d56c34edb506 Mon Sep 17 00:00:00 2001 From: Ame <123734885+luokerenx4@users.noreply.github.com> Date: Sat, 11 Jul 2026 16:28:11 +0800 Subject: [PATCH 4/5] test(automation): cover current Claude and Codex streams --- src/workspaces/headless-output.spec.ts | 82 ++++++++++++++++++++++++++ 1 file changed, 82 insertions(+) diff --git a/src/workspaces/headless-output.spec.ts b/src/workspaces/headless-output.spec.ts index 09eb85862..6a58e3dbb 100644 --- a/src/workspaces/headless-output.spec.ts +++ b/src/workspaces/headless-output.spec.ts @@ -41,6 +41,53 @@ describe('headless structured output', () => { expect(output.metrics).toEqual({ textBlocks: 1, toolCalls: 1, toolFailures: 0 }) }) + it('normalizes the Claude 2.1.202 failed-tool stream while ignoring thinking as a reply', () => { + const output = parse(claudeAdapter, [ + { type: 'system', subtype: 'init', session_id: 'claude-session', model: 'MiniMax-M3' }, + { type: 'system', subtype: 'thinking_tokens', session_id: 'claude-session', estimated_tokens: 42 }, + { + type: 'assistant', + message: { role: 'assistant', content: [{ type: 'thinking', thinking: 'I should inspect the file.' }] }, + }, + { + type: 'assistant', + message: { + role: 'assistant', + content: [{ + type: 'tool_use', id: 'call-live', name: 'Read', + input: { file_path: '/workspace/AGENTS.md', limit: 1, pages: '' }, + }], + }, + }, + { + type: 'user', + message: { + role: 'user', + content: [{ + type: 'tool_result', tool_use_id: 'call-live', is_error: true, + content: 'Invalid pages parameter', + }], + }, + }, + { + type: 'assistant', + message: { role: 'assistant', content: [{ type: 'text', text: 'CLAUDE_ASYNC_RUN_OK' }] }, + }, + { type: 'result', subtype: 'success', is_error: false, result: 'CLAUDE_ASYNC_RUN_OK' }, + ]) + + expect(output.assistantText).toBe('CLAUDE_ASYNC_RUN_OK') + expect(output.blocks).toEqual([ + { + type: 'tool', id: 'call-live', name: 'Read', status: 'failed', + input: { file_path: '/workspace/AGENTS.md', limit: 1, pages: '' }, + output: 'Invalid pages parameter', + }, + { type: 'text', text: 'CLAUDE_ASYNC_RUN_OK' }, + ]) + expect(output.metrics).toEqual({ textBlocks: 1, toolCalls: 1, toolFailures: 1 }) + }) + it('normalizes Codex command execution and file-change items', () => { const output = parse(codexAdapter, [ { type: 'item.started', item: { id: 'c1', type: 'command_execution', command: 'alice status', status: 'in_progress' } }, @@ -55,6 +102,41 @@ describe('headless structured output', () => { expect(output.blocks).toContainEqual(expect.objectContaining({ type: 'tool', id: 'f1', name: 'File changes', status: 'completed' })) }) + it('normalizes the Codex 0.144.1 command stream and keeps the latest reply', () => { + const output = parse(codexAdapter, [ + { type: 'thread.started', thread_id: 'codex-session' }, + { type: 'turn.started' }, + { type: 'item.completed', item: { id: 'item_0', type: 'agent_message', text: 'I will run the check.' } }, + { + type: 'item.started', + item: { + id: 'item_1', type: 'command_execution', command: "/bin/zsh -lc 'printf CODEX_TOOL_OK'", + aggregated_output: '', exit_code: null, status: 'in_progress', + }, + }, + { + type: 'item.completed', + item: { + id: 'item_1', type: 'command_execution', command: "/bin/zsh -lc 'printf CODEX_TOOL_OK'", + aggregated_output: 'CODEX_TOOL_OK', exit_code: 0, status: 'completed', + }, + }, + { type: 'item.completed', item: { id: 'item_2', type: 'agent_message', text: 'CODEX_HEADLESS_OK' } }, + { type: 'turn.completed', usage: { input_tokens: 1, output_tokens: 1 } }, + ]) + + expect(output.assistantText).toBe('CODEX_HEADLESS_OK') + expect(output.blocks).toEqual([ + { type: 'text', text: 'I will run the check.' }, + { + type: 'tool', id: 'item_1', name: 'Shell', status: 'completed', + input: { command: "/bin/zsh -lc 'printf CODEX_TOOL_OK'" }, output: 'CODEX_TOOL_OK', + }, + { type: 'text', text: 'CODEX_HEADLESS_OK' }, + ]) + expect(output.metrics).toEqual({ textBlocks: 2, toolCalls: 1, toolFailures: 0 }) + }) + it('keeps runtime-level failures as error blocks', () => { const codex = parse(codexAdapter, [ { type: 'item.completed', item: { id: 'e1', type: 'error', message: 'Codex reconnecting' } }, From d957ed06e89125f0413d0a94c9bf44e42d019d86 Mon Sep 17 00:00:00 2001 From: Ame <123734885+luokerenx4@users.noreply.github.com> Date: Sat, 11 Jul 2026 16:44:13 +0800 Subject: [PATCH 5/5] feat(automation): paginate run history --- docs/workspace-issues-and-scheduling.md | 7 +- src/webui/routes/headless.spec.ts | 37 +++++-- src/webui/routes/headless.ts | 30 +++++- src/workspaces/headless-task-registry.spec.ts | 13 +++ src/workspaces/headless-task-registry.ts | 30 +++++- ui/src/api/headless.ts | 7 +- ui/src/demo/handlers/headless.ts | 10 +- ui/src/pages/AutomationRunsSection.tsx | 98 ++++++++++++++++--- 8 files changed, 206 insertions(+), 26 deletions(-) diff --git a/docs/workspace-issues-and-scheduling.md b/docs/workspace-issues-and-scheduling.md index 23ddb70e2..1704910d4 100644 --- a/docs/workspace-issues-and-scheduling.md +++ b/docs/workspace-issues-and-scheduling.md @@ -129,8 +129,11 @@ The native stream contracts differ materially: Automation reads a debounced `.structured.json` snapshot instead of replaying an entire vendor log. This makes live polling cheap and gives future workbench -orchestration a stable contract independent of CLI versions. Runs created before -this contract are parsed best-effort from the last 2 MB of stdout when opened. +orchestration a stable contract independent of CLI versions. The Runs panel +loads records newest-first in cursor pages (25 initially and 25 older records +on demand), so polling refreshes the active page without repeatedly transferring +the full bounded history. Runs created before this contract are parsed +best-effort from the last 2 MB of stdout when opened. Bounded stdout/stderr diagnostics remain as a fallback. Adapters may discard documented high-frequency transient events before persistence: Pi drops diff --git a/src/webui/routes/headless.spec.ts b/src/webui/routes/headless.spec.ts index b1a8e35bb..4bd507450 100644 --- a/src/webui/routes/headless.spec.ts +++ b/src/webui/routes/headless.spec.ts @@ -16,20 +16,28 @@ const TASKS = [ ] function build(logsDir = '/tmp/openalice-headless-route-test') { - const list = vi.fn((opts: any = {}) => - TASKS.filter( + const list = vi.fn((opts: any = {}) => { + let tasks = TASKS.filter( (t) => (!opts.wsId || t.wsId === opts.wsId) && (!opts.status || t.status === opts.status), - ), - ) + ).slice().reverse() + if (opts.cursor) { + const index = tasks.findIndex((task) => task.taskId === opts.cursor) + tasks = index === -1 ? [] : tasks.slice(index + 1) + } + return opts.limit ? tasks.slice(0, opts.limit) : tasks + }) + const count = vi.fn((opts: any = {}) => TASKS.filter( + (t) => (!opts.wsId || t.wsId === opts.wsId) && (!opts.status || t.status === opts.status), + ).length) const get = vi.fn((id: string) => TASKS.find((t) => t.taskId === id) ?? null) const runningCount = vi.fn(() => TASKS.filter((task) => task.status === 'running').length) const svc = { - headlessTasks: { list, get, runningCount }, + headlessTasks: { list, count, get, runningCount }, headlessCapacity: 8, headlessLogsDir: logsDir, adapters: { get: vi.fn(() => null) }, } as unknown as WorkspaceService - return { app: createHeadlessRoutes(svc), list, get } + return { app: createHeadlessRoutes(svc), list, count, get } } describe('GET /api/headless', () => { @@ -39,13 +47,28 @@ describe('GET /api/headless', () => { expect(r.status).toBe(200) const body = (await r.json()) as any expect(body.tasks.length).toBe(2) + expect(body.page).toEqual({ total: 2, hasMore: false, nextCursor: null }) + expect(body.summary).toEqual({ done: 1, needsAttention: 0 }) expect(body.capacity).toEqual({ running: 1, limit: 8 }) }) it('passes wsId/status/limit filters through to the registry', async () => { const { app, list } = build() await app.request('/?wsId=w1&status=done&limit=5') - expect(list).toHaveBeenCalledWith({ wsId: 'w1', status: 'done', limit: 5 }) + expect(list).toHaveBeenCalledWith({ wsId: 'w1', status: 'done', cursor: undefined, limit: 6 }) + }) + + it('returns a stable next-page cursor', async () => { + const { app } = build() + const first = await app.request('/?limit=1') + const firstBody = (await first.json()) as any + expect(firstBody.tasks.map((task: any) => task.taskId)).toEqual(['t2']) + expect(firstBody.page).toEqual({ total: 2, hasMore: true, nextCursor: 't2' }) + + const second = await app.request('/?limit=1&cursor=t2') + const secondBody = (await second.json()) as any + expect(secondBody.tasks.map((task: any) => task.taskId)).toEqual(['t1']) + expect(secondBody.page).toEqual({ total: 2, hasMore: false, nextCursor: null }) }) it('ignores an invalid status (→ undefined)', async () => { diff --git a/src/webui/routes/headless.ts b/src/webui/routes/headless.ts index 686072dfc..df646171c 100644 --- a/src/webui/routes/headless.ts +++ b/src/webui/routes/headless.ts @@ -58,7 +58,9 @@ async function readStructured(path: string): Promise { const wsId = c.req.query('wsId') || undefined const statusRaw = c.req.query('status') @@ -68,8 +70,32 @@ export function createHeadlessRoutes(svc: WorkspaceService): Hono { : undefined const limitRaw = Number(c.req.query('limit')) const limit = Number.isFinite(limitRaw) && limitRaw > 0 ? Math.min(limitRaw, 500) : 100 + const cursor = c.req.query('cursor') || undefined + const filters = { wsId, status } + const page = svc.headlessTasks.list({ ...filters, cursor, limit: limit + 1 }) + const hasMore = page.length > limit + const tasks = hasMore ? page.slice(0, limit) : page + const total = svc.headlessTasks.count(filters) return c.json({ - tasks: svc.headlessTasks.list({ wsId, status, limit }), + tasks, + page: { + total, + hasMore, + nextCursor: hasMore ? tasks.at(-1)?.taskId ?? null : null, + }, + summary: { + done: + !status || status === 'done' + ? svc.headlessTasks.count({ wsId, status: 'done' }) + : 0, + needsAttention: + (!status || status === 'failed' + ? svc.headlessTasks.count({ wsId, status: 'failed' }) + : 0) + + (!status || status === 'interrupted' + ? svc.headlessTasks.count({ wsId, status: 'interrupted' }) + : 0), + }, capacity: { running: svc.headlessTasks.runningCount(), limit: svc.headlessCapacity, diff --git a/src/workspaces/headless-task-registry.spec.ts b/src/workspaces/headless-task-registry.spec.ts index 5858e9d11..832637865 100644 --- a/src/workspaces/headless-task-registry.spec.ts +++ b/src/workspaces/headless-task-registry.spec.ts @@ -112,6 +112,19 @@ describe('HeadlessTaskRegistry', () => { expect(reloaded.list().every((task) => task.status === 'done' && task.output?.toolCalls === 1)).toBe(true) }) + it('pages newest-first with a stable task cursor', async () => { + const reg = await HeadlessTaskRegistry.load(path, noopLogger) + const oldest = await reg.create({ wsId: 'w1', agent: 'claude', prompt: 'oldest', startedAt: 1 }) + const middle = await reg.create({ wsId: 'w1', agent: 'codex', prompt: 'middle', startedAt: 2 }) + const newest = await reg.create({ wsId: 'w1', agent: 'pi', prompt: 'newest', startedAt: 3 }) + + expect(reg.list({ limit: 2 }).map((task) => task.taskId)).toEqual([newest.taskId, middle.taskId]) + expect(reg.list({ cursor: middle.taskId, limit: 2 }).map((task) => task.taskId)).toEqual([oldest.taskId]) + expect(reg.list({ cursor: 'pruned-task', limit: 2 })).toEqual([]) + expect(reg.count({ wsId: 'w1' })).toBe(3) + expect(reg.count({ status: 'running' })).toBe(3) + }) + it('reconcile-on-boot flips a leftover running task → interrupted', async () => { const reg = await HeadlessTaskRegistry.load(path, noopLogger) await reg.create({ wsId: 'w1', agent: 'codex', prompt: 'x', startedAt: 1 }) // stays running diff --git a/src/workspaces/headless-task-registry.ts b/src/workspaces/headless-task-registry.ts index 22278d625..f2d9a0999 100644 --- a/src/workspaces/headless-task-registry.ts +++ b/src/workspaces/headless-task-registry.ts @@ -176,7 +176,14 @@ export class HeadlessTaskRegistry { /** Records newest-first, optionally filtered. */ list( - opts: { wsId?: string; issueId?: string; status?: HeadlessTaskStatus; limit?: number } = {}, + opts: { + wsId?: string + issueId?: string + status?: HeadlessTaskStatus + /** Return records older than this task in the filtered newest-first view. */ + cursor?: string + limit?: number + } = {}, ): HeadlessTaskRecord[] { let out = this.tasks.filter( (t) => @@ -185,9 +192,30 @@ export class HeadlessTaskRegistry { (!opts.status || t.status === opts.status), ) out = out.slice().reverse() // newest-first + if (opts.cursor) { + const cursorIndex = out.findIndex((task) => task.taskId === opts.cursor) + // A cursor can disappear when the bounded registry prunes old records. + // Returning an empty page is safer than silently restarting at page one + // and duplicating rows in a polling client. + out = cursorIndex === -1 ? [] : out.slice(cursorIndex + 1) + } return opts.limit && opts.limit > 0 ? out.slice(0, opts.limit) : out } + /** Count filtered records without materializing them over the HTTP boundary. */ + count(opts: { wsId?: string; issueId?: string; status?: HeadlessTaskStatus } = {}): number { + return this.tasks.reduce( + (count, task) => count + ( + (!opts.wsId || task.wsId === opts.wsId) && + (!opts.issueId || task.issueId === opts.issueId) && + (!opts.status || task.status === opts.status) + ? 1 + : 0 + ), + 0, + ) + } + runningCount(): number { return this.tasks.reduce((n, t) => (t.status === 'running' ? n + 1 : n), 0) } diff --git a/ui/src/api/headless.ts b/ui/src/api/headless.ts index 8d175e45b..e24c6b366 100644 --- a/ui/src/api/headless.ts +++ b/ui/src/api/headless.ts @@ -50,6 +50,8 @@ export interface HeadlessStructuredOutput { export interface HeadlessListSnapshot { tasks: HeadlessTaskRecord[] + page: { total: number; hasMore: boolean; nextCursor: string | null } + summary: { done: number; needsAttention: number } capacity: { running: number; limit: number } } @@ -70,19 +72,20 @@ export interface HeadlessOutput { export const headlessApi = { async snapshot( - opts: { wsId?: string; status?: HeadlessTaskStatus; limit?: number } = {}, + opts: { wsId?: string; status?: HeadlessTaskStatus; limit?: number; cursor?: string } = {}, ): Promise { const q = new URLSearchParams() if (opts.wsId) q.set('wsId', opts.wsId) if (opts.status) q.set('status', opts.status) if (opts.limit) q.set('limit', String(opts.limit)) + if (opts.cursor) q.set('cursor', opts.cursor) const qs = q.toString() return fetchJson(`/api/headless${qs ? `?${qs}` : ''}`) }, /** List headless runs across all workspaces, newest-first. */ async list( - opts: { wsId?: string; status?: HeadlessTaskStatus; limit?: number } = {}, + opts: { wsId?: string; status?: HeadlessTaskStatus; limit?: number; cursor?: string } = {}, ): Promise { return (await this.snapshot(opts)).tasks }, diff --git a/ui/src/demo/handlers/headless.ts b/ui/src/demo/handlers/headless.ts index 36b00546d..263df523b 100644 --- a/ui/src/demo/handlers/headless.ts +++ b/ui/src/demo/handlers/headless.ts @@ -67,7 +67,15 @@ export const headlessHandlers = [ http.get('/api/headless', ({ request }) => { const wsId = new URL(request.url).searchParams.get('wsId') const tasks = wsId ? demoHeadlessTasks.filter((t) => t.wsId === wsId) : demoHeadlessTasks - return HttpResponse.json({ tasks, capacity: { running: tasks.filter((task) => task.status === 'running').length, limit: 8 } }) + return HttpResponse.json({ + tasks, + page: { total: tasks.length, hasMore: false, nextCursor: null }, + summary: { + done: tasks.filter((task) => task.status === 'done').length, + needsAttention: tasks.filter((task) => task.status === 'failed' || task.status === 'interrupted').length, + }, + capacity: { running: tasks.filter((task) => task.status === 'running').length, limit: 8 }, + }) }), // Path-specific route BEFORE the :taskId catch-all (msw matches in order). http.get('/api/headless/:taskId/output', ({ params }) => { diff --git a/ui/src/pages/AutomationRunsSection.tsx b/ui/src/pages/AutomationRunsSection.tsx index fd47dc1cc..d37d1f3a9 100644 --- a/ui/src/pages/AutomationRunsSection.tsx +++ b/ui/src/pages/AutomationRunsSection.tsx @@ -1,4 +1,4 @@ -import { useCallback, useEffect, useMemo, useState } from 'react' +import { useCallback, useEffect, useState } from 'react' import { Bot, ChevronDown, @@ -30,6 +30,8 @@ const STATUS_STYLE: Record = { interrupted: 'bg-amber-500/15 text-amber-400', } +const RUNS_PAGE_SIZE = 25 + function fmtDuration(ms?: number): string { if (ms == null) return '—' if (ms < 1000) return `${ms}ms` @@ -200,6 +202,7 @@ function SummaryCard({ label, value, detail }: { label: string; value: string; d export function AutomationRunsSection() { const [snapshot, setSnapshot] = useState(null) const [error, setError] = useState(null) + const [loadingMore, setLoadingMore] = useState(false) const [expanded, setExpanded] = useState>(new Set()) const { openHeadlessRun } = useWorkspaces() @@ -212,7 +215,31 @@ export function AutomationRunsSection() { const load = useCallback(async () => { try { - setSnapshot(await api.headless.snapshot({ limit: 100 })) + const fresh = await api.headless.snapshot({ limit: RUNS_PAGE_SIZE }) + setSnapshot((previous) => { + if (!previous) { + return fresh + } + // Poll only the cheap first page, then retain already-loaded older + // pages. Cursor pagination stays stable even when new runs arrive at + // the top between refreshes. + const seen = new Set() + const tasks = [...fresh.tasks, ...previous.tasks].filter((task) => { + if (seen.has(task.taskId)) return false + seen.add(task.taskId) + return true + }).slice(0, fresh.page.total) + const hasMore = tasks.length < fresh.page.total + return { + ...fresh, + tasks, + page: { + ...fresh.page, + hasMore, + nextCursor: hasMore ? tasks.at(-1)?.taskId ?? null : null, + }, + } + }) setError(null) } catch (e) { setError(e instanceof Error ? e.message : String(e)) @@ -225,15 +252,35 @@ export function AutomationRunsSection() { return () => clearInterval(id) }, [load]) - const counts = useMemo(() => { - const tasks = snapshot?.tasks ?? [] - return { - done: tasks.filter((task) => task.status === 'done').length, - failed: tasks.filter((task) => task.status === 'failed' || task.status === 'interrupted').length, + const loadMore = async () => { + const cursor = snapshot?.page.nextCursor + if (!snapshot || !cursor || loadingMore) return + setLoadingMore(true) + try { + const older = await api.headless.snapshot({ limit: RUNS_PAGE_SIZE, cursor }) + setSnapshot((previous) => { + if (!previous) return older + const seen = new Set(previous.tasks.map((task) => task.taskId)) + const tasks = [...previous.tasks, ...older.tasks.filter((task) => !seen.has(task.taskId))] + return { + ...older, + tasks, + page: { + ...older.page, + hasMore: older.page.hasMore, + nextCursor: older.page.hasMore ? tasks.at(-1)?.taskId ?? null : null, + }, + } + }) + setError(null) + } catch (e) { + setError(e instanceof Error ? e.message : String(e)) + } finally { + setLoadingMore(false) } - }, [snapshot]) + } - if (error) return
Failed to load runs: {error}
+ if (error && !snapshot) return
Failed to load runs: {error}
if (!snapshot) { return ( @@ -263,6 +314,11 @@ export function AutomationRunsSection() {
) : (
+ {error && ( +
+ Refresh failed: {error} +
+ )} {snapshot.tasks.map((task) => { const isExpanded = expanded.has(task.taskId) const openable = task.status !== 'running' && !!task.agentSessionId @@ -272,7 +328,11 @@ export function AutomationRunsSection() { ? 'No tools used' : 'Parse on open' return ( -
+
) })} + {snapshot.page.hasMore && ( +
+ + + {snapshot.tasks.length} of {snapshot.page.total} loaded + +
+ )}
)}