Merge pull request #40 from closedloop-ai/symphony/plan-73

PLAN-73: Add intermediate event reporting to Electron loop harness

Author: shafty023

apps/desktop/package.json

@@ -1,6 +1,6 @@
 {
   "name": "desktop",
-  "version": "0.8.1",
+  "version": "0.8.2",
   "description": "ClosedLoop Desktop",
   "author": "ClosedLoop AI <support@closedloop.ai>",
   "private": true,

apps/desktop/src/server/operations/output-tailer.ts

@@ -0,0 +1,235 @@
+import { openSync, readSync, closeSync, existsSync } from "node:fs";
+import { randomUUID } from "node:crypto";
+
+export function isRecord(v: unknown): v is Record<string, unknown> {
+  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; input?: Record<string, unknown> };
+type ThinkingBlock = { type: "thinking" };
+type ToolResultBlock = { type: "tool_result"; is_error?: boolean; content?: string | unknown[] };
+
+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;
+}
+
+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]");
+}
+
+function summarizeToolInput(name: string, input: Record<string, unknown>): 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, unknown>): string | null {
+  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;
+        switch (block.type) {
+          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(summarizeToolResult(block as ToolResultBlock));
+        }
+      }
+      return null;
+    }
+    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;
+    }
+    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;
+    }
+    default:
+      return null;
+  }
+}
+
+// ---------------------------------------------------------------------------
+// API communication
+// ---------------------------------------------------------------------------
+
+async function postLoopEvent(
+  apiBaseUrl: string,
+  loopId: string,
+  token: string,
+  event: { type: string; data: { chunk: string } }
+): Promise<void> {
+  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: { chunk: event.data.chunk },
+        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<void> } {
+  let stopped = false;
+  let byteOffset = initialByteOffset;
+  let pendingRemainder = Buffer.alloc(0);
+  let lastSentAt: number | null = null;
+
+  async function pollOnce(): Promise<void> {
+    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: { chunk: lastDisplay } });
+      }
+    }
+  }
+
+  const intervalId = setInterval(() => { pollOnce().catch(() => {}); }, 2000);
+
+  return {
+    stop: () => { stopped = true; clearInterval(intervalId); },
+    flush: async () => {
+      clearInterval(intervalId);
+      await pollOnce();
+      stopped = true;
+    },
+  };
+}

apps/desktop/src/server/operations/symphony-loop.ts

@@ -1,7 +1,7 @@
 import { execSync, spawn } from "node:child_process";
 import { gatewayLog } from "../../main/gateway-logger.js";
 import crypto from "node:crypto";
-import { closeSync, existsSync, mkdirSync, openSync, readFileSync, unlinkSync, writeFileSync } from "node:fs";
+import { closeSync, existsSync, mkdirSync, openSync, readFileSync, statSync, unlinkSync, writeFileSync } from "node:fs";
 import fs from "node:fs/promises";
 import os from "node:os";
 import path from "node:path";
@@ -23,6 +23,7 @@ import {
   resolveWorktreeParentDir,
   tryAssertRepoAllowed,
 } from "./symphony-utils.js";
+import { startOutputTailer } from "./output-tailer.js";
 
 // ---------------------------------------------------------------------------
 // Types
@@ -216,11 +217,13 @@ 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"). */
@@ -1765,14 +1768,24 @@ 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<void> } = {
+      stop: () => {},
+      flush: () => Promise.resolve(),
+    };
+    const onceComplete = async (code: number): Promise<void> => {
+      if (completionHandled) return;
       completionHandled = true;
       loopLog(body.loopId, `onceComplete fired, code=${code}`);
+      try {
+        await stopTailer.flush();
+      } catch (err) {
+        loopError(body.loopId, "Tailer flush error:", err);
+      }
       handleProcessCompletion(
         code,
         body,
@@ -1793,15 +1806,15 @@ async function handleLoopRequest(
     // 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
     // inherited file descriptors (not pipes), 'close' may never fire
     // 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;
@@ -1815,6 +1828,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}`);
     gatewayLog.debug("loop-harness", `Spawned ${body.command} pid=${pid}, loopId=${body.loopId}, worktree=${worktreeDir}`);