From ea4eb7f7e3e3b8cd0b71fe16e11df2f27de68763 Mon Sep 17 00:00:00 2001 From: Davide Rodo Date: Fri, 23 Oct 2026 00:16:15 +0200 Subject: [PATCH 1/4] perf(core): keep the server responsive during log bursts A process printing 100k lines left /api/health unanswered for ~9s. Each line was its own autocommit INSERT (re-prepared every time), and pumpStream read already-buffered pipe chunks as microtasks, so the whole burst ran in one event-loop turn. Log lines are now buffered and written once per turn in a single transaction with a cached statement; the stored rows (ids and session_id included) are what the broadcaster publishes, and close() flushes any pending lines before the DB closes. pumpStream yields with setImmediate after each chunk. The FTS trigger stays on the insert path: batching alone brings the worst /api/health latency during the burst to ~70ms. Co-Authored-By: Claude Opus 5.5 --- packages/core/src/db/queries.ts | 48 +++++++++++++------- packages/core/src/executor/wrapper.ts | 3 ++ packages/core/src/workspace/session.ts | 25 ++++++++-- packages/core/test/log-writes.test.ts | 32 +++++++++++++ packages/core/test/workspace-manager.test.ts | 2 + packages/core/test/workspace-session.test.ts | 41 ++++++++++++++++- packages/core/test/wrapper.test.ts | 31 +++++++++++++ 7 files changed, 159 insertions(+), 23 deletions(-) create mode 100644 packages/core/test/log-writes.test.ts diff --git a/packages/core/src/db/queries.ts b/packages/core/src/db/queries.ts index a96500d..35b37e7 100644 --- a/packages/core/src/db/queries.ts +++ b/packages/core/src/db/queries.ts @@ -74,23 +74,37 @@ export class ConductorQueries { } insertLog(row: Omit): 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>): 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: { diff --git a/packages/core/src/executor/wrapper.ts b/packages/core/src/executor/wrapper.ts index 876d771..da7f4b2 100644 --- a/packages/core/src/executor/wrapper.ts +++ b/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) { diff --git a/packages/core/src/workspace/session.ts b/packages/core/src/workspace/session.ts index e77f89f..adc924b 100644 --- a/packages/core/src/workspace/session.ts +++ b/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> = []; + 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(); } diff --git a/packages/core/test/log-writes.test.ts b/packages/core/test/log-writes.test.ts new file mode 100644 index 0000000..6ed34f5 --- /dev/null +++ b/packages/core/test/log-writes.test.ts @@ -0,0 +1,32 @@ +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([]); +}); diff --git a/packages/core/test/workspace-manager.test.ts b/packages/core/test/workspace-manager.test.ts index 571dd8d..64a9350 100644 --- a/packages/core/test/workspace-manager.test.ts +++ b/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(); diff --git a/packages/core/test/workspace-session.test.ts b/packages/core/test/workspace-session.test.ts index 179c9d6..d9a0324 100644 --- a/packages/core/test/workspace-session.test.ts +++ b/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 () => { diff --git a/packages/core/test/wrapper.test.ts b/packages/core/test/wrapper.test.ts index 78db4c3..3000fc4 100644 --- a/packages/core/test/wrapper.test.ts +++ b/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(); + 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", From 471178592cee2b4e9d9762480ceb9eeff52e9f2c Mon Sep 17 00:00:00 2001 From: Davide Rodo Date: Fri, 23 Oct 2026 00:21:36 +0200 Subject: [PATCH 2/4] perf(ui): stop the log viewer freezing during a live burst With the viewer open while a process printed 20k lines, the browser spent ~8.6s in main-thread long tasks: every SSE event set state, mergeLogs rebuilt a Set and sorted the whole array per line, and every row re-rendered (and re-parsed its ANSI codes) each time. SSE entries are now queued and applied once per animation frame. mergeLogs appends a batch directly when its ids follow what is held (the live tail) and only dedups and sorts when they overlap, as with the history fetch and the SSE replay. The viewer keeps the newest 5000 lines and says older output is in History. Rows are memoized, and auto-scroll follows the array rather than its length, which stops changing at the cap. Co-Authored-By: Claude Opus 5.5 --- packages/ui/src/components/LogLines.tsx | 45 ++++++++++++++---------- packages/ui/src/components/LogViewer.tsx | 35 ++++++++++++++---- packages/ui/src/lib/mergeLogs.ts | 39 +++++++++++++++----- packages/ui/test/mergeLogs.test.ts | 14 ++++++++ 4 files changed, 98 insertions(+), 35 deletions(-) diff --git a/packages/ui/src/components/LogLines.tsx b/packages/ui/src/components/LogLines.tsx index 2adad57..f98111d 100644 --- a/packages/ui/src/components/LogLines.tsx +++ b/packages/ui/src/components/LogLines.tsx @@ -1,3 +1,4 @@ +import { memo } from "react"; import type { LogRow } from "../lib/api"; import { renderAnsiLine } from "../lib/ansi"; @@ -6,26 +7,32 @@ export function LogLines({ logs }: { logs: LogRow[] }) { return (
{logs.map((log) => ( -
- - {new Date(log.timestamp).toLocaleTimeString()} - {" "} - - {renderAnsiLine(log.message, String(log.id))} -
+ ))}
); } + +// Memoized so a live tail only renders (and ANSI-parses) the lines it adds. +const LogLine = memo(function LogLine({ log }: { log: LogRow }) { + return ( +
+ + {new Date(log.timestamp).toLocaleTimeString()} + {" "} + + {renderAnsiLine(log.message, String(log.id))} +
+ ); +}); diff --git a/packages/ui/src/components/LogViewer.tsx b/packages/ui/src/components/LogViewer.tsx index b804016..ec4a4d3 100644 --- a/packages/ui/src/components/LogViewer.tsx +++ b/packages/ui/src/components/LogViewer.tsx @@ -25,7 +25,7 @@ import { fetchLogs, streamLogs, type LogRow, type ProcessInfo } from "../lib/api import { useUiStore } from "../store/ui"; import { useStopProcess, useRestartCommand } from "../hooks/useProcessActions"; import { LogLines } from "./LogLines"; -import { mergeLogs } from "../lib/mergeLogs"; +import { MAX_LOG_ROWS, mergeLogs } from "../lib/mergeLogs"; import { STATUS_COLOR } from "../lib/statusColor"; type StreamFilter = "all" | "stdout" | "stderr"; @@ -69,15 +69,28 @@ export function LogViewer({ process }: { process: ProcessInfo }) { // Live tail: SSE replays recent history too, overlapping the fetch above // (kept for an instant first paint) in either order, so both merge by id. + // Entries are applied once per frame: a burst sends thousands a second. + let queued: LogRow[] = []; + let frame = 0; const unsubscribe = streamLogs(pid, (entry) => { - setLogState((prev) => ({ - pid, - rows: mergeLogs(prev.pid === pid ? prev.rows : [], [entry]), - })); + queued.push(entry); + // rAF doesn't run in a hidden window; only the newest lines survive the cap anyway. + if (queued.length >= 2 * MAX_LOG_ROWS) queued = queued.slice(-MAX_LOG_ROWS); + if (frame) return; + frame = requestAnimationFrame(() => { + frame = 0; + const batch = queued; + queued = []; + setLogState((prev) => ({ + pid, + rows: mergeLogs(prev.pid === pid ? prev.rows : [], batch), + })); + }); }); return () => { cancelled = true; + cancelAnimationFrame(frame); unsubscribe(); }; }, [process.pid]); @@ -91,11 +104,12 @@ export function LogViewer({ process }: { process: ProcessInfo }) { }); }, [logs, search, streamFilter]); + // On the array, not its length: once the cap is reached new lines no longer change it. useEffect(() => { if (autoScroll) { viewportRef.current?.scrollTo({ top: viewportRef.current.scrollHeight }); } - }, [filteredLogs.length, autoScroll]); + }, [filteredLogs, autoScroll]); // Auto-scroll pauses if the user scrolls up to read history, and // resumes once they scroll back to the bottom. @@ -203,7 +217,14 @@ export function LogViewer({ process }: { process: ProcessInfo }) { {logs.length === 0 ? "No log output yet." : "No lines match the current filter."} ) : ( - + <> + {logs.length >= MAX_LOG_ROWS && ( + + Showing the last {MAX_LOG_ROWS} lines. Older output is in History. + + )} + + )} diff --git a/packages/ui/src/lib/mergeLogs.ts b/packages/ui/src/lib/mergeLogs.ts index 5802725..64b6ba9 100644 --- a/packages/ui/src/lib/mergeLogs.ts +++ b/packages/ui/src/lib/mergeLogs.ts @@ -1,14 +1,35 @@ import type { LogRow } from "./api"; +/** The live viewer keeps only the newest lines; older output stays in History. */ +export const MAX_LOG_ROWS = 5000; + /** - * Adds `incoming` to `rows`, skipping ids already present, ordered by id. - * LogViewer fills from two overlapping sources - the /api/logs history fetch - * and the SSE stream, which replays the same recent history before tailing - - * and either can land first, so both go through here. + * Adds `incoming` to `rows`, skipping ids already present, ordered by id, and + * keeps the newest `max`. LogViewer fills from two overlapping sources - the + * /api/logs history fetch and the SSE stream, which replays the same recent + * history before tailing - and either can land first, so both go through here. */ -export function mergeLogs(rows: LogRow[], incoming: LogRow[]): LogRow[] { - const seen = new Set(rows.map((r) => r.id)); - const fresh = incoming.filter((r) => !seen.has(r.id) && seen.add(r.id)); - if (fresh.length === 0) return rows; - return [...rows, ...fresh].sort((a, b) => a.id - b.id); +export function mergeLogs(rows: LogRow[], incoming: LogRow[], max = MAX_LOG_ROWS): LogRow[] { + if (incoming.length === 0) return rows; + let merged: LogRow[]; + if (isAppend(rows, incoming)) { + // The live tail: ids only grow, so no dedup or sort needed. + merged = rows.concat(incoming); + } else { + const seen = new Set(rows.map((r) => r.id)); + const fresh = incoming.filter((r) => !seen.has(r.id) && seen.add(r.id)); + if (fresh.length === 0) return rows; + merged = [...rows, ...fresh].sort((a, b) => a.id - b.id); + } + return merged.length > max ? merged.slice(-max) : merged; +} + +/** Whether `incoming` is strictly increasing and starts after the last of `rows`. */ +function isAppend(rows: LogRow[], incoming: LogRow[]): boolean { + let last = rows.at(-1)?.id ?? -Infinity; + for (const row of incoming) { + if (row.id <= last) return false; + last = row.id; + } + return true; } diff --git a/packages/ui/test/mergeLogs.test.ts b/packages/ui/test/mergeLogs.test.ts index b6d26b0..c989ac1 100644 --- a/packages/ui/test/mergeLogs.test.ts +++ b/packages/ui/test/mergeLogs.test.ts @@ -17,7 +17,21 @@ test("history fetch landing after SSE lines merges in id order", () => { expect(ids(mergeLogs(rows, [1, 2, 3].map(row)))).toEqual([1, 2, 3, 4]); }); +test("a batch overlapping what is held is deduplicated", () => { + const rows = [1, 2, 3].map(row); + expect(ids(mergeLogs(rows, [2, 3, 4, 5].map(row)))).toEqual([1, 2, 3, 4, 5]); +}); + test("returns the same array when nothing is new", () => { const rows = [row(1)]; expect(mergeLogs(rows, [row(1)])).toBe(rows); + expect(mergeLogs(rows, [])).toBe(rows); +}); + +test("keeps only the newest rows past the cap", () => { + let rows: LogRow[] = []; + for (let id = 1; id <= 10; id += 2) rows = mergeLogs(rows, [row(id), row(id + 1)], 4); + expect(ids(rows)).toEqual([7, 8, 9, 10]); + // A late history fetch can't push older lines back in. + expect(ids(mergeLogs(rows, [1, 2, 3].map(row), 4))).toEqual([7, 8, 9, 10]); }); From ec21d9788e1f538a2d06ddd701ba10e4e52351dc Mon Sep 17 00:00:00 2001 From: Davide Rodo Date: Fri, 23 Oct 2026 00:23:54 +0200 Subject: [PATCH 3/4] perf(core): index log history by process queryLogs filtered by process_id with no index on it, so ORDER BY timestamp DESC walked idx_logs_timestamp across every process until it found 500 matches. On a 1M-row table a quiet or freshly started process took ~190ms per query, and the viewer runs it twice on open (the /api/logs fetch and the SSE replay). Add idx_logs_process on (process_id, id) and order history by id, which follows insertion order and matches the index, so neither the pid nor the command filter needs a sort. The same queries now take ~0.3ms. Building the index on an existing 1M-row database takes ~0.3s, once, at startup. Co-Authored-By: Claude Opus 5.5 --- packages/core/src/db/queries.ts | 4 +++- packages/core/src/db/schema.sql | 4 ++++ packages/core/test/log-writes.test.ts | 20 ++++++++++++++++++++ 3 files changed, 27 insertions(+), 1 deletion(-) diff --git a/packages/core/src/db/queries.ts b/packages/core/src/db/queries.ts index 35b37e7..c129d57 100644 --- a/packages/core/src/db/queries.ts +++ b/packages/core/src/db/queries.ts @@ -149,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[]; } diff --git a/packages/core/src/db/schema.sql b/packages/core/src/db/schema.sql index ce81cb8..dd35827 100644 --- a/packages/core/src/db/schema.sql +++ b/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); diff --git a/packages/core/test/log-writes.test.ts b/packages/core/test/log-writes.test.ts index 6ed34f5..6544f92 100644 --- a/packages/core/test/log-writes.test.ts +++ b/packages/core/test/log-writes.test.ts @@ -30,3 +30,23 @@ test("insertLogs with an empty batch is a no-op", () => { 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=?)", + ]); +}); From b82e875cf9df4744703003edd42f472ff7f11950 Mon Sep 17 00:00:00 2001 From: Davide Rodo Date: Fri, 23 Oct 2026 00:26:10 +0200 Subject: [PATCH 4/4] fix(core): run Python commands unbuffered by default Python block-buffers stdout when it is a pipe, so a service printing a line every 200ms showed nothing in the viewer until ~8KB had piled up (or until it exited). PYTHONUNBUFFERED=1 is now the lowest env layer, below process.env, so the shell, global_env, profile env and env_overrides can all override it (an empty value turns it off). Co-Authored-By: Claude Opus 5.5 --- docs/CONFIG.md | 2 ++ packages/core/src/config/env-resolution.ts | 3 +++ packages/core/test/env-resolution.test.ts | 20 ++++++++++++++++++++ 3 files changed, 25 insertions(+) diff --git a/docs/CONFIG.md b/docs/CONFIG.md index d0331a4..121616a 100644 --- a/docs/CONFIG.md +++ b/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`). diff --git a/packages/core/src/config/env-resolution.ts b/packages/core/src/config/env-resolution.ts index 829526f..4932b11 100644 --- a/packages/core/src/config/env-resolution.ts +++ b/packages/core/src/config/env-resolution.ts @@ -29,6 +29,9 @@ export interface BuildCommandEnvParams extends BuildEnvParams { function baseLayers(params: BuildEnvParams): Array | 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, { BASE_PATH: resolveBasePath(configFilePath, config.base_path) }, config.default_shell ? { CONDUCTOR_SHELL: config.default_shell } : undefined, diff --git a/packages/core/test/env-resolution.test.ts b/packages/core/test/env-resolution.test.ts index 9b19ea9..8e9b70d 100644 --- a/packages/core/test/env-resolution.test.ts +++ b/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; + } + }); });