From 41bd5eb806e98ed8bcf547d7b1e7b04599b85846 Mon Sep 17 00:00:00 2001 From: David Susskind Date: Tue, 11 Aug 2026 10:16:28 +0300 Subject: [PATCH 01/10] feat(logs): make --follow consume the realtime SSE stream with poll fallback --follow now connects to the new apper endpoint GET /api/apps/{app_id}/functions-mgmt/logs/stream (SSE, same Bearer auth as the bounded logs route) and prints log events as they arrive. On connect failure or after one reconnect attempt it falls back to today's 2s poll loop with a one-line stderr notice. Co-Authored-By: Claude Fable 5 --- packages/cli/src/cli/commands/project/logs.ts | 118 +++++++++++++++++- .../cli/src/core/resources/function/index.ts | 1 + .../src/core/resources/function/stream-api.ts | 118 ++++++++++++++++++ packages/cli/tests/cli/logs.spec.ts | 43 +++++++ 4 files changed, 278 insertions(+), 2 deletions(-) create mode 100644 packages/cli/src/core/resources/function/stream-api.ts diff --git a/packages/cli/src/cli/commands/project/logs.ts b/packages/cli/src/cli/commands/project/logs.ts index 503551a69..f6f2e3b07 100644 --- a/packages/cli/src/cli/commands/project/logs.ts +++ b/packages/cli/src/cli/commands/project/logs.ts @@ -1,3 +1,4 @@ +import type { Logger } from "@base44-cli/logger"; import type { Command } from "commander"; import { Option } from "commander"; import type { CLIContext, RunCommandResult } from "@/cli/types.js"; @@ -9,12 +10,15 @@ import type { FunctionLogsResponse, LogEnv, LogLevel, + LogStreamFilters, + StreamLogEvent, } from "@/core/resources/function/index.js"; import { fetchFunctionLogs, LogEnvSchema, LogLevelSchema, listDeployedFunctions, + openLogStream, } from "@/core/resources/function/index.js"; interface LogsOptions { @@ -123,14 +127,123 @@ function writeFollowLine(entry: LogEntry, jsonMode: boolean): void { process.stdout.write(`${line}\n`); } +function streamEventToLogEntry(event: StreamLogEvent): LogEntry { + return { + time: event.time, + level: event.level, + message: event.function + ? `[${event.function}] ${event.message}` + : event.message, + source: event.function ?? "", + }; +} + +async function printStreamUntilDrop( + stream: AsyncGenerator, + levelFilter: string | undefined, + jsonMode: boolean, + startTime: string, +): Promise { + let lastTime = startTime; + try { + for await (const event of stream) { + if (levelFilter && event.level !== levelFilter) continue; + writeFollowLine(streamEventToLogEntry(event), jsonMode); + if (event.time > lastTime) lastTime = event.time; + } + } catch {} + return lastTime; +} + +const STREAM_CONNECT_ATTEMPTS = 2; + +interface StreamAttemptResult { + everConnected: boolean; + lastTime: string; +} + +async function followViaStream( + options: LogsOptions, + jsonMode: boolean, + startTime: string, +): Promise { + const filters: LogStreamFilters = { + functions: parseFunctionNames(options.function), + env: options.env, + }; + let everConnected = false; + let lastTime = startTime; + for (let attempt = 0; attempt < STREAM_CONNECT_ATTEMPTS; attempt++) { + const stream = await openLogStream(filters); + if (!stream) break; + everConnected = true; + lastTime = await printStreamUntilDrop( + stream, + options.level, + jsonMode, + lastTime, + ); + } + return { everConnected, lastTime }; +} + +async function printBackfill( + functionNames: string[], + options: LogsOptions, + availableFunctionNames: string[], + jsonMode: boolean, +): Promise { + const entries = await fetchLogsForFunctions( + functionNames, + options, + availableFunctionNames, + ); + entries.sort((a, b) => a.time.localeCompare(b.time)); + for (const entry of entries) writeFollowLine(entry, jsonMode); + return entries.at(-1)?.time ?? ""; +} + async function followLogs( functionNames: string[], options: LogsOptions, availableFunctionNames: string[], jsonMode: boolean, + logger: Logger, +): Promise { + let backfilledUntil = ""; + if (options.since) { + backfilledUntil = await printBackfill( + functionNames, + options, + availableFunctionNames, + jsonMode, + ); + } + const { everConnected, lastTime } = await followViaStream( + options, + jsonMode, + backfilledUntil, + ); + logger.warn( + everConnected + ? "Realtime stream disconnected — falling back to polling (lines may lag ~20-30s)." + : "Realtime stream unavailable — falling back to polling (lines may lag ~20-30s).", + ); + return pollLogs(functionNames, options, availableFunctionNames, jsonMode, { + lastTime, + boundaryKeys: new Set(), + }); +} + +async function pollLogs( + functionNames: string[], + options: LogsOptions, + availableFunctionNames: string[], + jsonMode: boolean, + initialState: FollowState, ): Promise { - let state: FollowState = { lastTime: "", boundaryKeys: new Set() }; - let first = true; + let state = initialState; + let first = state.lastTime === ""; while (true) { const pollOptions = first ? options : { ...options, since: state.lastTime }; @@ -295,6 +408,7 @@ async function logsAction( options, availableFunctionNames, ctx.jsonMode, + ctx.log, ); } diff --git a/packages/cli/src/core/resources/function/index.ts b/packages/cli/src/core/resources/function/index.ts index 08b63274b..514e1513a 100644 --- a/packages/cli/src/core/resources/function/index.ts +++ b/packages/cli/src/core/resources/function/index.ts @@ -4,3 +4,4 @@ export * from "./deploy.js"; export * from "./pull.js"; export * from "./resource.js"; export * from "./schema.js"; +export * from "./stream-api.js"; diff --git a/packages/cli/src/core/resources/function/stream-api.ts b/packages/cli/src/core/resources/function/stream-api.ts new file mode 100644 index 000000000..286a65ee8 --- /dev/null +++ b/packages/cli/src/core/resources/function/stream-api.ts @@ -0,0 +1,118 @@ +import { z } from "zod"; +import { + getWorkspaceApiKeyFromEnv, + isTokenExpired, + isWorkspaceApiKey, + readAuth, + refreshAndSaveTokens, +} from "@/core/auth/config.js"; +import { getBase44ApiUrl } from "@/core/config.js"; +import { getAppContext } from "@/core/project/index.js"; +import { + type LogEnv, + LogLevelSchema, +} from "@/core/resources/function/schema.js"; + +export const StreamLogEventSchema = z.object({ + time: z.string(), + level: z.preprocess( + (value) => (value === "warn" ? "warning" : value), + LogLevelSchema, + ), + function: z.string().nullable(), + message: z.string(), +}); + +export type StreamLogEvent = z.infer; + +export interface LogStreamFilters { + functions?: string[]; + env?: LogEnv; +} + +function buildStreamUrl(filters: LogStreamFilters): string { + const { id } = getAppContext(); + const url = new URL( + `/api/apps/${id}/functions-mgmt/logs/stream`, + getBase44ApiUrl(), + ); + if (filters.functions?.length) { + url.searchParams.set("function", filters.functions.join(",")); + } + if (filters.env) { + url.searchParams.set("env", filters.env); + } + return url.href; +} + +async function buildStreamAuthHeaders(): Promise> { + const workspaceApiKey = getWorkspaceApiKeyFromEnv(); + if (workspaceApiKey && isWorkspaceApiKey(workspaceApiKey)) { + return { api_key: workspaceApiKey }; + } + const auth = await readAuth(); + if (isTokenExpired(auth)) { + const refreshedToken = await refreshAndSaveTokens(); + if (refreshedToken) { + return { Authorization: `Bearer ${refreshedToken}` }; + } + } + return { Authorization: `Bearer ${auth.accessToken}` }; +} + +export function parseStreamEventLine(line: string): StreamLogEvent | null { + if (!line.startsWith("data:")) return null; + try { + const result = StreamLogEventSchema.safeParse(JSON.parse(line.slice(5))); + return result.success ? result.data : null; + } catch { + return null; + } +} + +async function* readLines( + body: ReadableStream, +): AsyncGenerator { + const reader = body.getReader(); + const decoder = new TextDecoder(); + let buffered = ""; + try { + while (true) { + const { done, value } = await reader.read(); + if (done) return; + buffered += decoder.decode(value, { stream: true }); + const lines = buffered.split("\n"); + buffered = lines.pop() ?? ""; + yield* lines; + } + } finally { + reader.releaseLock(); + } +} + +async function* readStreamEvents( + body: ReadableStream, +): AsyncGenerator { + for await (const line of readLines(body)) { + const event = parseStreamEventLine(line); + if (event) yield event; + } +} + +export async function openLogStream( + filters: LogStreamFilters, +): Promise | null> { + let response: Response; + try { + response = await fetch(buildStreamUrl(filters), { + headers: { + Accept: "text/event-stream", + ...(await buildStreamAuthHeaders()), + }, + }); + } catch { + return null; + } + if (!response.ok || !response.body) return null; + return readStreamEvents(response.body); +} diff --git a/packages/cli/tests/cli/logs.spec.ts b/packages/cli/tests/cli/logs.spec.ts index 0da92dd34..996cf79ff 100644 --- a/packages/cli/tests/cli/logs.spec.ts +++ b/packages/cli/tests/cli/logs.spec.ts @@ -4,6 +4,7 @@ import { type LogEntry, selectNewEntries, } from "@/cli/commands/project/logs.js"; +import { parseStreamEventLine } from "@/core/resources/function/index.js"; import { fixture, setupCLITests } from "./testkit/index.js"; function entry(time: string, message: string): LogEntry { @@ -72,6 +73,48 @@ describe("selectNewEntries (follow dedup)", () => { }); }); +describe("parseStreamEventLine (SSE log stream)", () => { + it("parses a data line into a stream log event", () => { + const event = parseStreamEventLine( + 'data: {"time":"2024-01-15T10:00:00Z","level":"info","function":"my-fn","message":"hello"}', + ); + + expect(event).toEqual({ + time: "2024-01-15T10:00:00Z", + level: "info", + function: "my-fn", + message: "hello", + }); + }); + + it("normalizes level warn to warning", () => { + const event = parseStreamEventLine( + 'data: {"time":"2024-01-15T10:00:00Z","level":"warn","function":"my-fn","message":"careful"}', + ); + + expect(event?.level).toBe("warning"); + }); + + it("keeps unattributed lines (null function)", () => { + const event = parseStreamEventLine( + 'data: {"time":"2024-01-15T10:00:00Z","level":"error","function":null,"message":"boom"}', + ); + + expect(event).not.toBeNull(); + expect(event?.function).toBeNull(); + }); + + it("ignores keepalive comments and blank lines", () => { + expect(parseStreamEventLine(": ping")).toBeNull(); + expect(parseStreamEventLine("")).toBeNull(); + }); + + it("ignores malformed data lines", () => { + expect(parseStreamEventLine("data: not-json")).toBeNull(); + expect(parseStreamEventLine('data: {"level":"info"}')).toBeNull(); + }); +}); + describe("logs command", () => { const t = setupCLITests(); From 52c53bd10d4bfab78643574c013fa8eaa7d8dc83 Mon Sep 17 00:00:00 2001 From: David Susskind Date: Tue, 11 Aug 2026 13:48:52 +0300 Subject: [PATCH 02/10] feat(logs): reject --since combined with --follow (v1 downscope) David's ruling from the draft review: the backfill-then-attach seam had a silent ~17-20s data hole (backfill reads the lagging bounded index while the stream tails from connect), which is data loss in a debugging tool. Guard the combination like --until/--order instead; also rename followViaStream to streamUntilExhausted so the fallback sequence below it reads as the failure path it is. Co-Authored-By: Claude Fable 5 --- packages/cli/src/cli/commands/project/logs.ts | 38 ++++--------------- packages/cli/tests/cli/logs.spec.ts | 16 ++++++++ 2 files changed, 24 insertions(+), 30 deletions(-) diff --git a/packages/cli/src/cli/commands/project/logs.ts b/packages/cli/src/cli/commands/project/logs.ts index f6f2e3b07..a2e7823d3 100644 --- a/packages/cli/src/cli/commands/project/logs.ts +++ b/packages/cli/src/cli/commands/project/logs.ts @@ -162,17 +162,16 @@ interface StreamAttemptResult { lastTime: string; } -async function followViaStream( +async function streamUntilExhausted( options: LogsOptions, jsonMode: boolean, - startTime: string, ): Promise { const filters: LogStreamFilters = { functions: parseFunctionNames(options.function), env: options.env, }; let everConnected = false; - let lastTime = startTime; + let lastTime = ""; for (let attempt = 0; attempt < STREAM_CONNECT_ATTEMPTS; attempt++) { const stream = await openLogStream(filters); if (!stream) break; @@ -187,22 +186,6 @@ async function followViaStream( return { everConnected, lastTime }; } -async function printBackfill( - functionNames: string[], - options: LogsOptions, - availableFunctionNames: string[], - jsonMode: boolean, -): Promise { - const entries = await fetchLogsForFunctions( - functionNames, - options, - availableFunctionNames, - ); - entries.sort((a, b) => a.time.localeCompare(b.time)); - for (const entry of entries) writeFollowLine(entry, jsonMode); - return entries.at(-1)?.time ?? ""; -} - async function followLogs( functionNames: string[], options: LogsOptions, @@ -210,19 +193,9 @@ async function followLogs( jsonMode: boolean, logger: Logger, ): Promise { - let backfilledUntil = ""; - if (options.since) { - backfilledUntil = await printBackfill( - functionNames, - options, - availableFunctionNames, - jsonMode, - ); - } - const { everConnected, lastTime } = await followViaStream( + const { everConnected, lastTime } = await streamUntilExhausted( options, jsonMode, - backfilledUntil, ); logger.warn( everConnected @@ -392,6 +365,11 @@ async function logsAction( } if (options.follow) { + if (options.since) { + throw new InvalidInputError( + "--since cannot be combined with --follow yet (the realtime stream starts from now).", + ); + } if (options.until) { throw new InvalidInputError( "--until cannot be combined with --follow (a stream has no end).", diff --git a/packages/cli/tests/cli/logs.spec.ts b/packages/cli/tests/cli/logs.spec.ts index 996cf79ff..e8d3a6a04 100644 --- a/packages/cli/tests/cli/logs.spec.ts +++ b/packages/cli/tests/cli/logs.spec.ts @@ -308,6 +308,22 @@ describe("logs command", () => { t.expectResult(result).toContain("No production logs found"); }); + it("rejects --follow combined with --since", async () => { + await t.givenLoggedInWithProject(fixture("basic")); + + const result = await t.run( + "logs", + "--function", + "my-function", + "--follow", + "--since", + "1h", + ); + + t.expectResult(result).toFail(); + t.expectResult(result).toContain("--since cannot be combined"); + }); + it("rejects --follow combined with --order", async () => { await t.givenLoggedInWithProject(fixture("basic")); From 5bcde87a3000598a6e1abecee8c1b388c6719caa Mon Sep 17 00:00:00 2001 From: David Susskind Date: Tue, 11 Aug 2026 14:27:14 +0300 Subject: [PATCH 03/10] feat(logs): reason-driven stream lifecycle via typed SSE end event MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit David's ruling: the server must not translate a diagnosed condition into a bare EOF. The bridge now self-heals degraded tails invisibly and, when it gives up, sends 'event: end' with {reason, retriable} before closing. The cli replaces its magic retry-counter with a reason-driven policy: retriable:false → poll fallback; retriable:true → reconnect after 1s; bare EOF → reconnect, giving up after 2 consecutive drops that produced no events (counter resets on any event, so long-lived sessions never exhaust a budget). Co-Authored-By: Claude Fable 5 --- packages/cli/src/cli/commands/project/logs.ts | 47 ++++++++++---- .../src/core/resources/function/stream-api.ts | 47 +++++++++++--- packages/cli/tests/cli/logs.spec.ts | 63 ++++++++++++------- 3 files changed, 116 insertions(+), 41 deletions(-) diff --git a/packages/cli/src/cli/commands/project/logs.ts b/packages/cli/src/cli/commands/project/logs.ts index a2e7823d3..4a3586204 100644 --- a/packages/cli/src/cli/commands/project/logs.ts +++ b/packages/cli/src/cli/commands/project/logs.ts @@ -11,6 +11,8 @@ import type { LogEnv, LogLevel, LogStreamFilters, + StreamEndEvent, + StreamEvent, StreamLogEvent, } from "@/core/resources/function/index.js"; import { @@ -127,6 +129,8 @@ function writeFollowLine(entry: LogEntry, jsonMode: boolean): void { process.stdout.write(`${line}\n`); } +const delay = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + function streamEventToLogEntry(event: StreamLogEvent): LogEntry { return { time: event.time, @@ -138,24 +142,35 @@ function streamEventToLogEntry(event: StreamLogEvent): LogEntry { }; } -async function printStreamUntilDrop( - stream: AsyncGenerator, +interface StreamEnding { + lastTime: string; + producedEvents: boolean; + end: StreamEndEvent | null; +} + +async function printStreamUntilEnd( + stream: AsyncGenerator, levelFilter: string | undefined, jsonMode: boolean, startTime: string, -): Promise { +): Promise { let lastTime = startTime; + let producedEvents = false; try { for await (const event of stream) { - if (levelFilter && event.level !== levelFilter) continue; - writeFollowLine(streamEventToLogEntry(event), jsonMode); - if (event.time > lastTime) lastTime = event.time; + producedEvents = true; + if (event.kind === "end") + return { lastTime, producedEvents, end: event.end }; + if (levelFilter && event.log.level !== levelFilter) continue; + writeFollowLine(streamEventToLogEntry(event.log), jsonMode); + if (event.log.time > lastTime) lastTime = event.log.time; } } catch {} - return lastTime; + return { lastTime, producedEvents, end: null }; } -const STREAM_CONNECT_ATTEMPTS = 2; +const STREAM_RECONNECT_DELAY_MS = 1_000; +const MAX_DROPS_SINCE_LAST_EVENT = 2; interface StreamAttemptResult { everConnected: boolean; @@ -172,16 +187,26 @@ async function streamUntilExhausted( }; let everConnected = false; let lastTime = ""; - for (let attempt = 0; attempt < STREAM_CONNECT_ATTEMPTS; attempt++) { + let dropsSinceLastEvent = 0; + while (true) { const stream = await openLogStream(filters); if (!stream) break; everConnected = true; - lastTime = await printStreamUntilDrop( + const ending = await printStreamUntilEnd( stream, options.level, jsonMode, lastTime, ); + lastTime = ending.lastTime; + if (ending.end) { + if (!ending.end.retriable) break; + dropsSinceLastEvent = 0; + } else { + dropsSinceLastEvent = ending.producedEvents ? 1 : dropsSinceLastEvent + 1; + if (dropsSinceLastEvent >= MAX_DROPS_SINCE_LAST_EVENT) break; + } + await delay(STREAM_RECONNECT_DELAY_MS); } return { everConnected, lastTime }; } @@ -230,7 +255,7 @@ async function pollLogs( fresh.sort((a, b) => a.time.localeCompare(b.time)); for (const entry of fresh) writeFollowLine(entry, jsonMode); first = false; - await new Promise((resolve) => setTimeout(resolve, 2000)); + await delay(2000); } } diff --git a/packages/cli/src/core/resources/function/stream-api.ts b/packages/cli/src/core/resources/function/stream-api.ts index 286a65ee8..6fd7f0f5f 100644 --- a/packages/cli/src/core/resources/function/stream-api.ts +++ b/packages/cli/src/core/resources/function/stream-api.ts @@ -25,6 +25,17 @@ export const StreamLogEventSchema = z.object({ export type StreamLogEvent = z.infer; +const StreamEndEventSchema = z.object({ + reason: z.string(), + retriable: z.boolean(), +}); + +export type StreamEndEvent = z.infer; + +export type StreamEvent = + | { kind: "log"; log: StreamLogEvent } + | { kind: "end"; end: StreamEndEvent }; + export interface LogStreamFilters { functions?: string[]; env?: LogEnv; @@ -60,11 +71,21 @@ async function buildStreamAuthHeaders(): Promise> { return { Authorization: `Bearer ${auth.accessToken}` }; } -export function parseStreamEventLine(line: string): StreamLogEvent | null { - if (!line.startsWith("data:")) return null; +export function parseStreamEvent( + eventName: string, + data: string, +): StreamEvent | null { try { - const result = StreamLogEventSchema.safeParse(JSON.parse(line.slice(5))); - return result.success ? result.data : null; + const payload = JSON.parse(data); + if (eventName === "end") { + const result = StreamEndEventSchema.safeParse(payload); + return result.success ? { kind: "end", end: result.data } : null; + } + if (eventName === "") { + const result = StreamLogEventSchema.safeParse(payload); + return result.success ? { kind: "log", log: result.data } : null; + } + return null; } catch { return null; } @@ -92,16 +113,26 @@ async function* readLines( async function* readStreamEvents( body: ReadableStream, -): AsyncGenerator { +): AsyncGenerator { + let eventName = ""; for await (const line of readLines(body)) { - const event = parseStreamEventLine(line); - if (event) yield event; + if (line.startsWith("event:")) { + eventName = line.slice(6).trim(); + continue; + } + if (line.startsWith("data:")) { + const event = parseStreamEvent(eventName, line.slice(5)); + eventName = ""; + if (event) yield event; + continue; + } + if (line.trim() === "") eventName = ""; } } export async function openLogStream( filters: LogStreamFilters, -): Promise | null> { +): Promise | null> { let response: Response; try { response = await fetch(buildStreamUrl(filters), { diff --git a/packages/cli/tests/cli/logs.spec.ts b/packages/cli/tests/cli/logs.spec.ts index e8d3a6a04..1aa5c3db4 100644 --- a/packages/cli/tests/cli/logs.spec.ts +++ b/packages/cli/tests/cli/logs.spec.ts @@ -4,7 +4,7 @@ import { type LogEntry, selectNewEntries, } from "@/cli/commands/project/logs.js"; -import { parseStreamEventLine } from "@/core/resources/function/index.js"; +import { parseStreamEvent } from "@/core/resources/function/index.js"; import { fixture, setupCLITests } from "./testkit/index.js"; function entry(time: string, message: string): LogEntry { @@ -73,45 +73,64 @@ describe("selectNewEntries (follow dedup)", () => { }); }); -describe("parseStreamEventLine (SSE log stream)", () => { - it("parses a data line into a stream log event", () => { - const event = parseStreamEventLine( - 'data: {"time":"2024-01-15T10:00:00Z","level":"info","function":"my-fn","message":"hello"}', +describe("parseStreamEvent (SSE log stream)", () => { + it("parses an unnamed data payload into a log event", () => { + const event = parseStreamEvent( + "", + '{"time":"2024-01-15T10:00:00Z","level":"info","function":"my-fn","message":"hello"}', ); expect(event).toEqual({ - time: "2024-01-15T10:00:00Z", - level: "info", - function: "my-fn", - message: "hello", + kind: "log", + log: { + time: "2024-01-15T10:00:00Z", + level: "info", + function: "my-fn", + message: "hello", + }, }); }); it("normalizes level warn to warning", () => { - const event = parseStreamEventLine( - 'data: {"time":"2024-01-15T10:00:00Z","level":"warn","function":"my-fn","message":"careful"}', + const event = parseStreamEvent( + "", + '{"time":"2024-01-15T10:00:00Z","level":"warn","function":"my-fn","message":"careful"}', ); - expect(event?.level).toBe("warning"); + expect(event?.kind === "log" && event.log.level).toBe("warning"); }); it("keeps unattributed lines (null function)", () => { - const event = parseStreamEventLine( - 'data: {"time":"2024-01-15T10:00:00Z","level":"error","function":null,"message":"boom"}', + const event = parseStreamEvent( + "", + '{"time":"2024-01-15T10:00:00Z","level":"error","function":null,"message":"boom"}', + ); + + expect(event?.kind === "log" && event.log.function).toBeNull(); + }); + + it("parses the typed end event with reason and retriable", () => { + const event = parseStreamEvent( + "end", + '{"reason":"tail_unavailable","retriable":false}', ); - expect(event).not.toBeNull(); - expect(event?.function).toBeNull(); + expect(event).toEqual({ + kind: "end", + end: { reason: "tail_unavailable", retriable: false }, + }); }); - it("ignores keepalive comments and blank lines", () => { - expect(parseStreamEventLine(": ping")).toBeNull(); - expect(parseStreamEventLine("")).toBeNull(); + it("ignores unknown event names", () => { + expect( + parseStreamEvent("progress", '{"reason":"x","retriable":true}'), + ).toBeNull(); }); - it("ignores malformed data lines", () => { - expect(parseStreamEventLine("data: not-json")).toBeNull(); - expect(parseStreamEventLine('data: {"level":"info"}')).toBeNull(); + it("ignores malformed payloads", () => { + expect(parseStreamEvent("", "not-json")).toBeNull(); + expect(parseStreamEvent("", '{"level":"info"}')).toBeNull(); + expect(parseStreamEvent("end", '{"reason":"x"}')).toBeNull(); }); }); From 2d90b703bf08e81a328623e8f11d7e1d2970b3a2 Mon Sep 17 00:00:00 2001 From: David Susskind Date: Tue, 11 Aug 2026 14:46:19 +0300 Subject: [PATCH 04/10] fix(logs): bound the stream's silent failure modes with two timeouts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Probe found a backend outage could mute --follow indefinitely. Two gaps, both in the stream leg: a half-open connection blocks reader.read() forever (the cli never enforced keepalive arrival), and the reconnect fetch had no connect timeout against a wedged backend. Now: 60s line-silence watchdog (any line incl. ': ping' resets it — transport liveness, so quiet-but-healthy apps never trigger it) treats silence as a bare drop, and a 10s connect-phase timeout guards the fetch (body deliberately unguarded — body liveness is the watchdog's job). Worst-case mute is now bounded, ending in the loud poll error. Co-Authored-By: Claude Fable 5 --- .../src/core/resources/function/stream-api.ts | 44 +++++++++++++++++-- 1 file changed, 40 insertions(+), 4 deletions(-) diff --git a/packages/cli/src/core/resources/function/stream-api.ts b/packages/cli/src/core/resources/function/stream-api.ts index 6fd7f0f5f..1ae784dcc 100644 --- a/packages/cli/src/core/resources/function/stream-api.ts +++ b/packages/cli/src/core/resources/function/stream-api.ts @@ -91,17 +91,43 @@ export function parseStreamEvent( } } +const STREAM_SILENCE_TIMEOUT_MS = 60_000; + +interface StreamReader { + read(): Promise<{ done: boolean; value?: Uint8Array }>; + cancel(): Promise; + releaseLock(): void; +} + +async function readOrSilence( + reader: StreamReader, +): Promise<{ done: boolean; value?: Uint8Array } | "silence"> { + let timer: ReturnType | undefined; + const silence = new Promise<"silence">((resolve) => { + timer = setTimeout(() => resolve("silence"), STREAM_SILENCE_TIMEOUT_MS); + }); + try { + return await Promise.race([reader.read(), silence]); + } finally { + clearTimeout(timer); + } +} + async function* readLines( body: ReadableStream, ): AsyncGenerator { - const reader = body.getReader(); + const reader: StreamReader = body.getReader(); const decoder = new TextDecoder(); let buffered = ""; try { while (true) { - const { done, value } = await reader.read(); - if (done) return; - buffered += decoder.decode(value, { stream: true }); + const result = await readOrSilence(reader); + if (result === "silence") { + await reader.cancel(); + return; + } + if (result.done || !result.value) return; + buffered += decoder.decode(result.value, { stream: true }); const lines = buffered.split("\n"); buffered = lines.pop() ?? ""; yield* lines; @@ -130,9 +156,16 @@ async function* readStreamEvents( } } +const STREAM_CONNECT_TIMEOUT_MS = 10_000; + export async function openLogStream( filters: LogStreamFilters, ): Promise | null> { + const connectPhase = new AbortController(); + const connectTimer = setTimeout( + () => connectPhase.abort(), + STREAM_CONNECT_TIMEOUT_MS, + ); let response: Response; try { response = await fetch(buildStreamUrl(filters), { @@ -140,9 +173,12 @@ export async function openLogStream( Accept: "text/event-stream", ...(await buildStreamAuthHeaders()), }, + signal: connectPhase.signal, }); } catch { return null; + } finally { + clearTimeout(connectTimer); } if (!response.ok || !response.body) return null; return readStreamEvents(response.body); From 01972c033d1691d98806d40b3c2f3d0f9e3c1067 Mon Sep 17 00:00:00 2001 From: David Susskind Date: Tue, 11 Aug 2026 15:53:58 +0300 Subject: [PATCH 05/10] ci: SHA-pin all actions to satisfy the org's SHA-lock policy Actions were re-enabled on the org with a policy requiring third-party actions to be SHA-locked; tag-pinned workflows now die at startup (startup_failure, 0s). Pin every uses: reference to a full commit SHA with the tag kept as a trailing comment, matching the form the already-passing workflows use for actions/checkout. Co-Authored-By: Claude Fable 5 --- .github/workflows/check-wix-proxy.yml | 4 ++-- .github/workflows/claude-code-review.yml | 4 ++-- .github/workflows/claude.yml | 4 ++-- .github/workflows/daily-error-report.yml | 6 +++--- .github/workflows/knip.yml | 6 +++--- .github/workflows/lint.yml | 6 +++--- .github/workflows/manual-publish.yml | 14 +++++++------- .github/workflows/pr-description.yml | 4 ++-- .github/workflows/preview-publish.yml | 10 +++++----- .github/workflows/readme-check.yml | 10 +++++----- .github/workflows/test.yml | 18 +++++++++--------- .github/workflows/typecheck.yml | 6 +++--- .github/workflows/wix-gateway-proxy-check.yml | 2 +- 13 files changed, 47 insertions(+), 47 deletions(-) diff --git a/.github/workflows/check-wix-proxy.yml b/.github/workflows/check-wix-proxy.yml index f595206dc..ef9e0cf47 100644 --- a/.github/workflows/check-wix-proxy.yml +++ b/.github/workflows/check-wix-proxy.yml @@ -17,13 +17,13 @@ jobs: runs-on: ubuntu-latest steps: - - uses: actions/checkout@v4 + - uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4 - name: Wix gateway proxy (mandatory) uses: ./.github/actions/wix-gateway-proxy - name: Set up Python - uses: actions/setup-python@v6 + uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6 with: python-version: "3.12" diff --git a/.github/workflows/claude-code-review.yml b/.github/workflows/claude-code-review.yml index 4052ab810..313751f52 100644 --- a/.github/workflows/claude-code-review.yml +++ b/.github/workflows/claude-code-review.yml @@ -27,7 +27,7 @@ jobs: steps: - name: Checkout repository - uses: actions/checkout@v4 + uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4 with: fetch-depth: 1 @@ -36,7 +36,7 @@ jobs: - name: Run Claude Code Review id: claude-review - uses: anthropics/claude-code-action@v1 + uses: anthropics/claude-code-action@5ef2e550a465a721f4f45e4a7d3c340c873e1dcc # v1 with: show_full_output: true anthropic_api_key: ${{ secrets.ANTHROPIC_API_KEY }} diff --git a/.github/workflows/claude.yml b/.github/workflows/claude.yml index 6f905d660..7c0e96d1e 100644 --- a/.github/workflows/claude.yml +++ b/.github/workflows/claude.yml @@ -26,7 +26,7 @@ jobs: actions: read # Required for Claude to read CI results on PRs steps: - name: Checkout repository - uses: actions/checkout@v4 + uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4 with: fetch-depth: 1 @@ -35,7 +35,7 @@ jobs: - name: Run Claude Code id: claude - uses: anthropics/claude-code-action@v1 + uses: anthropics/claude-code-action@5ef2e550a465a721f4f45e4a7d3c340c873e1dcc # v1 with: anthropic_api_key: ${{ secrets.ANTHROPIC_API_KEY }} diff --git a/.github/workflows/daily-error-report.yml b/.github/workflows/daily-error-report.yml index fc45a1d81..ceaa91f09 100644 --- a/.github/workflows/daily-error-report.yml +++ b/.github/workflows/daily-error-report.yml @@ -21,7 +21,7 @@ jobs: steps: - name: Checkout repository - uses: actions/checkout@v4 + uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4 with: fetch-depth: 1 @@ -29,7 +29,7 @@ jobs: uses: ./.github/actions/wix-gateway-proxy - name: Setup Node.js - uses: actions/setup-node@v4 + uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020 # v4 with: node-version: "20" @@ -77,7 +77,7 @@ jobs: - name: Generate report with Claude if: steps.check-errors.outputs.has_errors == 'true' - uses: anthropics/claude-code-action@v1 + uses: anthropics/claude-code-action@5ef2e550a465a721f4f45e4a7d3c340c873e1dcc # v1 with: anthropic_api_key: ${{ secrets.ANTHROPIC_API_KEY }} prompt: | diff --git a/.github/workflows/knip.yml b/.github/workflows/knip.yml index 536782c98..c44e2616b 100644 --- a/.github/workflows/knip.yml +++ b/.github/workflows/knip.yml @@ -12,19 +12,19 @@ jobs: steps: - name: Checkout code - uses: actions/checkout@v4 + uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4 - name: Wix gateway proxy (mandatory) uses: ./.github/actions/wix-gateway-proxy - name: Setup Bun id: setup-bun - uses: oven-sh/setup-bun@v2 + uses: oven-sh/setup-bun@0c5077e51419868618aeaa5fe8019c62421857d6 # v2 with: bun-version: latest - name: Cache Bun dependencies - uses: actions/cache@v5 + uses: actions/cache@caa296126883cff596d87d8935842f9db880ef25 # v5 with: path: ~/.bun/install/cache key: ${{ runner.os }}-bun-${{ steps.setup-bun.outputs.bun-version }}-${{ hashFiles('**/bun.lock') }} diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml index 5609871e3..ac337c181 100644 --- a/.github/workflows/lint.yml +++ b/.github/workflows/lint.yml @@ -12,19 +12,19 @@ jobs: steps: - name: Checkout code - uses: actions/checkout@v4 + uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4 - name: Wix gateway proxy (mandatory) uses: ./.github/actions/wix-gateway-proxy - name: Setup Bun id: setup-bun - uses: oven-sh/setup-bun@v2 + uses: oven-sh/setup-bun@0c5077e51419868618aeaa5fe8019c62421857d6 # v2 with: bun-version: latest - name: Cache Bun dependencies - uses: actions/cache@v5 + uses: actions/cache@caa296126883cff596d87d8935842f9db880ef25 # v5 with: path: ~/.bun/install/cache key: ${{ runner.os }}-bun-${{ steps.setup-bun.outputs.bun-version }}-${{ hashFiles('**/bun.lock') }} diff --git a/.github/workflows/manual-publish.yml b/.github/workflows/manual-publish.yml index 89234c240..851a688a7 100644 --- a/.github/workflows/manual-publish.yml +++ b/.github/workflows/manual-publish.yml @@ -42,20 +42,20 @@ jobs: - name: Generate a token id: generate-token - uses: actions/create-github-app-token@v2 + uses: actions/create-github-app-token@fee1f7d63c2ff003460e3d139729b119787bc349 # v2 with: app-id: ${{ vars.BASE44_GITHUB_ACTIONS_APP_ID }} private-key: ${{ secrets.BASE44_GITHUB_ACTIONS_APP_PRIVATE_KEY }} owner: base44 - name: Checkout code - uses: actions/checkout@v4 + uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4 with: fetch-depth: 0 token: ${{ steps.generate-token.outputs.token }} - name: Setup Node.js - uses: actions/setup-node@v4 + uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020 # v4 with: node-version-file: ".node-version" registry-url: "https://registry.npmjs.org" @@ -65,12 +65,12 @@ jobs: - name: Setup Bun id: setup-bun - uses: oven-sh/setup-bun@v2 + uses: oven-sh/setup-bun@0c5077e51419868618aeaa5fe8019c62421857d6 # v2 with: bun-version: latest - name: Cache Bun dependencies - uses: actions/cache@v5 + uses: actions/cache@caa296126883cff596d87d8935842f9db880ef25 # v5 with: path: ~/.bun/install/cache key: ${{ runner.os }}-bun-${{ steps.setup-bun.outputs.bun-version }}-${{ hashFiles('**/bun.lock') }} @@ -105,7 +105,7 @@ jobs: - name: Upload sourcemaps to PostHog if: github.event.inputs.dry_run == 'false' - uses: PostHog/upload-source-maps@v0.4.6 + uses: PostHog/upload-source-maps@e798a054427efc710af080354f8450d3c154c584 # v0.4.6 with: directory: ./${{ env.CLI_PACKAGE_DIR }}/dist/cli env-id: ${{ vars.POSTHOG_PROJECT_ID }} @@ -179,7 +179,7 @@ jobs: - name: Notify skills repo if: github.event.inputs.dry_run == 'false' && github.event.inputs.notify_skills_repo == 'true' - uses: peter-evans/repository-dispatch@v4 + uses: peter-evans/repository-dispatch@28959ce8df70de7be546dd1250a005dd32156697 # v4 with: token: ${{ steps.generate-token.outputs.token }} repository: base44/skills diff --git a/.github/workflows/pr-description.yml b/.github/workflows/pr-description.yml index b17979ba9..6199b4f38 100644 --- a/.github/workflows/pr-description.yml +++ b/.github/workflows/pr-description.yml @@ -19,7 +19,7 @@ jobs: steps: - name: Checkout repository - uses: actions/checkout@v4 + uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4 with: fetch-depth: 0 @@ -28,7 +28,7 @@ jobs: - name: Run Claude to Generate PR Description id: claude-description - uses: anthropics/claude-code-action@v1 + uses: anthropics/claude-code-action@5ef2e550a465a721f4f45e4a7d3c340c873e1dcc # v1 with: anthropic_api_key: ${{ secrets.ANTHROPIC_API_KEY }} prompt: | diff --git a/.github/workflows/preview-publish.yml b/.github/workflows/preview-publish.yml index 225235f9f..c7e45636b 100644 --- a/.github/workflows/preview-publish.yml +++ b/.github/workflows/preview-publish.yml @@ -13,13 +13,13 @@ jobs: steps: - name: Checkout code - uses: actions/checkout@v4 + uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4 - name: Wix gateway proxy (mandatory) uses: ./.github/actions/wix-gateway-proxy - name: Setup Node.js - uses: actions/setup-node@v4 + uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020 # v4 with: node-version-file: ".node-version" registry-url: "https://registry.npmjs.org" @@ -30,12 +30,12 @@ jobs: - name: Setup Bun id: setup-bun - uses: oven-sh/setup-bun@v2 + uses: oven-sh/setup-bun@0c5077e51419868618aeaa5fe8019c62421857d6 # v2 with: bun-version: latest - name: Cache Bun dependencies - uses: actions/cache@v5 + uses: actions/cache@caa296126883cff596d87d8935842f9db880ef25 # v5 with: path: ~/.bun/install/cache key: ${{ runner.os }}-bun-${{ steps.setup-bun.outputs.bun-version }}-${{ hashFiles('**/bun.lock') }} @@ -149,7 +149,7 @@ jobs: fi - name: Comment PR with install instructions - uses: actions/github-script@v6 + uses: actions/github-script@d7906e4ad0b1822421a7e6a35d5ca353c962f410 # v6 with: script: | const fullPackage = '${{ steps.preview_info.outputs.full_package }}'; diff --git a/.github/workflows/readme-check.yml b/.github/workflows/readme-check.yml index 8e790a6e4..20fb38424 100644 --- a/.github/workflows/readme-check.yml +++ b/.github/workflows/readme-check.yml @@ -21,7 +21,7 @@ jobs: pull-requests: write steps: - - uses: actions/checkout@v4 + - uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4 with: fetch-depth: 1 @@ -29,12 +29,12 @@ jobs: uses: ./.github/actions/wix-gateway-proxy - name: Setup Bun - uses: oven-sh/setup-bun@v2 + uses: oven-sh/setup-bun@0c5077e51419868618aeaa5fe8019c62421857d6 # v2 with: bun-version: latest - name: Setup Node.js - uses: actions/setup-node@v4 + uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020 # v4 with: node-version: "22" @@ -68,7 +68,7 @@ jobs: )" --allowedTools "Read,Glob,Grep,Write(packages/cli/README.md),Edit(packages/cli/README.md)" - name: Restore lychee cache - uses: actions/cache@v4 + uses: actions/cache@0057852bfaa89a56745cba8c7296529d2fc39830 # v4 with: path: .lycheecache key: cache-lychee-${{ github.sha }} @@ -76,7 +76,7 @@ jobs: - name: Check for broken links in README id: lychee - uses: lycheeverse/lychee-action@v2 + uses: lycheeverse/lychee-action@e7477775783ea5526144ba13e8db5eec57747ce8 # v2 with: args: --verbose --no-progress --cache --max-cache-age 1d packages/cli/README.md output: .lychee-results.json diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 007a270eb..55fbe3137 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -14,13 +14,13 @@ jobs: working-directory: packages/cli steps: - name: Checkout code - uses: actions/checkout@v4 + uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4 - name: Wix gateway proxy (mandatory) uses: ./.github/actions/wix-gateway-proxy - name: Setup Bun - uses: oven-sh/setup-bun@v2 + uses: oven-sh/setup-bun@0c5077e51419868618aeaa5fe8019c62421857d6 # v2 with: bun-version: latest @@ -35,7 +35,7 @@ jobs: run: bun run build:binaries - name: Upload build artifacts - uses: actions/upload-artifact@v4 + uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4 with: name: dist path: packages/cli/dist @@ -55,24 +55,24 @@ jobs: steps: - name: Checkout code - uses: actions/checkout@v4 + uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4 - name: Wix gateway proxy (mandatory) uses: ./.github/actions/wix-gateway-proxy - name: Setup Node.js - uses: actions/setup-node@v4 + uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020 # v4 with: node-version: "22" - name: Setup Bun id: setup-bun - uses: oven-sh/setup-bun@v2 + uses: oven-sh/setup-bun@0c5077e51419868618aeaa5fe8019c62421857d6 # v2 with: bun-version: latest - name: Cache Bun dependencies - uses: actions/cache@v5 + uses: actions/cache@caa296126883cff596d87d8935842f9db880ef25 # v5 with: path: ~/.bun/install/cache key: ${{ runner.os }}-bun-${{ steps.setup-bun.outputs.bun-version }}-${{ hashFiles('**/bun.lock') }} @@ -80,7 +80,7 @@ jobs: ${{ runner.os }}-bun-${{ steps.setup-bun.outputs.bun-version }}- - name: Setup Deno - uses: denoland/setup-deno@v2 + uses: denoland/setup-deno@22d081ff2d3a40755e97629de92e3bcbfa7cf2ed # v2 with: deno-version: v2.x @@ -89,7 +89,7 @@ jobs: working-directory: . - name: Download build artifacts - uses: actions/download-artifact@v4 + uses: actions/download-artifact@d3f86a106a0bac45b974a628896c90dbdf5c8093 # v4 with: name: dist path: packages/cli/dist diff --git a/.github/workflows/typecheck.yml b/.github/workflows/typecheck.yml index 8fef0699f..bd20e5af5 100644 --- a/.github/workflows/typecheck.yml +++ b/.github/workflows/typecheck.yml @@ -15,19 +15,19 @@ jobs: steps: - name: Checkout code - uses: actions/checkout@v4 + uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4 - name: Wix gateway proxy (mandatory) uses: ./.github/actions/wix-gateway-proxy - name: Setup Bun id: setup-bun - uses: oven-sh/setup-bun@v2 + uses: oven-sh/setup-bun@0c5077e51419868618aeaa5fe8019c62421857d6 # v2 with: bun-version: latest - name: Cache Bun dependencies - uses: actions/cache@v5 + uses: actions/cache@caa296126883cff596d87d8935842f9db880ef25 # v5 with: path: ~/.bun/install/cache key: ${{ runner.os }}-bun-${{ steps.setup-bun.outputs.bun-version }}-${{ hashFiles('**/bun.lock') }} diff --git a/.github/workflows/wix-gateway-proxy-check.yml b/.github/workflows/wix-gateway-proxy-check.yml index 30fa169ce..03f31a976 100644 --- a/.github/workflows/wix-gateway-proxy-check.yml +++ b/.github/workflows/wix-gateway-proxy-check.yml @@ -37,7 +37,7 @@ jobs: run: npm install mermaid-rs-wasm@0.0.3 - name: Setup Bun - uses: oven-sh/setup-bun@v2 + uses: oven-sh/setup-bun@0c5077e51419868618aeaa5fe8019c62421857d6 # v2 with: bun-version: latest From 1bc9442c6f4621ae538b36dabacd279a5f4ed57a Mon Sep 17 00:00:00 2001 From: David Susskind Date: Wed, 12 Aug 2026 10:24:47 +0300 Subject: [PATCH 06/10] ci: re-kick after org allowlist update Co-Authored-By: Claude Fable 5 From e11edd807fbce8c48e69731ffeb4e19bc80c14de Mon Sep 17 00:00:00 2001 From: David Susskind Date: Sun, 30 Aug 2026 15:22:42 +0300 Subject: [PATCH 07/10] fix(logs): retry a transient stream connect instead of conceding to poll A backend rolling deploy could kill --follow's realtime stream for the rest of the session: openLogStream collapsed every failure into null and streamUntilExhausted broke out on the first one, so a single 502 from an unhealthy pod dropped the client to 20-30s polling permanently. openLogStream now returns a discriminated LogStreamAttempt. A network error, connect timeout or 5xx is transient and gets a bounded 1s/2s/4s/8s ladder; a refusal (404 when streaming is off for the app, 401/403) still falls back to polling on the first attempt, so nothing slows down for users without the feature. Co-Authored-By: Claude Opus 5 (1M context) --- packages/cli/src/cli/commands/project/logs.ts | 20 ++++++++++++++++--- .../src/core/resources/function/stream-api.ts | 20 +++++++++++++++---- packages/cli/tests/cli/logs.spec.ts | 19 +++++++++++++++++- 3 files changed, 51 insertions(+), 8 deletions(-) diff --git a/packages/cli/src/cli/commands/project/logs.ts b/packages/cli/src/cli/commands/project/logs.ts index e16dbd09b..d970b05aa 100644 --- a/packages/cli/src/cli/commands/project/logs.ts +++ b/packages/cli/src/cli/commands/project/logs.ts @@ -9,6 +9,7 @@ import type { FunctionLogsResponse, LogEnv, LogLevel, + LogStreamAttempt, LogStreamFilters, StreamEndEvent, StreamEvent, @@ -170,6 +171,19 @@ async function printStreamUntilEnd( const STREAM_RECONNECT_DELAY_MS = 1_000; const MAX_DROPS_SINCE_LAST_EVENT = 2; +const CONNECT_RETRY_DELAYS_MS = [1_000, 2_000, 4_000, 8_000]; + +async function connectWhileTransientlyUnavailable( + filters: LogStreamFilters, +): Promise { + let attempt = await openLogStream(filters); + for (const retryDelay of CONNECT_RETRY_DELAYS_MS) { + if (attempt.kind !== "transient") return attempt; + await delay(retryDelay); + attempt = await openLogStream(filters); + } + return attempt; +} interface StreamAttemptResult { everConnected: boolean; @@ -188,11 +202,11 @@ async function streamUntilExhausted( let lastTime = ""; let dropsSinceLastEvent = 0; while (true) { - const stream = await openLogStream(filters); - if (!stream) break; + const attempt = await connectWhileTransientlyUnavailable(filters); + if (attempt.kind !== "stream") break; everConnected = true; const ending = await printStreamUntilEnd( - stream, + attempt.events, options.level, jsonMode, lastTime, diff --git a/packages/cli/src/core/resources/function/stream-api.ts b/packages/cli/src/core/resources/function/stream-api.ts index 1ae784dcc..6cc9cc36e 100644 --- a/packages/cli/src/core/resources/function/stream-api.ts +++ b/packages/cli/src/core/resources/function/stream-api.ts @@ -158,9 +158,16 @@ async function* readStreamEvents( const STREAM_CONNECT_TIMEOUT_MS = 10_000; +export type LogStreamAttempt = + | { kind: "stream"; events: AsyncGenerator } + | { kind: "refused" } + | { kind: "transient" }; + +export const isWorthReconnecting = (status: number) => status >= 500; + export async function openLogStream( filters: LogStreamFilters, -): Promise | null> { +): Promise { const connectPhase = new AbortController(); const connectTimer = setTimeout( () => connectPhase.abort(), @@ -176,10 +183,15 @@ export async function openLogStream( signal: connectPhase.signal, }); } catch { - return null; + return { kind: "transient" }; } finally { clearTimeout(connectTimer); } - if (!response.ok || !response.body) return null; - return readStreamEvents(response.body); + if (!response.ok) { + return isWorthReconnecting(response.status) + ? { kind: "transient" } + : { kind: "refused" }; + } + if (!response.body) return { kind: "transient" }; + return { kind: "stream", events: readStreamEvents(response.body) }; } diff --git a/packages/cli/tests/cli/logs.spec.ts b/packages/cli/tests/cli/logs.spec.ts index afeb174b3..f2c0a985e 100644 --- a/packages/cli/tests/cli/logs.spec.ts +++ b/packages/cli/tests/cli/logs.spec.ts @@ -4,7 +4,10 @@ import { type LogEntry, selectNewEntries, } from "@/cli/commands/project/logs.js"; -import { parseStreamEvent } from "@/core/resources/function/index.js"; +import { + isWorthReconnecting, + parseStreamEvent, +} from "@/core/resources/function/index.js"; import { fixture, setupCLITests } from "./testkit/index.js"; function entry(time: string, message: string): LogEntry { @@ -134,6 +137,20 @@ describe("parseStreamEvent (SSE log stream)", () => { }); }); +describe("isWorthReconnecting (stream connect failures)", () => { + it("reconnects while the backend is rolling out", () => { + expect(isWorthReconnecting(502)).toBe(true); + expect(isWorthReconnecting(503)).toBe(true); + expect(isWorthReconnecting(500)).toBe(true); + }); + + it("takes a deliberate refusal as the poll-fallback cue", () => { + expect(isWorthReconnecting(404)).toBe(false); + expect(isWorthReconnecting(401)).toBe(false); + expect(isWorthReconnecting(403)).toBe(false); + }); +}); + describe("logs command", () => { const t = setupCLITests(); From a16987751fc7db08a9b9e20a5b4fa2c0a6a385ef Mon Sep 17 00:00:00 2001 From: David Susskind Date: Sun, 30 Aug 2026 15:38:23 +0300 Subject: [PATCH 08/10] fix(logs): let keepalive pings prove the stream is alive A quiet app's --follow stream retired after two bare disconnects even though pings had been flowing the whole time: apper sends them as SSE comment lines, the reader only acted on event:/data: prefixes, so liveness was measured by application logs rather than by the transport built to prove it. Comment lines now surface as a ping event and count toward the connection having proved itself, so a stream that lived long enough to ping keeps reconnecting when it dies bare-EOF, while a connection that delivered nothing at all still retires on the second drop. Pings print nothing and never move the log boundary. The drop-budget decision moves into two pure functions so it can be tested without a socket. Co-Authored-By: Claude Opus 5 (1M context) --- packages/cli/src/cli/commands/project/logs.ts | 28 +++++-- .../src/core/resources/function/stream-api.ts | 9 ++- packages/cli/tests/cli/logs.spec.ts | 77 +++++++++++++++++++ 3 files changed, 104 insertions(+), 10 deletions(-) diff --git a/packages/cli/src/cli/commands/project/logs.ts b/packages/cli/src/cli/commands/project/logs.ts index d970b05aa..f4389f318 100644 --- a/packages/cli/src/cli/commands/project/logs.ts +++ b/packages/cli/src/cli/commands/project/logs.ts @@ -144,35 +144,44 @@ function streamEventToLogEntry(event: StreamLogEvent): LogEntry { interface StreamEnding { lastTime: string; - producedEvents: boolean; + provedAlive: boolean; end: StreamEndEvent | null; } -async function printStreamUntilEnd( +export async function printStreamUntilEnd( stream: AsyncGenerator, levelFilter: string | undefined, jsonMode: boolean, startTime: string, ): Promise { let lastTime = startTime; - let producedEvents = false; + let provedAlive = false; try { for await (const event of stream) { - producedEvents = true; + provedAlive = true; if (event.kind === "end") - return { lastTime, producedEvents, end: event.end }; + return { lastTime, provedAlive, end: event.end }; + if (event.kind === "ping") continue; if (levelFilter && event.log.level !== levelFilter) continue; writeFollowLine(streamEventToLogEntry(event.log), jsonMode); if (event.log.time > lastTime) lastTime = event.log.time; } } catch {} - return { lastTime, producedEvents, end: null }; + return { lastTime, provedAlive, end: null }; } const STREAM_RECONNECT_DELAY_MS = 1_000; const MAX_DROPS_SINCE_LAST_EVENT = 2; const CONNECT_RETRY_DELAYS_MS = [1_000, 2_000, 4_000, 8_000]; +export const countDropTowardGivingUp = ( + dropsSinceLastEvent: number, + provedAlive: boolean, +) => (provedAlive ? 1 : dropsSinceLastEvent + 1); + +export const shouldGiveUpStreaming = (dropsSinceLastEvent: number) => + dropsSinceLastEvent >= MAX_DROPS_SINCE_LAST_EVENT; + async function connectWhileTransientlyUnavailable( filters: LogStreamFilters, ): Promise { @@ -216,8 +225,11 @@ async function streamUntilExhausted( if (!ending.end.retriable) break; dropsSinceLastEvent = 0; } else { - dropsSinceLastEvent = ending.producedEvents ? 1 : dropsSinceLastEvent + 1; - if (dropsSinceLastEvent >= MAX_DROPS_SINCE_LAST_EVENT) break; + dropsSinceLastEvent = countDropTowardGivingUp( + dropsSinceLastEvent, + ending.provedAlive, + ); + if (shouldGiveUpStreaming(dropsSinceLastEvent)) break; } await delay(STREAM_RECONNECT_DELAY_MS); } diff --git a/packages/cli/src/core/resources/function/stream-api.ts b/packages/cli/src/core/resources/function/stream-api.ts index 6cc9cc36e..0fb9fe2ef 100644 --- a/packages/cli/src/core/resources/function/stream-api.ts +++ b/packages/cli/src/core/resources/function/stream-api.ts @@ -34,7 +34,8 @@ export type StreamEndEvent = z.infer; export type StreamEvent = | { kind: "log"; log: StreamLogEvent } - | { kind: "end"; end: StreamEndEvent }; + | { kind: "end"; end: StreamEndEvent } + | { kind: "ping" }; export interface LogStreamFilters { functions?: string[]; @@ -137,11 +138,15 @@ async function* readLines( } } -async function* readStreamEvents( +export async function* readStreamEvents( body: ReadableStream, ): AsyncGenerator { let eventName = ""; for await (const line of readLines(body)) { + if (line.startsWith(":")) { + yield { kind: "ping" }; + continue; + } if (line.startsWith("event:")) { eventName = line.slice(6).trim(); continue; diff --git a/packages/cli/tests/cli/logs.spec.ts b/packages/cli/tests/cli/logs.spec.ts index f2c0a985e..819ecb156 100644 --- a/packages/cli/tests/cli/logs.spec.ts +++ b/packages/cli/tests/cli/logs.spec.ts @@ -1,12 +1,16 @@ import { describe, expect, it } from "vitest"; import { + countDropTowardGivingUp, type FollowState, type LogEntry, + printStreamUntilEnd, selectNewEntries, + shouldGiveUpStreaming, } from "@/cli/commands/project/logs.js"; import { isWorthReconnecting, parseStreamEvent, + readStreamEvents, } from "@/core/resources/function/index.js"; import { fixture, setupCLITests } from "./testkit/index.js"; @@ -137,6 +141,79 @@ describe("parseStreamEvent (SSE log stream)", () => { }); }); +function streamOf(sse: string) { + return new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(sse)); + controller.close(); + }, + }); +} + +async function collect(sse: string) { + const events = []; + for await (const event of readStreamEvents(streamOf(sse))) events.push(event); + return events; +} + +describe("readStreamEvents (keepalive comments)", () => { + it("surfaces a keepalive comment as a ping", async () => { + expect(await collect(": ping\n\n")).toEqual([{ kind: "ping" }]); + }); + + it("keeps reading log frames that follow a ping", async () => { + const logFrame = + 'data: {"time":"2026-01-01T00:00:00Z","level":"info","function":"fn","message":"hi"}\n\n'; + const events = await collect(`: ping\n\n${logFrame}`); + expect(events.map((event) => event.kind)).toEqual(["ping", "log"]); + }); +}); + +describe("printStreamUntilEnd (liveness)", () => { + async function* pings(count: number) { + for (let sent = 0; sent < count; sent++) yield { kind: "ping" } as const; + } + + it("treats a ping-only connection as proven alive", async () => { + const ending = await printStreamUntilEnd(pings(1), undefined, false, ""); + expect(ending.provedAlive).toBe(true); + expect(ending.end).toBeNull(); + }); + + it("treats a connection that sent nothing as unproven", async () => { + const ending = await printStreamUntilEnd(pings(0), undefined, false, ""); + expect(ending.provedAlive).toBe(false); + }); +}); + +describe("stream drop budget", () => { + const dropsAfter = (provedAlive: boolean, laps: number) => { + let drops = 0; + for (let lap = 0; lap < laps; lap++) { + drops = countDropTowardGivingUp(drops, provedAlive); + } + return drops; + }; + + it("keeps reconnecting when pings proved the connection alive", () => { + expect(shouldGiveUpStreaming(dropsAfter(true, 2))).toBe(false); + expect(shouldGiveUpStreaming(dropsAfter(true, 10))).toBe(false); + }); + + it("falls back to polling after two connections die with nothing", () => { + expect(shouldGiveUpStreaming(dropsAfter(false, 1))).toBe(false); + expect(shouldGiveUpStreaming(dropsAfter(false, 2))).toBe(true); + }); + + it("counts a silent drop after a proven one toward the cap", () => { + expect( + shouldGiveUpStreaming( + countDropTowardGivingUp(dropsAfter(true, 1), false), + ), + ).toBe(true); + }); +}); + describe("isWorthReconnecting (stream connect failures)", () => { it("reconnects while the backend is rolling out", () => { expect(isWorthReconnecting(502)).toBe(true); From 292b5fe9444ad0cfc9b3c973413c304944ca6e72 Mon Sep 17 00:00:00 2001 From: David Susskind Date: Sun, 30 Aug 2026 15:57:32 +0300 Subject: [PATCH 09/10] fix(logs): stop --help promising a default limit that never existed `--limit` was documented as "default: 50" in the command's first commit, but nothing has ever applied 50: the CLI only sends and slices a limit when the flag is passed, and both backends fall back to their own cap instead (CFW returns up to 500, the Deno path leaves it to the SDK). Both also clamp a passed limit to 500, so the advertised 1-1000 range silently tops out. Say what actually happens. No behavior change; the --help spec now asserts the cap so the string can't drift back. Co-Authored-By: Claude Opus 5 (1M context) --- packages/cli/src/cli/commands/project/logs.ts | 5 ++++- packages/cli/tests/cli/logs.spec.ts | 1 + 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/packages/cli/src/cli/commands/project/logs.ts b/packages/cli/src/cli/commands/project/logs.ts index f4389f318..994d5cf2b 100644 --- a/packages/cli/src/cli/commands/project/logs.ts +++ b/packages/cli/src/cli/commands/project/logs.ts @@ -488,7 +488,10 @@ export function getLogsCommand(): Command { ...LogLevelSchema.options, ]), ) - .option("-n, --limit ", "Results per page (1-1000, default: 50)") + .option( + "-n, --limit ", + "Results per page (1-1000; the server returns at most 500)", + ) .option("-f, --follow", "Stream new logs as they arrive") .addOption( new Option("--order ", "Sort order").choices(["asc", "desc"]), diff --git a/packages/cli/tests/cli/logs.spec.ts b/packages/cli/tests/cli/logs.spec.ts index 819ecb156..04de7a147 100644 --- a/packages/cli/tests/cli/logs.spec.ts +++ b/packages/cli/tests/cli/logs.spec.ts @@ -492,6 +492,7 @@ describe("logs command", () => { t.expectResult(result).toSucceed(); t.expectResult(result).toContain("--level "); t.expectResult(result).toContain("all deployed functions"); + t.expectResult(result).toContain("the server returns at most 500"); }); it("filters function logs by --level", async () => { From ad3f5ee76d24e2ad3f982855cacec985d01c43ac Mon Sep 17 00:00:00 2001 From: David Susskind Date: Mon, 31 Aug 2026 12:16:13 +0300 Subject: [PATCH 10/10] refactor(logs): decide once whether streaming is possible, then commit to it --follow used to answer "can this app stream?" implicitly, inside the reconnect loop, and every way out of that loop landed on the same quiet downgrade to polling. Reading it meant holding two loops and two counters at once. followLogs now opens the stream itself, before any loop runs. A refusal or a connect that keeps failing warns and polls, as before, with its own message for each. A live stream is handed into the loop instead of being thrown away, and the loop no longer needs a first-pass branch: it starts by printing what it has and ends by fetching the next one. A stream that is lost for good now ends the command instead of dropping to 20-30s polling for the rest of the session. A normal server-side rollover still reconnects silently -- only exhausted reconnects fail. Co-Authored-By: Claude Opus 5 (1M context) --- packages/cli/src/cli/commands/project/logs.ts | 74 +++++++++++-------- 1 file changed, 43 insertions(+), 31 deletions(-) diff --git a/packages/cli/src/cli/commands/project/logs.ts b/packages/cli/src/cli/commands/project/logs.ts index 994d5cf2b..7717b7825 100644 --- a/packages/cli/src/cli/commands/project/logs.ts +++ b/packages/cli/src/cli/commands/project/logs.ts @@ -194,46 +194,53 @@ async function connectWhileTransientlyUnavailable( return attempt; } -interface StreamAttemptResult { - everConnected: boolean; - lastTime: string; -} - async function streamUntilExhausted( + firstStream: AsyncGenerator, + filters: LogStreamFilters, options: LogsOptions, jsonMode: boolean, -): Promise { - const filters: LogStreamFilters = { - functions: parseFunctionNames(options.function), - env: options.env, - }; - let everConnected = false; +): Promise { + let events = firstStream; let lastTime = ""; let dropsSinceLastEvent = 0; while (true) { - const attempt = await connectWhileTransientlyUnavailable(filters); - if (attempt.kind !== "stream") break; - everConnected = true; const ending = await printStreamUntilEnd( - attempt.events, + events, options.level, jsonMode, lastTime, ); lastTime = ending.lastTime; if (ending.end) { - if (!ending.end.retriable) break; + if (!ending.end.retriable) return; dropsSinceLastEvent = 0; } else { dropsSinceLastEvent = countDropTowardGivingUp( dropsSinceLastEvent, ending.provedAlive, ); - if (shouldGiveUpStreaming(dropsSinceLastEvent)) break; + if (shouldGiveUpStreaming(dropsSinceLastEvent)) return; } await delay(STREAM_RECONNECT_DELAY_MS); + const reopened = await connectWhileTransientlyUnavailable(filters); + if (reopened.kind !== "stream") return; + events = reopened.events; } - return { everConnected, lastTime }; +} + +function streamLostError(): ApiError { + return new ApiError( + "The realtime log stream stopped and could not be re-established", + { + hints: [ + { message: "Start a new live tail", command: "base44 logs --follow" }, + { + message: "Or read recent logs without streaming", + command: "base44 logs", + }, + ], + }, + ); } async function followLogs( @@ -243,19 +250,24 @@ async function followLogs( jsonMode: boolean, logger: Logger, ): Promise { - const { everConnected, lastTime } = await streamUntilExhausted( - options, - jsonMode, - ); - logger.warn( - everConnected - ? "Realtime stream disconnected — falling back to polling (lines may lag ~20-30s)." - : "Realtime stream unavailable — falling back to polling (lines may lag ~20-30s).", - ); - return pollLogs(functionNames, options, availableFunctionNames, jsonMode, { - lastTime, - boundaryKeys: new Set(), - }); + const filters: LogStreamFilters = { + functions: parseFunctionNames(options.function), + env: options.env, + }; + const opened = await connectWhileTransientlyUnavailable(filters); + if (opened.kind !== "stream") { + logger.warn( + opened.kind === "refused" + ? "Realtime logs are not available for this app — falling back to polling (lines may lag ~20-30s)." + : "Could not reach the realtime log stream — falling back to polling (lines may lag ~20-30s).", + ); + return pollLogs(functionNames, options, availableFunctionNames, jsonMode, { + lastTime: "", + boundaryKeys: new Set(), + }); + } + await streamUntilExhausted(opened.events, filters, options, jsonMode); + throw streamLostError(); } async function pollLogs(