fix: stabilize logs follow output (#1318) (thanks @sebslight)

CHANGELOG.md

@@ -39,6 +39,7 @@ Docs: https://docs.clawd.bot
 - TUI: keep thinking blocks ordered before content during streaming and isolate per-run assembly. (#1202) — thanks @aaronveklabs.
 - TUI: align custom editor initialization with the latest pi-tui API. (#1298) — thanks @sibbl.
 - CLI: avoid duplicating --profile/--dev flags when formatting commands.
+- CLI: make `clawdbot logs --follow` output stable, keep spinners, and surface closed pipes. (#1318) — thanks @sebslight.
 - Status: route native `/status` to the active agent so model selection reflects the correct profile. (#1301)
 - Exec: prefer bash when fish is default shell, falling back to sh if bash is missing. (#1297) — thanks @ysqander.
 - Exec: merge login-shell PATH for host=gateway exec while keeping daemon PATH minimal. (#1304)

src/cli/logs-cli.test.ts

@@ -0,0 +1,85 @@
+import { Command } from "commander";
+import { afterEach, describe, expect, it, vi } from "vitest";
+
+const callGatewayFromCli = vi.fn();
+
+vi.mock("./gateway-rpc.js", async () => {
+  const actual = await vi.importActual<typeof import("./gateway-rpc.js")>("./gateway-rpc.js");
+  return {
+    ...actual,
+    callGatewayFromCli: (...args: unknown[]) => callGatewayFromCli(...args),
+  };
+});
+
+describe("logs cli", () => {
+  afterEach(() => {
+    callGatewayFromCli.mockReset();
+  });
+
+  it("writes output directly to stdout/stderr", async () => {
+    callGatewayFromCli.mockResolvedValueOnce({
+      file: "/tmp/clawdbot.log",
+      cursor: 1,
+      size: 123,
+      lines: ["raw line"],
+      truncated: true,
+      reset: true,
+    });
+
+    const stdoutWrites: string[] = [];
+    const stderrWrites: string[] = [];
+    const stdoutSpy = vi.spyOn(process.stdout, "write").mockImplementation((chunk: unknown) => {
+      stdoutWrites.push(String(chunk));
+      return true;
+    });
+    const stderrSpy = vi.spyOn(process.stderr, "write").mockImplementation((chunk: unknown) => {
+      stderrWrites.push(String(chunk));
+      return true;
+    });
+
+    const { registerLogsCli } = await import("./logs-cli.js");
+    const program = new Command();
+    program.exitOverride();
+    registerLogsCli(program);
+
+    await program.parseAsync(["logs"], { from: "user" });
+
+    stdoutSpy.mockRestore();
+    stderrSpy.mockRestore();
+
+    expect(stdoutWrites.join("")).toContain("Log file:");
+    expect(stdoutWrites.join("")).toContain("raw line");
+    expect(stderrWrites.join("")).toContain("Log tail truncated");
+    expect(stderrWrites.join("")).toContain("Log cursor reset");
+  });
+
+  it("warns when the output pipe closes", async () => {
+    callGatewayFromCli.mockResolvedValueOnce({
+      file: "/tmp/clawdbot.log",
+      lines: ["line one"],
+    });
+
+    const stderrWrites: string[] = [];
+    const stdoutSpy = vi.spyOn(process.stdout, "write").mockImplementation(() => {
+      const err = new Error("EPIPE") as NodeJS.ErrnoException;
+      err.code = "EPIPE";
+      throw err;
+    });
+    const stderrSpy = vi.spyOn(process.stderr, "write").mockImplementation((chunk: unknown) => {
+      stderrWrites.push(String(chunk));
+      return true;
+    });
+
+    const { registerLogsCli } = await import("./logs-cli.js");
+    const program = new Command();
+    program.exitOverride();
+    registerLogsCli(program);
+
+    await program.parseAsync(["logs"], { from: "user" });
+
+    stdoutSpy.mockRestore();
+    stderrSpy.mockRestore();
+
+    expect(stderrWrites.join("")).toContain("output stdout closed");
+  });
+});

src/cli/logs-cli.ts

@@ -2,8 +2,8 @@ import { setTimeout as delay } from "node:timers/promises";
 import type { Command } from "commander";
 import { buildGatewayConnectionDetails } from "../gateway/call.js";
 import { parseLogLine } from "../logging/parse-log-line.js";
-import { defaultRuntime } from "../runtime.js";
 import { formatDocsLink } from "../terminal/links.js";
+import { clearActiveProgressLine } from "../terminal/progress-line.js";
 import { colorize, isRich, theme } from "../terminal/theme.js";
 import { formatCliCommand } from "./command-format.js";
 import { addGatewayClientOptions, callGatewayFromCli } from "./gateway-rpc.js";
@@ -104,10 +104,68 @@ function formatLogLine(
   return [head, messageValue].filter(Boolean).join(" ").trim();
 }
 
-function emitJsonLine(payload: Record<string, unknown>, toStdErr = false) {
+let outputClosed = false;
+let outputClosedNotice = false;
+
+function isBrokenPipeError(err: unknown): err is NodeJS.ErrnoException {
+  const code = (err as NodeJS.ErrnoException)?.code;
+  return code === "EPIPE" || code === "EIO";
+}
+
+function noteOutputClosed(err: NodeJS.ErrnoException, stream: NodeJS.WriteStream) {
+  if (outputClosedNotice) return;
+  outputClosedNotice = true;
+  const code = err.code ?? "EPIPE";
+  const target = stream === process.stdout ? "stdout" : "stderr";
+  const message = `clawdbot logs: output ${target} closed (${code}). Stopping tail.`;
+  try {
+    clearActiveProgressLine();
+    process.stderr.write(`${message}\n`);
+  } catch {
+    // ignore secondary failures while reporting the broken pipe
+  }
+}
+
+function handleWriteError(err: unknown, stream: NodeJS.WriteStream): boolean {
+  if (!isBrokenPipeError(err)) {
+    throw err;
+  }
+  outputClosed = true;
+  noteOutputClosed(err, stream);
+  return false;
+}
+
+function writeText(text: string, stream: NodeJS.WriteStream): boolean {
+  if (outputClosed) return false;
+  try {
+    clearActiveProgressLine();
+  } catch (err) {
+    if (!handleWriteError(err, process.stderr)) return false;
+  }
+  try {
+    stream.write(text);
+    return true;
+  } catch (err) {
+    return handleWriteError(err, stream);
+  }
+}
+
+function writeLine(text: string, stream: NodeJS.WriteStream): boolean {
+  // Avoid feeding CLI output back into the log file via console capture.
+  return writeText(`${text}\n`, stream);
+}
+
+function logLine(text: string): boolean {
+  return writeLine(text, process.stdout);
+}
+
+function errorLine(text: string): boolean {
+  return writeLine(text, process.stderr);
+}
+
+function emitJsonLine(payload: Record<string, unknown>, toStdErr = false): boolean {
   const text = `${JSON.stringify(payload)}\n`;
-  if (toStdErr) process.stderr.write(text);
+  return writeText(text, toStdErr ? process.stderr : process.stdout);
-  else process.stdout.write(text);
 }
 
 function emitGatewayError(
@@ -122,22 +180,26 @@ function emitGatewayError(
   const errorText = err instanceof Error ? err.message : String(err);
 
   if (mode === "json") {
-    emitJsonLine(
+    if (
-      {
+      !emitJsonLine(
-        type: "error",
+        {
-        message,
+          type: "error",
-        error: errorText,
+          message,
-        details,
+          error: errorText,
-        hint,
+          details,
-      },
+          hint,
-      true,
+        },
-    );
+        true,
+      )
+    ) {
+      return;
+    }
     return;
   }
 
-  defaultRuntime.error(colorize(rich, theme.error, message));
+  if (!errorLine(colorize(rich, theme.error, message))) return;
-  defaultRuntime.error(details.message);
+  if (!errorLine(details.message)) return;
-  defaultRuntime.error(colorize(rich, theme.muted, hint));
+  errorLine(colorize(rich, theme.muted, hint));
 }
 
 export function registerLogsCli(program: Command) {
@@ -159,6 +221,8 @@ export function registerLogsCli(program: Command) {
   addGatewayClientOptions(logs);
 
   logs.action(async (opts: LogsCliOptions) => {
+    outputClosed = false;
+    outputClosedNotice = false;
     const interval = parsePositiveInt(opts.interval, 1000);
     let cursor: number | undefined;
     let first = true;
@@ -172,57 +236,83 @@ export function registerLogsCli(program: Command) {
         payload = await fetchLogs(opts, cursor);
       } catch (err) {
         emitGatewayError(err, opts, jsonMode ? "json" : "text", rich);
-        defaultRuntime.exit(1);
+        process.exit(1);
         return;
       }
       const lines = Array.isArray(payload.lines) ? payload.lines : [];
       if (jsonMode) {
         if (first) {
-          emitJsonLine({
+          if (
-            type: "meta",
+            !emitJsonLine({
-            file: payload.file,
+              type: "meta",
-            cursor: payload.cursor,
+              file: payload.file,
-            size: payload.size,
+              cursor: payload.cursor,
-          });
+              size: payload.size,
+            })
+          ) {
+            return;
+          }
         }
         for (const line of lines) {
           const parsed = parseLogLine(line);
           if (parsed) {
-            emitJsonLine({ type: "log", ...parsed });
+            if (!emitJsonLine({ type: "log", ...parsed })) {
+              return;
+            }
           } else {
-            emitJsonLine({ type: "raw", raw: line });
+            if (!emitJsonLine({ type: "raw", raw: line })) {
+              return;
+            }
           }
         }
         if (payload.truncated) {
-          emitJsonLine({
+          if (
-            type: "notice",
+            !emitJsonLine({
-            message: "Log tail truncated (increase --max-bytes).",
+              type: "notice",
-          });
+              message: "Log tail truncated (increase --max-bytes).",
+            })
+          ) {
+            return;
+          }
         }
         if (payload.reset) {
-          emitJsonLine({
+          if (
-            type: "notice",
+            !emitJsonLine({
-            message: "Log cursor reset (file rotated).",
+              type: "notice",
-          });
+              message: "Log cursor reset (file rotated).",
+            })
+          ) {
+            return;
+          }
         }
       } else {
         if (first && payload.file) {
           const prefix = pretty ? colorize(rich, theme.muted, "Log file:") : "Log file:";
-          defaultRuntime.log(`${prefix} ${payload.file}`);
+          if (!logLine(`${prefix} ${payload.file}`)) {
+            return;
+          }
         }
         for (const line of lines) {
-          defaultRuntime.log(
+          if (
-            formatLogLine(line, {
+            !logLine(
-              pretty,
+              formatLogLine(line, {
-              rich,
+                pretty,
-            }),
+                rich,
-          );
+              }),
+            )
+          ) {
+            return;
+          }
         }
         if (payload.truncated) {
-          defaultRuntime.error("Log tail truncated (increase --max-bytes).");
+          if (!errorLine("Log tail truncated (increase --max-bytes).")) {
+            return;
+          }
         }
         if (payload.reset) {
-          defaultRuntime.error("Log cursor reset (file rotated).");
+          if (!errorLine("Log cursor reset (file rotated).")) {
+            return;
+          }
         }
       }
       cursor =