feat(automation): normalize headless runtime output

apps/desktop/src/workspace-acceptance-smoke.spec.ts

@@ -14,6 +14,10 @@ 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('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()
   })
 })

apps/desktop/src/workspace-acceptance-smoke.ts

@@ -16,6 +16,8 @@ export interface WorkspaceAcceptanceReceipt {
     readonly allCliManifestsLoaded: boolean
     readonly shellCliRoundTrip: boolean
     readonly managedPiAssistantReply: boolean
+    readonly managedPiStructuredOutput: boolean
+    readonly managedPiDiagnosticCompaction: boolean
     readonly managedPiCliSideEffect: boolean
     readonly cleanupComplete: boolean
   }
@@ -52,6 +54,8 @@ export async function runRendererWorkspaceAcceptanceSmoke(
       allCliManifestsLoaded: false,
       shellCliRoundTrip: false,
       managedPiAssistantReply: false,
+      managedPiStructuredOutput: false,
+      managedPiDiagnosticCompaction: false,
       managedPiCliSideEffect: false,
       cleanupComplete: false,
     }
@@ -66,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)
@@ -206,22 +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 (
+        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(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)) {

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,50 @@ 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.
+
+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. 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
+`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
 
-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 <path> --comments "<summary>"
@@ -140,6 +178,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 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 |
 | `src/workspaces/session-registry.ts` | Durable Session identity and run → Session source index |
@@ -157,6 +201,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 \

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,29 +15,60 @@ const TASKS = [
   { taskId: 't2', wsId: 'w2', agent: 'pi', status: 'running', startedAt: 2 },
 ]
 
-function build() {
-  const list = vi.fn((opts: any = {}) =>
-    TASKS.filter(
+function build(logsDir = '/tmp/openalice-headless-route-test') {
+  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 svc = { headlessTasks: { list, get } } as unknown as WorkspaceService
-  return { app: createHeadlessRoutes(svc), list, get }
+  const runningCount = vi.fn(() => TASKS.filter((task) => task.status === 'running').length)
+  const svc = {
+    headlessTasks: { list, count, get, runningCount },
+    headlessCapacity: 8,
+    headlessLogsDir: logsDir,
+    adapters: { get: vi.fn(() => null) },
+  } as unknown as WorkspaceService
+  return { app: createHeadlessRoutes(svc), list, count, get }
 }
 
 describe('GET /api/headless', () => {
   it('lists tasks', async () => {
     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.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 () => {
@@ -52,4 +88,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 })
+    }
+  })
 })