From aee6d328105db8a4814bd7037fd5b9b125172203 Mon Sep 17 00:00:00 2001 From: Daniel Ochoa Date: Tue, 24 Mar 2026 16:25:54 -0500 Subject: [PATCH 1/5] Symphony: implement plan --- apps/desktop/package.json | 2 +- .../src/server/operations/output-tailer.ts | 186 +++++ .../src/server/operations/symphony-loop.ts | 58 +- .../test/symphony-loop-output-events.test.ts | 686 ++++++++++++++++++ arch/cloud-command-executor.md | 5 + 5 files changed, 914 insertions(+), 23 deletions(-) create mode 100644 apps/desktop/src/server/operations/output-tailer.ts create mode 100644 apps/desktop/test/symphony-loop-output-events.test.ts create mode 100644 arch/cloud-command-executor.md diff --git a/apps/desktop/package.json b/apps/desktop/package.json index 0e1fa85c..45636ab5 100644 --- a/apps/desktop/package.json +++ b/apps/desktop/package.json @@ -1,6 +1,6 @@ { "name": "desktop", - "version": "0.7.1", + "version": "0.7.2", "description": "ClosedLoop Desktop", "author": "ClosedLoop AI ", "private": true, diff --git a/apps/desktop/src/server/operations/output-tailer.ts b/apps/desktop/src/server/operations/output-tailer.ts new file mode 100644 index 00000000..de9557fc --- /dev/null +++ b/apps/desktop/src/server/operations/output-tailer.ts @@ -0,0 +1,186 @@ +import { openSync, readSync, closeSync, existsSync } from "node:fs"; +import { randomUUID } from "node:crypto"; + +export function isRecord(v: unknown): v is Record { + return typeof v === "object" && v !== null && !Array.isArray(v); +} + +function truncate(s: string, n: number): string { + return s.length > n ? s.slice(0, n) + "..." : s; +} + +function redactSensitive(input: string): string { + return input + .replace(/AKIA[A-Z0-9]{16}/g, "[REDACTED]") + .replace(/sk-ant-[A-Za-z0-9\-_]+/g, "[REDACTED]") + .replace(/sk-[A-Za-z0-9]{32,}/g, "[REDACTED]") + .replace(/Bearer [A-Za-z0-9._\-]+/g, "Bearer [REDACTED]") + .replace(/-----BEGIN [A-Z ]+ KEY-----/g, "[REDACTED]"); +} + +export function summarizeJsonlRecord(record: Record): string | null { + if (record.type === "assistant") { + const message = isRecord(record.message) ? record.message : null; + if (message) { + const content = Array.isArray(message.content) ? (message.content as unknown[]) : []; + for (const block of content) { + if (!isRecord(block)) continue; + if (block.type === "tool_use") { + return redactSensitive(`Tool: ${String(block.name ?? "unknown")}`); + } + if (block.type === "text") { + return redactSensitive(truncate(String(block.text ?? ""), 200)); + } + if (block.type === "thinking") { + return redactSensitive("Thinking..."); + } + } + } + return null; + } + + if (record.type === "user") { + const message = isRecord(record.message) ? record.message : null; + if (message) { + const content = Array.isArray(message.content) ? (message.content as unknown[]) : []; + for (const block of content) { + if (!isRecord(block)) continue; + if (block.type === "tool_result") { + if (block.is_error === true) { + return redactSensitive("Tool error"); + } + return redactSensitive("Tool result"); + } + } + } + return null; + } + + if (record.type === "content_block_delta") { + const delta = isRecord(record.delta) ? record.delta : null; + if (delta && delta.type === "text_delta") { + return redactSensitive(truncate(String(delta.text ?? ""), 200)); + } + return null; + } + + if (record.type === "result") { + if (record.subtype === "success") { + return redactSensitive("Turn complete"); + } + if (record.subtype === "error" || record.is_error === true) { + return redactSensitive( + `Error: ${truncate(String(record.result ?? record.error ?? ""), 200)}` + ); + } + return null; + } + + return null; +} + +// --------------------------------------------------------------------------- +// API communication +// --------------------------------------------------------------------------- + +async function postLoopEvent( + apiBaseUrl: string, + loopId: string, + token: string, + event: { type: string; data: { text: string } } +): Promise { + try { + await fetch(`${apiBaseUrl}/loops/${loopId}/events`, { + method: "POST", + headers: { + "Authorization": `Bearer ${token}`, + "Content-Type": "application/json", + "x-loop-event-nonce": randomUUID(), + }, + body: JSON.stringify({ + type: event.type, + data: { text: event.data.text }, + timestamp: new Date().toISOString(), + }), + }); + } catch (err) { + console.error("[output-tailer] Failed to post loop event:", err); + } +} + +// --------------------------------------------------------------------------- +// Output tailer +// --------------------------------------------------------------------------- + +export function startOutputTailer( + jsonlPath: string, + apiBaseUrl: string, + loopId: string, + token: string, + initialByteOffset: number +): { stop: () => void; flush: () => Promise } { + let stopped = false; + let byteOffset = initialByteOffset; + let pendingRemainder = Buffer.alloc(0); + let lastSentAt: number | null = null; + + async function pollOnce(): Promise { + if (stopped) return; + if (!existsSync(jsonlPath)) return; + let fd: number | null = null; + try { + fd = openSync(jsonlPath, "r"); + const chunkSize = 65536; + const chunk = Buffer.alloc(chunkSize); + let bytesRead: number; + while ((bytesRead = readSync(fd, chunk, 0, chunkSize, byteOffset)) > 0) { + byteOffset += bytesRead; + pendingRemainder = Buffer.concat([pendingRemainder, chunk.subarray(0, bytesRead)]); + } + } catch { + return; + } finally { + if (fd !== null) closeSync(fd); + } + + const newlineIndex = pendingRemainder.lastIndexOf(10); // 0x0a = newline + if (newlineIndex === -1) return; + const completeLines = pendingRemainder.subarray(0, newlineIndex).toString("utf8"); + pendingRemainder = pendingRemainder.subarray(newlineIndex + 1); + + let lastDisplay: string | null = null; + for (const line of completeLines.split("\n")) { + const trimmed = line.trim(); + if (!trimmed) continue; + let parsed: unknown; + try { + parsed = JSON.parse(trimmed); + } catch { + continue; + } + if (!isRecord(parsed)) continue; + const display = summarizeJsonlRecord(parsed); + if (!display) continue; + lastDisplay = display; + } + + if (lastDisplay !== null) { + const now = Date.now(); + if (lastSentAt === null || now - lastSentAt >= 5000) { + lastSentAt = now; + await postLoopEvent(apiBaseUrl, loopId, token, { type: "output", data: { text: lastDisplay } }); + } + } + } + + const intervalId = setInterval(() => { pollOnce().catch(() => {}); }, 2000); + + return { + stop: () => { stopped = true; clearInterval(intervalId); }, + flush: async () => { + clearInterval(intervalId); + await pollOnce(); + stopped = true; + }, + }; +} diff --git a/apps/desktop/src/server/operations/symphony-loop.ts b/apps/desktop/src/server/operations/symphony-loop.ts index 8e598f45..f1300e0e 100644 --- a/apps/desktop/src/server/operations/symphony-loop.ts +++ b/apps/desktop/src/server/operations/symphony-loop.ts @@ -1,6 +1,6 @@ import { execSync, spawn } from "node:child_process"; import crypto from "node:crypto"; -import { closeSync, existsSync, openSync, readFileSync } from "node:fs"; +import { closeSync, existsSync, openSync, readFileSync, statSync } from "node:fs"; import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; @@ -21,6 +21,7 @@ import { resolveWorktreeParentDir, tryAssertRepoAllowed, } from "./symphony-utils.js"; +import { startOutputTailer } from "./output-tailer.js"; // --------------------------------------------------------------------------- // Types @@ -189,12 +190,15 @@ function buildClaudePipeline( return { cmd: "bash", args: ["-c", pipeline] }; } - // No formatter — run claude directly (raw stream-json to stdout) - if (stdinFile) { - return { cmd: "bash", args: ["-c", claudeCmd] }; - } - return { cmd: "claude", args: claudeArgs }; + // No formatter — wrap in bash pipeline so grep|tee still writes claude-output.jsonl + const pipeline = [ + `${claudeCmd} 2>${shellEscape(stderrFile)}`, + "grep --line-buffered '^{'", + `tee -a ${shellEscape(jsonlFile)}`, + ].join(" | "); + return { cmd: "bash", args: ["-c", pipeline] }; } + /** Find the local repo path for a given fullName (e.g. "org/repo"). */ function findLocalRepo( fullName: string, @@ -1490,31 +1494,34 @@ async function handleLoopRequest( } closeSync(logFd); + const tailerJsonlPath = path.join(claudeWorkDir, "claude-output.jsonl"); + const jsonlPreSpawnOffset = existsSync(tailerJsonlPath) ? statSync(tailerJsonlPath).size : 0; + // Guard against double-firing: both 'error' and 'exit' can emit. let completionHandled = false; - const onceComplete = (code: number) => { - if (completionHandled) { - return; - } + let stopTailer: { stop: () => void; flush: () => Promise } = { + stop: () => {}, + flush: () => Promise.resolve(), + }; + const onceComplete = async (code: number): Promise => { + if (completionHandled) return; completionHandled = true; loopLog(body.loopId, `onceComplete fired, code=${code}`); - handleProcessCompletion( - code, - body, - apiBaseUrl, - worktreeDir, - claudeWorkDir, - usedTempDir, - expandedRepoPath, - jobStore - ).catch((err) => loopError(body.loopId, "Completion handler error:", err)); + try { + await stopTailer.flush(); + await handleProcessCompletion( + code, body, apiBaseUrl, worktreeDir, claudeWorkDir, usedTempDir, expandedRepoPath, jobStore + ); + } catch (err) { + loopError(body.loopId, "Completion handler error:", err); + } }; // Prevent unhandled 'error' events (e.g. ENOENT if binary vanishes // between pre-flight check and spawn) from crashing Electron. child.on("error", (err) => { loopError(body.loopId, "Spawn error:", err.message); - onceComplete(1); + void onceComplete(1); }); // Use 'exit' instead of 'close' — with detached processes using @@ -1522,7 +1529,7 @@ async function handleLoopRequest( // because there are no Node.js streams to track closure of. child.on("exit", (code) => { loopLog(body.loopId, `Process exit event, code=${code}`); - onceComplete(code ?? 1); + void onceComplete(code ?? 1); }); const pid = child.pid ?? null; @@ -1536,6 +1543,13 @@ async function handleLoopRequest( // Replace sentinel with real entry — storing `child` prevents GC of the // ChildProcess handle which would silently drop the exit listener. runningLoops.set(body.loopId, { pid, child }); + stopTailer = startOutputTailer( + tailerJsonlPath, + apiBaseUrl, + body.loopId, + body.closedLoopAuthToken, + jsonlPreSpawnOffset + ); spawnedSuccessfully = true; loopLog(body.loopId, `Spawned pid=${pid}, worktree=${worktreeDir}`); diff --git a/apps/desktop/test/symphony-loop-output-events.test.ts b/apps/desktop/test/symphony-loop-output-events.test.ts new file mode 100644 index 00000000..aca7a4a3 --- /dev/null +++ b/apps/desktop/test/symphony-loop-output-events.test.ts @@ -0,0 +1,686 @@ +import assert from "node:assert/strict"; +import { mkdirSync, writeFileSync } from "node:fs"; +import fs from "node:fs/promises"; +import http from "node:http"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, describe, test } from "node:test"; +import { summarizeJsonlRecord, startOutputTailer } from "../src/server/operations/output-tailer.js"; +import { DesktopGatewayServer } from "../src/server/server.js"; +import { EMPTY_CAPABILITIES, PORT_PROBE_ORDER } from "../src/shared/contracts.js"; + +// --------------------------------------------------------------------------- +// Shared cleanup state +// --------------------------------------------------------------------------- + +const tempPathsToClean: string[] = []; +const serversToClose: DesktopGatewayServer[] = []; +const eventServersToClose: http.Server[] = []; +const originalPath = process.env.PATH; +const originalHome = process.env.HOME; + +// NOTE: We do NOT set CLOSEDLOOP_SYMPHONY_TEST_RAW_CLAUDE_PIPELINE globally. +// Tests that need the raw pipeline (no formatter) set it per-test. + +afterEach(async () => { + // Restore PATH + if (originalPath === undefined) { + delete process.env.PATH; + } else { + process.env.PATH = originalPath; + } + + // Restore HOME + if (originalHome === undefined) { + delete process.env.HOME; + } else { + process.env.HOME = originalHome; + } + + // Restore raw pipeline env var (tests may set it individually) + delete process.env.CLOSEDLOOP_SYMPHONY_TEST_RAW_CLAUDE_PIPELINE; + + for (const server of serversToClose.splice(0)) { + await server.stop(); + } + + for (const srv of eventServersToClose.splice(0)) { + await new Promise((resolve, reject) => { + srv.close((err) => { + if (err) { + reject(err); + } else { + resolve(); + } + }); + }); + } + + for (const p of tempPathsToClean.splice(0)) { + await fs.rm(p, { recursive: true, force: true }); + } +}); + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +function makeTempDir(): string { + const dir = path.join( + os.tmpdir(), + `output-events-test-${Date.now()}-${Math.random().toString(36).slice(2)}` + ); + mkdirSync(dir, { recursive: true }); + tempPathsToClean.push(dir); + return dir; +} + +function makeGatewayServer(options?: { + allowedDirs?: string[]; + tmpDir?: string; + getApiOrigin?: () => string; +}): DesktopGatewayServer { + const tmpDir = options?.tmpDir ?? makeTempDir(); + const server = new DesktopGatewayServer({ + host: "127.0.0.1", + preferredPort: PORT_PROBE_ORDER[0], + fallbackPorts: PORT_PROBE_ORDER.slice(1), + webAppOrigin: "https://app.symphony.com", + getGatewayAuthToken: () => "test-token", + getApiOrigin: options?.getApiOrigin ?? (() => "http://127.0.0.1:49152"), + getAllowedDirectories: () => options?.allowedDirs ?? [os.tmpdir()], + machineName: "output-events-test-machine", + version: "0.1.0-test", + capabilities: EMPTY_CAPABILITIES, + discoveryFilePath: path.join(tmpDir, "electron-port"), + }); + serversToClose.push(server); + return server; +} + +/** + * Start an event-capture HTTP server on a random port (port 0). + * The `collected` array tracks events in sequence order. + * Pass `outputStatusCode` to simulate the server returning non-200 for output events. + */ +async function startEventServer(options?: { outputStatusCode?: number }): Promise<{ + port: number; + getCollected: () => Array<{ seq: number; type: string } & Record>; + waitForEvent: ( + predicate: (body: Record) => boolean, + timeoutMs?: number + ) => Promise>; +}> { + const collected: Array<{ seq: number; type: string } & Record> = []; + const waiters: Array<{ + predicate: (b: Record) => boolean; + resolve: (b: Record) => void; + reject: (e: Error) => void; + }> = []; + + const server = http.createServer((req, res) => { + let raw = ""; + req.on("data", (chunk: Buffer) => { + raw += chunk.toString(); + }); + req.on("end", () => { + let body: Record; + try { + body = JSON.parse(raw) as Record; + } catch { + body = {}; + } + + // Decide status code + const isOutputEvent = typeof body.type === "string" && body.type === "output"; + const statusCode = + isOutputEvent && options?.outputStatusCode !== undefined + ? options.outputStatusCode + : 200; + + res.statusCode = statusCode; + res.end("{}"); + + // Track the event with sequence number + const entry = { seq: collected.length, type: String(body.type ?? ""), ...body }; + collected.push(entry); + + for (let i = waiters.length - 1; i >= 0; i--) { + const waiter = waiters[i]; + if (waiter.predicate(entry)) { + waiters.splice(i, 1); + waiter.resolve(entry); + } + } + }); + }); + + const port = await new Promise((resolve, reject) => { + server.listen(0, "127.0.0.1", () => { + const addr = server.address(); + if (!addr || typeof addr === "string") { + reject(new Error("Could not get server address")); + return; + } + resolve(addr.port); + }); + server.once("error", reject); + }); + + eventServersToClose.push(server); + + const waitForEvent = ( + predicate: (b: Record) => boolean, + timeoutMs = 10_000 + ) => { + const existing = collected.find(predicate); + if (existing) { + return Promise.resolve(existing); + } + return new Promise>((resolve, reject) => { + const timer = setTimeout(() => { + const idx = waiters.findIndex((w) => w.resolve === resolve); + if (idx !== -1) { + waiters.splice(idx, 1); + } + reject( + new Error( + `waitForEvent timed out after ${timeoutMs}ms. Collected so far: ${JSON.stringify(collected)}` + ) + ); + }, timeoutMs); + + waiters.push({ + predicate, + resolve: (b) => { + clearTimeout(timer); + resolve(b); + }, + reject, + }); + }); + }; + + return { + port, + getCollected: () => collected, + waitForEvent, + }; +} + +/** Build a valid EVALUATE_PRD request body. */ +function buildLoopBody(overrides?: Partial>): Record { + return { + loopId: "aaaaaaaa-0000-0000-0000-000000000001", + command: "EVALUATE_PRD", + closedLoopAuthToken: "cl-token", + apiBaseUrl: "https://api.example.com", + artifacts: [], + ...overrides, + }; +} + +// --------------------------------------------------------------------------- +// T-2.3: Unit tests for summarizeJsonlRecord +// --------------------------------------------------------------------------- + +describe("T-2.3: summarizeJsonlRecord", () => { + test("assistant/text: returns text content", () => { + const record = { + type: "assistant", + message: { content: [{ type: "text", text: "hello world" }] }, + }; + assert.equal(summarizeJsonlRecord(record), "hello world"); + }); + + test("assistant/text truncation: 201-char text ends with '...' and length 203", () => { + const longText = "a".repeat(201); + const record = { + type: "assistant", + message: { content: [{ type: "text", text: longText }] }, + }; + const result = summarizeJsonlRecord(record); + assert.ok(result !== null, "result should not be null"); + assert.ok(result!.endsWith("..."), `Expected result to end with '...', got: ${result}`); + assert.equal(result!.length, 203); + }); + + test("assistant/tool_use: returns 'Tool: '", () => { + const record = { + type: "assistant", + message: { content: [{ type: "tool_use", name: "Read" }] }, + }; + assert.equal(summarizeJsonlRecord(record), "Tool: Read"); + }); + + test("assistant/thinking: returns 'Thinking...'", () => { + const record = { + type: "assistant", + message: { content: [{ type: "thinking" }] }, + }; + assert.equal(summarizeJsonlRecord(record), "Thinking..."); + }); + + test("user/tool_result success: returns 'Tool result'", () => { + const record = { + type: "user", + message: { content: [{ type: "tool_result" }] }, + }; + assert.equal(summarizeJsonlRecord(record), "Tool result"); + }); + + test("user/tool_result error: returns 'Tool error'", () => { + const record = { + type: "user", + message: { content: [{ type: "tool_result", is_error: true }] }, + }; + assert.equal(summarizeJsonlRecord(record), "Tool error"); + }); + + test("content_block_delta/text_delta: returns delta text", () => { + const record = { + type: "content_block_delta", + delta: { type: "text_delta", text: "hi" }, + }; + assert.equal(summarizeJsonlRecord(record), "hi"); + }); + + test("result/success: returns 'Turn complete'", () => { + const record = { type: "result", subtype: "success", result: "", is_error: false }; + assert.equal(summarizeJsonlRecord(record), "Turn complete"); + }); + + test("result/error: returns 'Error: '", () => { + const record = { type: "result", subtype: "error", result: "oops", is_error: true }; + assert.equal(summarizeJsonlRecord(record), "Error: oops"); + }); + + test("unknown type: returns null", () => { + const record = { type: "unknown_type_xyz" }; + assert.equal(summarizeJsonlRecord(record), null); + }); + + test("redaction: sensitive key is replaced with [REDACTED]", () => { + const sensitiveText = "Here is my key: sk-ant-abc123xyz"; + const record = { + type: "assistant", + message: { content: [{ type: "text", text: sensitiveText }] }, + }; + const result = summarizeJsonlRecord(record); + assert.ok(result !== null, "result should not be null"); + assert.ok(result!.includes("[REDACTED]"), `Expected [REDACTED] in result, got: ${result}`); + assert.ok(!result!.includes("sk-ant-abc123xyz"), "Result should not contain original key"); + }); +}); + +// --------------------------------------------------------------------------- +// T-5.2: Output events arrive before completed +// --------------------------------------------------------------------------- + +describe("T-5.2: Output events arrive before completed event", () => { + test("(a) happy path: output event seq < completed event seq", async () => { + // This test uses CLOSEDLOOP_SYMPHONY_TEST_RAW_CLAUDE_PIPELINE=1 so the bash + // pipeline writes the stub's JSON lines directly to the jsonl file. + process.env.CLOSEDLOOP_SYMPHONY_TEST_RAW_CLAUDE_PIPELINE = "1"; + + const tmpDir = makeTempDir(); + const fakeBin = path.join(tmpDir, "fake-bin"); + await fs.mkdir(fakeBin, { recursive: true }); + + const eventSrv = await startEventServer(); + const apiBaseUrl = `http://127.0.0.1:${eventSrv.port}`; + + // Stub claude: emit an assistant text line then a result/success line + const stubScript = [ + "#!/bin/sh", + `echo '{"type":"assistant","message":{"content":[{"type":"text","text":"doing work"}]}}'`, + `echo '{"type":"result","subtype":"success","result":"","is_error":false}'`, + "exit 0", + ].join("\n"); + await fs.writeFile(path.join(fakeBin, "claude"), stubScript, { mode: 0o755 }); + process.env.PATH = `${fakeBin}:/usr/bin:/bin`; + + const loopId = "bbbbbbbb-0000-0000-0000-000000000001"; + const server = makeGatewayServer({ + allowedDirs: [tmpDir], + getApiOrigin: () => apiBaseUrl, + }); + await server.start(); + + const response = await fetch( + `http://127.0.0.1:${server.getActivePort()}/api/engineer/symphony/loop`, + { + method: "POST", + headers: { + "content-type": "application/json", + "x-desktop-gateway-token": "test-token", + }, + body: JSON.stringify( + buildLoopBody({ loopId, apiBaseUrl }) + ), + } + ); + + assert.equal(response.status, 200, `Expected 200, got ${response.status}`); + + // Wait for completed event + await eventSrv.waitForEvent( + (b) => b.type === "completed" || b.type === "error", + 15_000 + ); + + const collected = eventSrv.getCollected(); + const outputEvent = collected.find((e) => e.type === "output"); + const completedEvent = collected.find((e) => e.type === "completed" || e.type === "error"); + + assert.ok(outputEvent !== undefined, "Expected at least one output event"); + assert.ok(completedEvent !== undefined, "Expected a completed event"); + assert.ok( + outputEvent!.seq < completedEvent!.seq, + `Output event seq (${outputEvent!.seq}) should be less than completed seq (${completedEvent!.seq})` + ); + }); + + test("(b) JSONL absent before spawn: tailer handles non-existent file gracefully", async () => { + // Start the tailer against a non-existent file and flush — should not throw + const tmpDir = makeTempDir(); + const nonExistentJsonl = path.join(tmpDir, "does-not-exist.jsonl"); + + const eventSrv = await startEventServer(); + const apiBaseUrl = `http://127.0.0.1:${eventSrv.port}`; + + const tailer = startOutputTailer(nonExistentJsonl, apiBaseUrl, "test-loop-id", "token", 0); + await assert.doesNotReject(() => tailer.flush()); + + // No output events should be posted since file doesn't exist + const collected = eventSrv.getCollected(); + const outputEvents = collected.filter((e) => e.type === "output"); + assert.equal(outputEvents.length, 0, "No output events expected for non-existent file"); + }); + + test("(c) HTTP 500 for output events: loop completes even when event server returns 500", async () => { + process.env.CLOSEDLOOP_SYMPHONY_TEST_RAW_CLAUDE_PIPELINE = "1"; + + const tmpDir = makeTempDir(); + const fakeBin = path.join(tmpDir, "fake-bin"); + await fs.mkdir(fakeBin, { recursive: true }); + + // Event server returns 500 for output events + const eventSrv = await startEventServer({ outputStatusCode: 500 }); + const apiBaseUrl = `http://127.0.0.1:${eventSrv.port}`; + + const stubScript = [ + "#!/bin/sh", + `echo '{"type":"assistant","message":{"content":[{"type":"text","text":"working"}]}}'`, + `echo '{"type":"result","subtype":"success","result":"","is_error":false}'`, + "exit 0", + ].join("\n"); + await fs.writeFile(path.join(fakeBin, "claude"), stubScript, { mode: 0o755 }); + process.env.PATH = `${fakeBin}:/usr/bin:/bin`; + + const loopId = "cccccccc-0000-0000-0000-000000000001"; + const server = makeGatewayServer({ + allowedDirs: [tmpDir], + getApiOrigin: () => apiBaseUrl, + }); + await server.start(); + + const response = await fetch( + `http://127.0.0.1:${server.getActivePort()}/api/engineer/symphony/loop`, + { + method: "POST", + headers: { + "content-type": "application/json", + "x-desktop-gateway-token": "test-token", + }, + body: JSON.stringify( + buildLoopBody({ loopId, apiBaseUrl }) + ), + } + ); + + assert.equal(response.status, 200, `Expected 200, got ${response.status}`); + + // Loop should still complete even when output event POSTs return 500 + const completedEvent = await eventSrv.waitForEvent( + (b) => b.type === "completed" || b.type === "error", + 15_000 + ); + + assert.ok( + completedEvent.type === "completed" || completedEvent.type === "error", + `Expected completed or error event, got: ${JSON.stringify(completedEvent)}` + ); + }); +}); + +// --------------------------------------------------------------------------- +// T-5.3: Partial JSONL writes +// --------------------------------------------------------------------------- + +describe("T-5.3: Partial JSONL writes", () => { + test("incomplete line does not emit; completed line emits one event", async () => { + const tmpDir = makeTempDir(); + const jsonlPath = path.join(tmpDir, "claude-output.jsonl"); + + const eventSrv = await startEventServer(); + const apiBaseUrl = `http://127.0.0.1:${eventSrv.port}`; + + const tailer = startOutputTailer(jsonlPath, apiBaseUrl, "partial-test-loop", "token", 0); + + // Write an incomplete line (no trailing newline) + const incompleteLine = '{"type":"assistant","message":{"content":[{"type":"text","text":"hel'; + writeFileSync(jsonlPath, incompleteLine); + + // Flush: no complete lines yet — 0 events + await tailer.flush(); + // Re-create tailer since flush() stops it + const tailer2 = startOutputTailer(jsonlPath, apiBaseUrl, "partial-test-loop", "token", 0); + + const collectedBeforeComplete = eventSrv.getCollected().filter((e) => e.type === "output"); + assert.equal( + collectedBeforeComplete.length, + 0, + `Expected 0 output events for incomplete line, got ${collectedBeforeComplete.length}` + ); + + // Append the rest of the line to complete it + const rest = 'lo"}]}}\n'; + writeFileSync(jsonlPath, incompleteLine + rest); + + // Flush the second tailer: now 1 complete line — 1 event + await tailer2.flush(); + + const collectedAfterComplete = eventSrv.getCollected().filter((e) => e.type === "output"); + assert.equal( + collectedAfterComplete.length, + 1, + `Expected 1 output event after completing the line, got ${collectedAfterComplete.length}` + ); + }); +}); + +// --------------------------------------------------------------------------- +// T-5.4: Flush on exit +// --------------------------------------------------------------------------- + +describe("T-5.4: Flush on exit", () => { + test( + "final-before-exit output event arrives before completed event", + { timeout: 20_000 }, + async () => { + process.env.CLOSEDLOOP_SYMPHONY_TEST_RAW_CLAUDE_PIPELINE = "1"; + + const tmpDir = makeTempDir(); + const fakeBin = path.join(tmpDir, "fake-bin"); + await fs.mkdir(fakeBin, { recursive: true }); + + const eventSrv = await startEventServer(); + const apiBaseUrl = `http://127.0.0.1:${eventSrv.port}`; + + // Stub claude: emit first line, sleep 3s, emit second line, exit + // The sleep ensures the second line arrives after the tailer's first poll, + // exercising the flush-on-exit path. + const stubScript = [ + "#!/bin/sh", + `echo '{"type":"assistant","message":{"content":[{"type":"text","text":"first message"}]}}'`, + "sleep 3", + `echo '{"type":"result","subtype":"success","result":"","is_error":false}'`, + "exit 0", + ].join("\n"); + await fs.writeFile(path.join(fakeBin, "claude"), stubScript, { mode: 0o755 }); + process.env.PATH = `${fakeBin}:/usr/bin:/bin`; + + const loopId = "dddddddd-0000-0000-0000-000000000001"; + const server = makeGatewayServer({ + allowedDirs: [tmpDir], + getApiOrigin: () => apiBaseUrl, + }); + await server.start(); + + const response = await fetch( + `http://127.0.0.1:${server.getActivePort()}/api/engineer/symphony/loop`, + { + method: "POST", + headers: { + "content-type": "application/json", + "x-desktop-gateway-token": "test-token", + }, + body: JSON.stringify( + buildLoopBody({ loopId, apiBaseUrl }) + ), + } + ); + + assert.equal(response.status, 200, `Expected 200, got ${response.status}`); + + // Wait for the completed event + await eventSrv.waitForEvent( + (b) => b.type === "completed" || b.type === "error", + 18_000 + ); + + const collected = eventSrv.getCollected(); + const outputEvent = collected.find((e) => e.type === "output"); + const completedEvent = collected.find( + (e) => e.type === "completed" || e.type === "error" + ); + + assert.ok(outputEvent !== undefined, "Expected at least one output event"); + assert.ok(completedEvent !== undefined, "Expected a completed event"); + assert.ok( + outputEvent!.seq < completedEvent!.seq, + `Output event (seq=${outputEvent!.seq}) should arrive before completed (seq=${completedEvent!.seq})` + ); + } + ); +}); + +// --------------------------------------------------------------------------- +// T-5.5: No-formatter fallback +// --------------------------------------------------------------------------- + +describe("T-5.5: No-formatter fallback", () => { + test("JSONL file is created and output events received when formatter is absent", async () => { + // Override HOME to an empty temp dir so getPluginCacheRoot() -> ~/.claude/plugins/cache/... + // resolves to a non-existent path, making findStreamFormatter() return null. + // Do NOT set CLOSEDLOOP_SYMPHONY_TEST_RAW_CLAUDE_PIPELINE; we want the real + // code path that calls findStreamFormatter() and falls back to the raw pipeline. + const tmpDir = makeTempDir(); + const fakeHome = path.join(tmpDir, "fake-home"); + await fs.mkdir(fakeHome, { recursive: true }); + process.env.HOME = fakeHome; + + const fakeBin = path.join(tmpDir, "fake-bin"); + await fs.mkdir(fakeBin, { recursive: true }); + + const eventSrv = await startEventServer(); + const apiBaseUrl = `http://127.0.0.1:${eventSrv.port}`; + + const stubScript = [ + "#!/bin/sh", + `echo '{"type":"assistant","message":{"content":[{"type":"text","text":"no formatter output"}]}}'`, + `echo '{"type":"result","subtype":"success","result":"","is_error":false}'`, + "exit 0", + ].join("\n"); + await fs.writeFile(path.join(fakeBin, "claude"), stubScript, { mode: 0o755 }); + // Include system paths so 'bash', 'grep', 'tee' are available for the pipeline + process.env.PATH = `${fakeBin}:/usr/bin:/bin:/usr/local/bin:/opt/homebrew/bin`; + + const loopId = "eeeeeeee-0000-0000-0000-000000000001"; + const server = makeGatewayServer({ + allowedDirs: [tmpDir], + getApiOrigin: () => apiBaseUrl, + }); + await server.start(); + + const response = await fetch( + `http://127.0.0.1:${server.getActivePort()}/api/engineer/symphony/loop`, + { + method: "POST", + headers: { + "content-type": "application/json", + "x-desktop-gateway-token": "test-token", + }, + body: JSON.stringify( + buildLoopBody({ loopId, apiBaseUrl }) + ), + } + ); + + assert.equal(response.status, 200, `Expected 200, got ${response.status}`); + + // Loop completes and output events arrive + await eventSrv.waitForEvent( + (b) => b.type === "completed" || b.type === "error", + 15_000 + ); + + // The JSONL file may already be cleaned up by handleProcessCompletion's fire-and-forget rm. + // We verify output events arrived, which proves the JSONL file was created and processed. + const outputEvents = eventSrv.getCollected().filter((e) => e.type === "output"); + assert.ok( + outputEvents.length > 0, + `Expected output events with no-formatter fallback, but got none. Collected: ${JSON.stringify(eventSrv.getCollected())}` + ); + assert.ok( + typeof outputEvents[0].data === "object" && + outputEvents[0].data !== null && + typeof (outputEvents[0].data as Record).text === "string", + `Expected output event to have data.text, got: ${JSON.stringify(outputEvents[0])}` + ); + }); +}); + +// --------------------------------------------------------------------------- +// T-5.6: Throttle +// --------------------------------------------------------------------------- + +describe("T-5.6: Throttle", () => { + test("5 rapid JSON lines produce at most 2 output events (throttle window)", async () => { + const tmpDir = makeTempDir(); + const jsonlPath = path.join(tmpDir, "claude-output.jsonl"); + + const eventSrv = await startEventServer(); + const apiBaseUrl = `http://127.0.0.1:${eventSrv.port}`; + + // Write 5 complete JSON lines with text content at once (simulating rapid output) + const lines = [ + '{"type":"assistant","message":{"content":[{"type":"text","text":"line 1"}]}}', + '{"type":"assistant","message":{"content":[{"type":"text","text":"line 2"}]}}', + '{"type":"assistant","message":{"content":[{"type":"text","text":"line 3"}]}}', + '{"type":"assistant","message":{"content":[{"type":"text","text":"line 4"}]}}', + '{"type":"assistant","message":{"content":[{"type":"text","text":"line 5"}]}}', + ]; + writeFileSync(jsonlPath, lines.join("\n") + "\n"); + + const tailer = startOutputTailer(jsonlPath, apiBaseUrl, "throttle-test-loop", "token", 0); + await tailer.flush(); + + const outputEvents = eventSrv.getCollected().filter((e) => e.type === "output"); + assert.ok( + outputEvents.length <= 2, + `Expected at most 2 output events due to throttle, got ${outputEvents.length}: ${JSON.stringify(outputEvents)}` + ); + }); +}); diff --git a/arch/cloud-command-executor.md b/arch/cloud-command-executor.md new file mode 100644 index 00000000..fc5fc9a7 --- /dev/null +++ b/arch/cloud-command-executor.md @@ -0,0 +1,5 @@ +# Cloud Command Executor Architecture + +Not applicable -- this feature does not require changes to the command execution layer. + +**Rationale**: The feature adds a file-tailing side channel in `symphony-loop.ts` that reads `claude-output.jsonl` and posts `output` events via the existing `postLoopEvent()` helper; no cloud-dispatched command routing, queue scheduling, lock-key serialization, cancel/timeout handling, replay buffering, or retention pruning is affected. From d3d725f2fb2154d681708720afdafa372eb596de Mon Sep 17 00:00:00 2001 From: "daniel.ochoa" Date: Tue, 24 Mar 2026 17:04:29 -0500 Subject: [PATCH 2/5] PLAN-73: Bump desktop version to 0.8.2 - Patch version bump for desktop changes in this branch Testing: typecheck passes Risks: none --- apps/desktop/package.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/desktop/package.json b/apps/desktop/package.json index d71f2b83..ad9bbf1d 100644 --- a/apps/desktop/package.json +++ b/apps/desktop/package.json @@ -1,6 +1,6 @@ { "name": "desktop", - "version": "0.8.1", + "version": "0.8.2", "description": "ClosedLoop Desktop", "author": "ClosedLoop AI ", "private": true, From 9928ea971f88e754d63ae15fdb915c76db5e66e7 Mon Sep 17 00:00:00 2001 From: "daniel.ochoa" Date: Wed, 25 Mar 2026 11:41:41 -0500 Subject: [PATCH 3/5] PLAN-73: Fix output event field name mismatch - Rename text to chunk in postLoopEvent payload to match the API LoopEventOutput type contract - Update test assertion to check for data.chunk Testing: Ran full test suite (18/18 pass), lint, typecheck Risks: None identified --- apps/desktop/src/server/operations/output-tailer.ts | 6 +++--- apps/desktop/test/symphony-loop-output-events.test.ts | 4 ++-- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/apps/desktop/src/server/operations/output-tailer.ts b/apps/desktop/src/server/operations/output-tailer.ts index de9557fc..15801fcf 100644 --- a/apps/desktop/src/server/operations/output-tailer.ts +++ b/apps/desktop/src/server/operations/output-tailer.ts @@ -87,7 +87,7 @@ async function postLoopEvent( apiBaseUrl: string, loopId: string, token: string, - event: { type: string; data: { text: string } } + event: { type: string; data: { chunk: string } } ): Promise { try { await fetch(`${apiBaseUrl}/loops/${loopId}/events`, { @@ -99,7 +99,7 @@ async function postLoopEvent( }, body: JSON.stringify({ type: event.type, - data: { text: event.data.text }, + data: { chunk: event.data.chunk }, timestamp: new Date().toISOString(), }), }); @@ -168,7 +168,7 @@ export function startOutputTailer( const now = Date.now(); if (lastSentAt === null || now - lastSentAt >= 5000) { lastSentAt = now; - await postLoopEvent(apiBaseUrl, loopId, token, { type: "output", data: { text: lastDisplay } }); + await postLoopEvent(apiBaseUrl, loopId, token, { type: "output", data: { chunk: lastDisplay } }); } } } diff --git a/apps/desktop/test/symphony-loop-output-events.test.ts b/apps/desktop/test/symphony-loop-output-events.test.ts index aca7a4a3..72be355e 100644 --- a/apps/desktop/test/symphony-loop-output-events.test.ts +++ b/apps/desktop/test/symphony-loop-output-events.test.ts @@ -646,8 +646,8 @@ describe("T-5.5: No-formatter fallback", () => { assert.ok( typeof outputEvents[0].data === "object" && outputEvents[0].data !== null && - typeof (outputEvents[0].data as Record).text === "string", - `Expected output event to have data.text, got: ${JSON.stringify(outputEvents[0])}` + typeof (outputEvents[0].data as Record).chunk === "string", + `Expected output event to have data.chunk, got: ${JSON.stringify(outputEvents[0])}` ); }); }); From b6b2a89d1ecc47a30d9480324af54e941e86ba00 Mon Sep 17 00:00:00 2001 From: "daniel.ochoa" Date: Wed, 25 Mar 2026 11:44:34 -0500 Subject: [PATCH 4/5] PLAN-73: Add typed definitions for JSONL record shapes - Define ContentBlock, AssistantRecord, UserRecord, ContentBlockDeltaRecord, and ResultRecord types - Export JsonlRecord discriminated union - Refactor summarizeJsonlRecord to use switch/case over the typed union instead of if-chains on untyped records Testing: All 18 tests pass, lint and typecheck clean Risks: None identified --- .../src/server/operations/output-tailer.ts | 122 +++++++++++------- 1 file changed, 73 insertions(+), 49 deletions(-) diff --git a/apps/desktop/src/server/operations/output-tailer.ts b/apps/desktop/src/server/operations/output-tailer.ts index 15801fcf..070cf11f 100644 --- a/apps/desktop/src/server/operations/output-tailer.ts +++ b/apps/desktop/src/server/operations/output-tailer.ts @@ -5,6 +5,42 @@ export function isRecord(v: unknown): v is Record { return typeof v === "object" && v !== null && !Array.isArray(v); } +// --------------------------------------------------------------------------- +// JSONL record types (Claude CLI streaming output) +// --------------------------------------------------------------------------- + +type TextBlock = { type: "text"; text: string }; +type ToolUseBlock = { type: "tool_use"; name: string }; +type ThinkingBlock = { type: "thinking" }; +type ToolResultBlock = { type: "tool_result"; is_error?: boolean }; + +type ContentBlock = TextBlock | ToolUseBlock | ThinkingBlock | ToolResultBlock; + +type AssistantRecord = { + type: "assistant"; + message: { content: ContentBlock[] }; +}; + +type UserRecord = { + type: "user"; + message: { content: ContentBlock[] }; +}; + +type ContentBlockDeltaRecord = { + type: "content_block_delta"; + delta: { type: "text_delta"; text: string }; +}; + +type ResultRecord = { + type: "result"; + subtype?: "success" | "error"; + is_error?: boolean; + result?: string; + error?: string; +}; + +export type JsonlRecord = AssistantRecord | UserRecord | ContentBlockDeltaRecord | ResultRecord; + function truncate(s: string, n: number): string { return s.length > n ? s.slice(0, n) + "..." : s; } @@ -18,65 +54,53 @@ function redactSensitive(input: string): string { .replace(/-----BEGIN [A-Z ]+ KEY-----/g, "[REDACTED]"); } +/** Accepts a parsed JSONL record (untrusted) and returns a display summary, or null to skip. */ export function summarizeJsonlRecord(record: Record): string | null { - if (record.type === "assistant") { - const message = isRecord(record.message) ? record.message : null; - if (message) { - const content = Array.isArray(message.content) ? (message.content as unknown[]) : []; + const typed = record as JsonlRecord; + + switch (typed.type) { + case "assistant": + case "user": { + const message = isRecord(typed.message) ? typed.message : null; + if (!message) return null; + const content = Array.isArray(message.content) ? (message.content as ContentBlock[]) : []; for (const block of content) { if (!isRecord(block)) continue; - if (block.type === "tool_use") { - return redactSensitive(`Tool: ${String(block.name ?? "unknown")}`); - } - if (block.type === "text") { - return redactSensitive(truncate(String(block.text ?? ""), 200)); - } - if (block.type === "thinking") { - return redactSensitive("Thinking..."); + switch (block.type) { + case "tool_use": + return redactSensitive(`Tool: ${String((block as ToolUseBlock).name ?? "unknown")}`); + case "text": + return redactSensitive(truncate(String((block as TextBlock).text ?? ""), 200)); + case "thinking": + return redactSensitive("Thinking..."); + case "tool_result": + return redactSensitive((block as ToolResultBlock).is_error === true ? "Tool error" : "Tool result"); } } + return null; } - return null; - } - - if (record.type === "user") { - const message = isRecord(record.message) ? record.message : null; - if (message) { - const content = Array.isArray(message.content) ? (message.content as unknown[]) : []; - for (const block of content) { - if (!isRecord(block)) continue; - if (block.type === "tool_result") { - if (block.is_error === true) { - return redactSensitive("Tool error"); - } - return redactSensitive("Tool result"); - } + case "content_block_delta": { + const delta = isRecord(typed.delta) ? typed.delta : null; + if (delta && (delta as ContentBlockDeltaRecord["delta"]).type === "text_delta") { + return redactSensitive(truncate(String((delta as ContentBlockDeltaRecord["delta"]).text ?? ""), 200)); } + return null; } - return null; - } - - if (record.type === "content_block_delta") { - const delta = isRecord(record.delta) ? record.delta : null; - if (delta && delta.type === "text_delta") { - return redactSensitive(truncate(String(delta.text ?? ""), 200)); - } - return null; - } - - if (record.type === "result") { - if (record.subtype === "success") { - return redactSensitive("Turn complete"); - } - if (record.subtype === "error" || record.is_error === true) { - return redactSensitive( - `Error: ${truncate(String(record.result ?? record.error ?? ""), 200)}` - ); + case "result": { + const r = typed as ResultRecord; + if (r.subtype === "success") { + return redactSensitive("Turn complete"); + } + if (r.subtype === "error" || r.is_error === true) { + return redactSensitive( + `Error: ${truncate(String(r.result ?? r.error ?? ""), 200)}` + ); + } + return null; } - return null; + default: + return null; } - - return null; } // --------------------------------------------------------------------------- From b59b75a810be1170fbee5c9c320a694126b4132a Mon Sep 17 00:00:00 2001 From: "daniel.ochoa" Date: Wed, 25 Mar 2026 11:54:07 -0500 Subject: [PATCH 5/5] PLAN-73: Add descriptive context to tool call and result summaries - Tool calls now include key input (file path, command, pattern) e.g. 'Tool: Read(/src/server/app.ts)' instead of 'Tool: Read' - Tool results now include a truncated preview of the output e.g. 'Tool result: found 3 matches...' instead of 'Tool result' Testing: All 18 tests pass, lint and typecheck clean Risks: None identified --- .../src/server/operations/output-tailer.ts | 35 ++++++++++++++++--- 1 file changed, 30 insertions(+), 5 deletions(-) diff --git a/apps/desktop/src/server/operations/output-tailer.ts b/apps/desktop/src/server/operations/output-tailer.ts index 070cf11f..48952f31 100644 --- a/apps/desktop/src/server/operations/output-tailer.ts +++ b/apps/desktop/src/server/operations/output-tailer.ts @@ -10,9 +10,9 @@ export function isRecord(v: unknown): v is Record { // --------------------------------------------------------------------------- type TextBlock = { type: "text"; text: string }; -type ToolUseBlock = { type: "tool_use"; name: string }; +type ToolUseBlock = { type: "tool_use"; name: string; input?: Record }; type ThinkingBlock = { type: "thinking" }; -type ToolResultBlock = { type: "tool_result"; is_error?: boolean }; +type ToolResultBlock = { type: "tool_result"; is_error?: boolean; content?: string | unknown[] }; type ContentBlock = TextBlock | ToolUseBlock | ThinkingBlock | ToolResultBlock; @@ -54,6 +54,28 @@ function redactSensitive(input: string): string { .replace(/-----BEGIN [A-Z ]+ KEY-----/g, "[REDACTED]"); } +function summarizeToolInput(name: string, input: Record): string { + const filePath = input.file_path ?? input.path; + if (typeof filePath === "string") return `Tool: ${name}(${truncate(filePath, 80)})`; + if (typeof input.command === "string") return `Tool: ${name}(${truncate(input.command, 80)})`; + if (typeof input.pattern === "string") return `Tool: ${name}(${truncate(input.pattern, 80)})`; + return `Tool: ${name}`; +} + +function summarizeToolResult(block: ToolResultBlock): string { + if (block.is_error === true) return "Tool error"; + const content = block.content; + if (typeof content === "string" && content.length > 0) return `Tool result: ${truncate(content, 120)}`; + if (Array.isArray(content)) { + for (const part of content) { + if (isRecord(part) && part.type === "text" && typeof part.text === "string") { + return `Tool result: ${truncate(part.text, 120)}`; + } + } + } + return "Tool result"; +} + /** Accepts a parsed JSONL record (untrusted) and returns a display summary, or null to skip. */ export function summarizeJsonlRecord(record: Record): string | null { const typed = record as JsonlRecord; @@ -67,14 +89,17 @@ export function summarizeJsonlRecord(record: Record): string | for (const block of content) { if (!isRecord(block)) continue; switch (block.type) { - case "tool_use": - return redactSensitive(`Tool: ${String((block as ToolUseBlock).name ?? "unknown")}`); + case "tool_use": { + const b = block as ToolUseBlock; + const input = isRecord(b.input) ? b.input : {}; + return redactSensitive(summarizeToolInput(String(b.name ?? "unknown"), input)); + } case "text": return redactSensitive(truncate(String((block as TextBlock).text ?? ""), 200)); case "thinking": return redactSensitive("Thinking..."); case "tool_result": - return redactSensitive((block as ToolResultBlock).is_error === true ? "Tool error" : "Tool result"); + return redactSensitive(summarizeToolResult(block as ToolResultBlock)); } } return null;