perf: make log streaming fast under output bursts

docs/CONFIG.md

@@ -100,6 +100,8 @@ Profiles **do not embed commands**. They contain three possible keys:
 
 Example: if global_env sets `NODE_ENV=production`, profile env sets nothing, and command env_overrides sets `NODE_ENV=development`, the final value is `development`.
 
+Below all of these, Conductor sets `PYTHONUNBUFFERED=1`. Python block-buffers stdout when it is a pipe, so without it a Python service's output reaches the log viewer in ~8KB lumps instead of line by line. Any layer above can override it; set it to an empty string to turn it off. Other runtimes that buffer piped output need their own setting.
+
 ### Secret Masking
 
 Any environment variable whose **name** appears in the top-level `env_secrets` array will have its value replaced with `[FILTERED]` everywhere — logs, UI display, and API responses. Both exact match and substring match against the var name are supported (e.g., `API_TOKEN`, `DB_PASSWORD`).

packages/core/src/config/env-resolution.ts

@@ -29,6 +29,9 @@ export interface BuildCommandEnvParams extends BuildEnvParams {
 function baseLayers(params: BuildEnvParams): Array<Record<string, string> | undefined> {
   const { configFilePath, config, profile, dbGlobalEnv, dbProfileEnv } = params;
   return [
+    // Python block-buffers stdout into a pipe, so its output would reach
+    // the log viewer in ~8KB lumps. Lowest layer: anything can override it.
+    { PYTHONUNBUFFERED: "1" },
     process.env as Record<string, string>,
     { BASE_PATH: resolveBasePath(configFilePath, config.base_path) },
     config.default_shell ? { CONDUCTOR_SHELL: config.default_shell } : undefined,

packages/core/src/db/queries.ts

@@ -74,23 +74,37 @@ export class ConductorQueries {
   }
 
   insertLog(row: Omit<LogRow, "id" | "session_id">): LogRow {
-    const sessionId = this.currentSessionByProfile.get(row.profile) ?? null;
-    return this.db
-      .prepare(
-        `INSERT INTO logs (process_id, command_id, profile, timestamp, level, stream, message, session_id)
-         VALUES ($process_id, $command_id, $profile, $timestamp, $level, $stream, $message, $session_id)
-         RETURNING *`,
-      )
-      .get({
-        $process_id: row.process_id,
-        $command_id: row.command_id,
-        $profile: row.profile,
-        $timestamp: row.timestamp,
-        $level: row.level,
-        $stream: row.stream,
-        $message: row.message,
-        $session_id: sessionId,
-      }) as LogRow;
+    return this.insertLogs([row])[0] as LogRow;
+  }
+
+  /**
+   * Inserts a batch in one transaction, returning the stored rows (ids
+   * included) in order. A burst of output written one autocommit per line
+   * kept the event loop busy for seconds.
+   */
+  insertLogs(rows: Array<Omit<LogRow, "id" | "session_id">>): LogRow[] {
+    if (rows.length === 0) return [];
+    // db.query caches the compiled statement across calls.
+    const insert = this.db.query(
+      `INSERT INTO logs (process_id, command_id, profile, timestamp, level, stream, message, session_id)
+       VALUES ($process_id, $command_id, $profile, $timestamp, $level, $stream, $message, $session_id)
+       RETURNING *`,
+    );
+    return this.db.transaction(() =>
+      rows.map(
+        (row) =>
+          insert.get({
+            $process_id: row.process_id,
+            $command_id: row.command_id,
+            $profile: row.profile,
+            $timestamp: row.timestamp,
+            $level: row.level,
+            $stream: row.stream,
+            $message: row.message,
+            $session_id: this.currentSessionByProfile.get(row.profile) ?? null,
+          }) as LogRow,
+      ),
+    )();
   }
 
   queryLogs(filters: {
@@ -135,8 +149,10 @@ export class ConductorQueries {
     const where = clauses.length > 0 ? `WHERE ${clauses.join(" AND ")}` : "";
     const limit = filters.limit ?? 200;
 
+    // By id, not timestamp: ids follow insertion order, and the order matches
+    // idx_logs_process / idx_logs_command so no sort over a whole run is needed.
     return this.db
-      .prepare(`SELECT * FROM logs ${where} ORDER BY timestamp DESC, id DESC LIMIT $limit`)
+      .prepare(`SELECT * FROM logs ${where} ORDER BY id DESC LIMIT $limit`)
       .all({ ...params, $limit: limit }) as LogRow[];
   }
 

packages/core/src/db/schema.sql

@@ -26,7 +26,11 @@ CREATE TABLE IF NOT EXISTS logs (
   session_id INTEGER
 );
 
+-- Both end in the rowid (id), so "one process/command, newest first" reads
+-- just its own rows; without idx_logs_process the viewer's history query
+-- walked idx_logs_timestamp across every process.
 CREATE INDEX IF NOT EXISTS idx_logs_command ON logs(command_id);
+CREATE INDEX IF NOT EXISTS idx_logs_process ON logs(process_id, id);
 CREATE INDEX IF NOT EXISTS idx_logs_timestamp ON logs(timestamp);
 CREATE INDEX IF NOT EXISTS idx_logs_session ON logs(session_id);
 

packages/core/src/executor/wrapper.ts

@@ -496,6 +496,9 @@ export class ProcessWrapper {
       for (const line of lines) {
         this.emitLogFor(owner, line, kind);
       }
+      // Chunks already sitting in the pipe resolve as microtasks, so a burst
+      // would otherwise run start to finish without letting HTTP in.
+      await new Promise((resolve) => setImmediate(resolve));
     }
 
     if (buffer.length > 0) {

packages/core/src/workspace/session.ts

@@ -6,7 +6,7 @@ import type { ConductorConfig } from "../config/schema";
 import { ConfigStore, dbEnvLookup } from "../config/store";
 import { saveConfig } from "../config/writer";
 import { DEFAULT_DB_PATH, openDatabase } from "../db/init";
-import { ConductorQueries } from "../db/queries";
+import { ConductorQueries, type LogRow } from "../db/queries";
 import type { LogEntry, LogHandler } from "../executor/wrapper";
 import { createLogger, type ConductorLogger } from "../logger/pino";
 import type { LogBroadcaster } from "../logs/broadcaster";
@@ -118,14 +118,28 @@ export function openSession(dir: string, opts: SessionOptions): WorkspaceSession
 
   // Every log line from any managed process is persisted and broadcast
   // so both the CLI (via `conductor logs`) and the UI's live SSE stream
-  // can see it, regardless of who started the process.
+  // can see it, regardless of who started the process. Lines are buffered
+  // and written once per event-loop turn, in one transaction: the stored
+  // rows (with their ids) are what gets broadcast.
+  let pendingLogs: Array<Omit<LogRow, "id" | "session_id">> = [];
+  const flushLogs = () => {
+    const batch = pendingLogs;
+    pendingLogs = [];
+    if (batch.length === 0) return;
+    try {
+      for (const row of queries.insertLogs(batch)) broadcaster.publish(row);
+    } catch (err) {
+      logger.error({ err, lines: batch.length }, "Failed to store log lines");
+    }
+  };
   // Set once the DB is closed. A process's last lines can still be in its pipe
   // after stop() returns (waiting for them stalled every close on Windows),
-  // and they must not reach insertLog on a closed DB.
+  // and they must not reach insertLogs on a closed DB.
   let dbClosed = false;
   const onLog: LogHandler = (entry: LogEntry) => {
     if (dbClosed) return;
-    const row = queries.insertLog({
+    if (pendingLogs.length === 0) setImmediate(flushLogs);
+    pendingLogs.push({
       process_id: String(entry.pid),
       command_id: entry.commandId,
       profile: entry.profile,
@@ -134,7 +148,6 @@ export function openSession(dir: string, opts: SessionOptions): WorkspaceSession
       stream: entry.stream,
       message: entry.message,
     });
-    broadcaster.publish(row);
   };
 
   // Time-window log retention sweep. Session-scoped retention doesn't need
@@ -167,6 +180,8 @@ export function openSession(dir: string, opts: SessionOptions): WorkspaceSession
       } finally {
         collector.stop();
         clearInterval(logRetentionInterval);
+        // Lines already received are written; any later ones are dropped.
+        flushLogs();
         dbClosed = true;
         db.close();
       }

packages/core/test/env-resolution.test.ts

@@ -66,4 +66,24 @@ describe("buildCommandEnv", () => {
     expect(pathKey).toBeDefined();
     expect(env[pathKey!]).toBe(process.env[pathKey!]);
   });
+
+  // Python block-buffers stdout into a pipe, so a service's output showed
+  // up in the viewer ~8KB at a time instead of line by line.
+  test("defaults PYTHONUNBUFFERED=1 below every other layer", () => {
+    const params = { configFilePath, config, profile: config.profiles.dev };
+    const saved = process.env.PYTHONUNBUFFERED;
+    delete process.env.PYTHONUNBUFFERED;
+    try {
+      expect(buildCommandEnv({ ...params, cmd: config.commands[0]! }).PYTHONUNBUFFERED).toBe("1");
+
+      const overridden = { ...config.commands[0]!, env_overrides: { PYTHONUNBUFFERED: "" } };
+      expect(buildCommandEnv({ ...params, cmd: overridden }).PYTHONUNBUFFERED).toBe("");
+
+      process.env.PYTHONUNBUFFERED = "0";
+      expect(buildCommandEnv({ ...params, cmd: config.commands[0]! }).PYTHONUNBUFFERED).toBe("0");
+    } finally {
+      if (saved === undefined) delete process.env.PYTHONUNBUFFERED;
+      else process.env.PYTHONUNBUFFERED = saved;
+    }
+  });
 });

packages/core/test/log-writes.test.ts

@@ -0,0 +1,52 @@
+import { test, expect } from "bun:test";
+import { openDatabase } from "../src/db/init";
+import { ConductorQueries } from "../src/db/queries";
+
+const line = (message: string) => ({
+  process_id: "10",
+  command_id: "api",
+  profile: "dev",
+  timestamp: new Date().toISOString(),
+  level: "info",
+  stream: "stdout" as const,
+  message,
+});
+
+test("insertLogs writes a batch with increasing ids, session stamps and FTS entries", () => {
+  const queries = new ConductorQueries(openDatabase(":memory:"));
+  const sessionId = queries.insertSession("dev");
+
+  const rows = queries.insertLogs([line("alpha one"), line("beta two"), line("gamma three")]);
+
+  expect(rows.map((r) => r.message)).toEqual(["alpha one", "beta two", "gamma three"]);
+  expect(rows[1]!.id).toBe(rows[0]!.id + 1);
+  expect(rows[2]!.id).toBe(rows[1]!.id + 1);
+  expect(rows.every((r) => r.session_id === sessionId)).toBe(true);
+  expect(queries.queryLogs({ grep: "beta" }).map((r) => r.id)).toEqual([rows[1]!.id]);
+});
+
+test("insertLogs with an empty batch is a no-op", () => {
+  const queries = new ConductorQueries(openDatabase(":memory:"));
+  expect(queries.insertLogs([])).toEqual([]);
+  expect(queries.queryLogs({})).toEqual([]);
+});
+
+test("history for one process comes from its index, newest id first", () => {
+  const db = openDatabase(":memory:");
+  const queries = new ConductorQueries(db);
+  // Timestamps out of id order: the viewer's order is insertion order.
+  const [first, second] = queries.insertLogs([
+    { ...line("first"), timestamp: "2026-01-02T00:00:00.000Z" },
+    { ...line("second"), timestamp: "2026-01-01T00:00:00.000Z" },
+  ]);
+  expect(queries.queryLogs({ processId: "10" }).map((r) => r.id)).toEqual([second!.id, first!.id]);
+
+  const plan = db
+    .query(
+      "EXPLAIN QUERY PLAN SELECT * FROM logs WHERE process_id = $pid ORDER BY id DESC LIMIT 500",
+    )
+    .all({ $pid: "10" }) as Array<{ detail: string }>;
+  expect(plan.map((p) => p.detail)).toEqual([
+    "SEARCH logs USING INDEX idx_logs_process (process_id=?)",
+  ]);
+});

packages/core/test/workspace-manager.test.ts

@@ -286,6 +286,8 @@ describe("WorkspaceManager + API", () => {
       message: "hello-from-b",
       timestamp: new Date().toISOString(),
     });
+    // Log lines are written in a batch on the next event-loop turn.
+    await new Promise((resolve) => setImmediate(resolve));
     expect(manager.current!.queries.queryLogs({ grep: "hello-from-b" })).toHaveLength(1);
 
     await manager.close();

packages/core/test/workspace-session.test.ts

@@ -6,6 +6,7 @@ import {
   ConfigError,
   DEFAULT_DB_PATH,
   LogBroadcaster,
+  type LogRow,
   openSession,
   prepareSession,
   resolveWorkspaceDir,
@@ -110,9 +111,45 @@ describe("openSession", () => {
     },
   );
 
+  test("onLog writes a burst in one batch and publishes the stored rows", async () => {
+    const broadcaster = new LogBroadcaster();
+    const session = openSession(dir, { broadcaster, dbPath: ":memory:" });
+    const published: LogRow[] = [];
+    broadcaster.subscribe((row) => published.push(row));
+    const entry = (message: string) => ({
+      commandId: "api",
+      commandName: "API",
+      profile: "dev",
+      pid: 42,
+      stream: "stdout" as const,
+      message,
+      timestamp: new Date().toISOString(),
+    });
+
+    for (const message of ["one", "two", "three"]) session.onLog(entry(message));
+    expect(published).toEqual([]);
+
+    await new Promise((resolve) => setImmediate(resolve));
+    expect(published.map((r) => r.message)).toEqual(["one", "two", "three"]);
+    expect(published.map((r) => r.id)).toEqual(
+      session.queries
+        .queryLogs({ processId: "42" })
+        .map((r) => r.id)
+        .reverse(),
+    );
+
+    // A line still waiting for its flush is written before close() shuts the DB.
+    session.onLog(entry("last words"));
+    await session.close();
+    expect(published.at(-1)?.message).toBe("last words");
+  });
+
   test("a log line arriving after close() is dropped, not written to the closed DB", async () => {
     // A process's last lines can still be in its pipe after stop() returns.
-    const session = openSession(dir, { broadcaster: new LogBroadcaster(), dbPath: ":memory:" });
+    const broadcaster = new LogBroadcaster();
+    const published: LogRow[] = [];
+    broadcaster.subscribe((row) => published.push(row));
+    const session = openSession(dir, { broadcaster, dbPath: ":memory:" });
     await session.close();
     expect(() =>
       session.onLog({
@@ -125,6 +162,8 @@ describe("openSession", () => {
         timestamp: new Date().toISOString(),
       }),
     ).not.toThrow();
+    await new Promise((resolve) => setImmediate(resolve));
+    expect(published).toEqual([]);
   });
 
   test("close() still releases the DB when stopping processes fails", async () => {

packages/core/test/wrapper.test.ts

@@ -119,6 +119,37 @@ describe("ProcessWrapper.start", () => {
     expect(lines).toEqual(["a", "b", "c"]);
   });
 
+  test("yields to the event loop between chunks of an output burst", async () => {
+    // A slow handler (like the per-line SQLite write) lets the pipe fill up;
+    // already-buffered chunks then resolve as microtasks, so without a yield
+    // the whole burst runs in one event-loop turn and HTTP handling starves.
+    const cmd = makeCommand({
+      id: "burst",
+      name: "Burst",
+      run: `bun -e "for (let i = 0; i < 50000; i++) console.log('line ' + i + ' of a burst')"`,
+    });
+    const wrapper = new ProcessWrapper(cmd, "test", testEnv());
+    let turns = 0;
+    let running = true;
+    const tick = () => {
+      turns++;
+      if (running) setImmediate(tick);
+    };
+    setImmediate(tick);
+    const turnsSeen = new Set<number>();
+    wrapper.onLog(() => {
+      turnsSeen.add(turns);
+      const start = performance.now();
+      while (performance.now() - start < 0.03) {
+        // busy-wait ~30µs, about what a per-line SQLite write cost
+      }
+    });
+    await runToExit(wrapper);
+    await wrapper.drained;
+    running = false;
+    expect(turnsSeen.size).toBeGreaterThan(5);
+  }, 20_000);
+
   test("flags hasMatchedLogLine once a log_line healthcheck pattern appears in output", async () => {
     const cmd = makeCommand({
       id: "log-line-match",