From 5ccb2fb6b3e04f2374c510f28fd675d34d2f1318 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sat, 31 Oct 2026 01:28:10 -0700 Subject: [PATCH] fix(hub): stop swallowing errors in agent runtime paths --- apps/hub/src/artifacts/design-draft.ts | 22 +-- apps/hub/src/artifacts/materials.ts | 5 +- apps/hub/src/artifacts/routes.ts | 19 +- apps/hub/src/artifacts/service.ts | 58 +++--- apps/hub/src/builds/package-layout.ts | 8 +- apps/hub/src/lib/json.ts | 21 ++ apps/hub/src/lib/serial-queue.ts | 14 +- apps/hub/src/live-app/host.ts | 192 +++++++++++++------ apps/hub/src/live-app/proxy.ts | 15 +- apps/hub/src/live-app/runner.ts | 76 +++++--- apps/hub/src/mail/caller-mail.ts | 3 +- apps/hub/src/mail/committed-agent-state.ts | 11 +- apps/hub/src/mail/conversation-routes.ts | 9 +- apps/hub/src/mail/conversation-threads.ts | 7 +- apps/hub/src/mail/conversation-turns.ts | 16 +- apps/hub/src/mail/coordinator-question.ts | 47 +++-- apps/hub/src/mail/hub-mail.ts | 3 +- apps/hub/src/projects/coordinator-resume.ts | 24 ++- apps/hub/src/projects/coordinator-startup.ts | 18 +- apps/hub/src/projects/coordinator-upgrade.ts | 14 +- apps/hub/src/projects/next-step.ts | 47 +++-- apps/hub/src/projects/routes.ts | 52 +++-- apps/hub/src/projects/specialist-ended.ts | 17 +- apps/hub/src/request-error.ts | 8 + apps/hub/src/run-stream.ts | 29 ++- apps/hub/src/sidecar/local-provisioner.ts | 10 +- apps/hub/src/sidecar/manifest-store.ts | 8 +- apps/hub/src/sidecar/spawn.ts | 33 ++-- 28 files changed, 478 insertions(+), 308 deletions(-) create mode 100644 apps/hub/src/lib/json.ts diff --git a/apps/hub/src/artifacts/design-draft.ts b/apps/hub/src/artifacts/design-draft.ts index b353f127..e2cba032 100644 --- a/apps/hub/src/artifacts/design-draft.ts +++ b/apps/hub/src/artifacts/design-draft.ts @@ -3,7 +3,9 @@ import { open, realpath } from "node:fs/promises"; import { dirname, join } from "node:path"; import { type } from "arktype"; import { workflowRunRepoIdForAddress } from "@intx/hub-sessions"; +import { hasCode } from "@intx/types"; import { packApp } from "../builds/app-snapshot.ts"; +import { parseJson } from "../lib/json.ts"; import { ProjectError } from "../projects/errors.ts"; import { DESIGN_STYLES } from "./design-tweaks.ts"; import { designStarterFiles } from "@corbits-builder/designer/skills" with { type: "macro" }; @@ -28,9 +30,6 @@ export function warmStepWorkspace( ); } -const isMissing = (error: unknown) => - error instanceof Error && "code" in error && error.code === "ENOENT"; - // The mockup the designer is writing, keyed as the design bundle keys it (designs/...), or // undefined until it has an index.html. Neither .builder nor designs may be a symlink, and // packApp reads only regular files, so nothing outside .builder/designs is served. @@ -41,7 +40,7 @@ export async function readDesignDraft(workspace: string) { try { actual = await realpath(designs); } catch (error) { - if (isMissing(error)) return undefined; + if (hasCode(error) && error.code === "ENOENT") return undefined; throw error; } if (actual !== join(await realpath(workspace), ".builder", "designs")) @@ -89,17 +88,14 @@ async function readWorkspaceText( ) return undefined; } catch (error) { - if (isMissing(error)) return undefined; + if (hasCode(error) && error.code === "ENOENT") return undefined; throw error; } let file; try { file = await open(target, constants.O_RDONLY | constants.O_NOFOLLOW); } catch (error) { - if ( - isMissing(error) || - (error instanceof Error && "code" in error && error.code === "ELOOP") - ) + if (hasCode(error) && (error.code === "ENOENT" || error.code === "ELOOP")) return undefined; throw error; } @@ -165,13 +161,7 @@ export function demoLogin(files: Readonly>) { const text = files[DEMO_ACCOUNT]; if (text === undefined || text.length > MAX_DEMO_ACCOUNT_BYTES) return undefined; - let json: unknown; - try { - json = JSON.parse(text); - } catch { - return undefined; - } - const account = DemoAccount(json); + const account = DemoAccount(parseJson(text)); return account instanceof type.errors ? undefined : { email: account.email, password: account.password }; diff --git a/apps/hub/src/artifacts/materials.ts b/apps/hub/src/artifacts/materials.ts index 472a5940..bcc58315 100644 --- a/apps/hub/src/artifacts/materials.ts +++ b/apps/hub/src/artifacts/materials.ts @@ -47,8 +47,9 @@ export function materialMimeType(file: { name: string; type: string }): string { function utf8(bytes: Uint8Array): string | undefined { try { return new TextDecoder("utf-8", { fatal: true }).decode(bytes); - } catch { - return undefined; + } catch (error) { + if (error instanceof TypeError) return undefined; + throw error; } } diff --git a/apps/hub/src/artifacts/routes.ts b/apps/hub/src/artifacts/routes.ts index ff8b69c1..a7646ee2 100644 --- a/apps/hub/src/artifacts/routes.ts +++ b/apps/hub/src/artifacts/routes.ts @@ -34,6 +34,7 @@ import { } from "@corbits-builder/schemas/project"; import { actorOf } from "../projects/authorization.ts"; import { ProjectError } from "../projects/errors.ts"; +import { readJson } from "../lib/json.ts"; import { mailRunAsCaller } from "../mail/caller-mail.ts"; import type { createArtifactService } from "./service.ts"; @@ -51,11 +52,9 @@ export function mountArtifactRoutes( nativeFetch?: (request: Request) => Promise, ) { const body = async (c: { req: { json(): Promise } }) => { - try { - return await c.req.json(); - } catch { - throw new ProjectError(400, "invalid_json"); - } + const json = await readJson(c.req); + if (json === undefined) throw new ProjectError(400, "invalid_json"); + return json; }; const failure = ( c: { @@ -419,9 +418,13 @@ export function mountArtifactRoutes( async (c) => { try { const current = actorOf(c); - const form = await c.req.formData().catch(() => { - throw new ProjectError(400, "invalid_form"); - }); + let form: FormData; + try { + form = await c.req.formData(); + } catch (error) { + if (!(error instanceof TypeError)) throw error; + throw new ProjectError(400, "invalid_form", {}, { cause: error }); + } const message = form.get("message"); return c.json( await service.attachMaterials( diff --git a/apps/hub/src/artifacts/service.ts b/apps/hub/src/artifacts/service.ts index c5558489..2d1ccecc 100644 --- a/apps/hub/src/artifacts/service.ts +++ b/apps/hub/src/artifacts/service.ts @@ -17,7 +17,12 @@ import { } from "@corbits/artifacts"; import type { DB } from "@intx/db"; import { getLogger } from "@intx/log"; -import { base64Decode, base64Encode, validateAttachments } from "@intx/types"; +import { + base64Decode, + base64Encode, + hasCode, + validateAttachments, +} from "@intx/types"; import { WORKFLOW_RUN_REF, workflowRunRepoIdForAddress, @@ -89,6 +94,7 @@ import { } from "../db/schema.ts"; import type { AgentActivityTracker } from "./activity.ts"; import type { Actor } from "../lib/actor.ts"; +import { parseJson } from "../lib/json.ts"; import { authorizeProject } from "../projects/authorization.ts"; import { ProjectError, projectResource } from "../projects/errors.ts"; import { readAppRepository } from "../projects/app-repository.ts"; @@ -162,19 +168,20 @@ const appTooLarge = (error: unknown) => }) : error; -const parseBundle = (content: string): unknown => { - try { - return JSON.parse(content); - } catch { - return undefined; - } -}; - // An imported bundle's paths and a material's filename are the importer's text: none may climb // out of the directory it is archived under. const isContainedPath = (path: string) => !path.startsWith("/") && !path.split(/[\\/]/).includes(".."); +async function isDirectory(path: string) { + try { + return (await stat(path)).isDirectory(); + } catch (error) { + if (hasCode(error) && error.code === "ENOENT") return false; + throw error; + } +} + const artifactResource = (kind: string) => "sb-artifact:" + kind; // A document version records the open questions it leaves for the person; a version filed without @@ -359,13 +366,7 @@ export function createArtifactService( if (!sidecarDataDir || run.address === null) return unavailable("not_on_local_sidecar"); const dir = warmStepWorkspace(sidecarDataDir, run.address, run.stepId); - if ( - !(await stat(dir).then( - (found) => found.isDirectory(), - () => false, - )) - ) - return unavailable("workspace_missing"); + if (!(await isDirectory(dir))) return unavailable("workspace_missing"); if (!(await Bun.file(join(dir, "package.json")).exists())) return unavailable("package_json_missing"); return { runId: run.runId, dir }; @@ -465,7 +466,7 @@ export function createArtifactService( return stored ? { name: `v${filed.version}`, - bundle: AppBundle.assert(parseBundle(stored.content)), + bundle: AppBundle.assert(parseJson(stored.content)), } : undefined; } @@ -945,7 +946,8 @@ export function createArtifactService( throw new ProjectError(404, "design_not_found"); if (filed) { const stored = await getArtifactVersion(db, filed.id, filed.version); - const bundle = DesignBundle(parseBundle(stored?.content ?? "")); + if (!stored) throw new ProjectError(409, "invalid_design"); + const bundle = DesignBundle(parseJson(stored.content)); if (bundle instanceof type.errors) throw new ProjectError(409, "invalid_design", { issues: bundle.summary, @@ -1275,7 +1277,7 @@ export function createArtifactService( } if (bundled) { const bundle = (input.kind === "app" ? AppBundle : DesignBundle)( - parseBundle(input.content), + parseJson(input.content), ); if (bundle instanceof type.errors) throw new ProjectError(400, "invalid_" + input.kind, { @@ -1650,7 +1652,7 @@ export function createArtifactService( zip.file(`planning/${artifact.kind}.md`, stored.content); } else { const files = Object.entries( - DesignBundle.assert(parseBundle(stored.content)).files, + DesignBundle.assert(parseJson(stored.content)).files, ); if (artifact.kind === "design") add("design", files); else if (isContainedPath(stored.title)) @@ -1730,16 +1732,18 @@ export function createArtifactService( } catch (error) { // The coordinator's token only reads runs, so a designer left without its concept would // hold a slot as a waiting designer; the hub stops it and the original error stands. - await cancelRun?.( - actor.tenantId, - input.runId, - "design_concept_unrecorded", - ).catch((cancelError: unknown) => + try { + await cancelRun?.( + actor.tenantId, + input.runId, + "design_concept_unrecorded", + ); + } catch (cancelError) { log.warn("Stopping designer run {runId} without its concept failed", { runId: input.runId, ...requestErrorMetadata(cancelError), - }), - ); + }); + } throw error; } const [stored] = await rawDB diff --git a/apps/hub/src/builds/package-layout.ts b/apps/hub/src/builds/package-layout.ts index 353818df..84d63fee 100644 --- a/apps/hub/src/builds/package-layout.ts +++ b/apps/hub/src/builds/package-layout.ts @@ -2,11 +2,9 @@ import { readFile, realpath } from "node:fs/promises"; import { dirname, join, resolve } from "node:path"; import { fileURLToPath } from "node:url"; import { createRequire } from "node:module"; +import { hasCode } from "@intx/types"; import { rgPath } from "@vscode/ripgrep"; -const isMissing = (error: unknown) => - error instanceof Error && "code" in error && error.code === "ENOENT"; - // Resolve with Bun's installed package graph; source closures and npm installs // can place the same dependency outside the importing package's node_modules. async function packageRoot(name: string, from: string) { @@ -18,7 +16,7 @@ async function packageRoot(name: string, from: string) { ); if (manifest.name === name) return { directory, manifest }; } catch (error) { - if (!isMissing(error)) throw error; + if (!(hasCode(error) && error.code === "ENOENT")) throw error; } const parent = dirname(directory); if (parent === directory) @@ -62,7 +60,7 @@ export async function buildPackageLayout() { if ((await realpath(join(path, name))) === entry.directory) roots.add(await realpath(path)); } catch (error) { - if (!isMissing(error)) throw error; + if (!(hasCode(error) && error.code === "ENOENT")) throw error; } } if (known) continue; diff --git a/apps/hub/src/lib/json.ts b/apps/hub/src/lib/json.ts new file mode 100644 index 00000000..5fb34b51 --- /dev/null +++ b/apps/hub/src/lib/json.ts @@ -0,0 +1,21 @@ +/** A body's JSON, or undefined when it is not JSON; any other read failure propagates. */ +export async function readJson(body: { + json(): Promise; +}): Promise { + try { + return await body.json(); + } catch (error) { + if (error instanceof SyntaxError) return undefined; + throw error; + } +} + +/** The text's JSON, or undefined when it is not JSON. */ +export function parseJson(text: string): unknown { + try { + return JSON.parse(text); + } catch (error) { + if (error instanceof SyntaxError) return undefined; + throw error; + } +} diff --git a/apps/hub/src/lib/serial-queue.ts b/apps/hub/src/lib/serial-queue.ts index 84066e84..e6b2402e 100644 --- a/apps/hub/src/lib/serial-queue.ts +++ b/apps/hub/src/lib/serial-queue.ts @@ -1,18 +1,20 @@ +// The caller observes the failure; the queue only needs ordering. export async function settle( operation: Promise | undefined, ): Promise { - try { - await operation; - } catch { - // The caller observes the failure; the queue only needs ordering. - } + await Promise.allSettled([operation]); +} + +async function runAfter(previous: Promise, task: () => Promise) { + await previous; + return task(); } export function createSerialQueue() { let tail: Promise = Promise.resolve(); return { enqueue(task: () => Promise): Promise { - const result = tail.then(task); + const result = runAfter(tail, task); tail = settle(result); return result; }, diff --git a/apps/hub/src/live-app/host.ts b/apps/hub/src/live-app/host.ts index f6768aae..6eea154a 100644 --- a/apps/hub/src/live-app/host.ts +++ b/apps/hub/src/live-app/host.ts @@ -5,11 +5,14 @@ // they change, and `bun run dev` serves every later change. The app's state goes to stdout as one // JSON object per line whenever it changes. Closing stdin or SIGTERM ends it. import { createHash } from "node:crypto"; -import { watch, type FSWatcher } from "node:fs"; -import { access, readFile, rm, writeFile } from "node:fs/promises"; +import { existsSync, watch, type FSWatcher } from "node:fs"; +import { readFile, rename, rm, writeFile } from "node:fs/promises"; import { basename, dirname, join, sep } from "node:path"; import { type } from "arktype"; +import { hasCode } from "@intx/types"; import { liveAppWording } from "../config/live-app.ts"; +import { errorCode } from "../request-error.ts"; +import { parseJson } from "../lib/json.ts"; import { writeBundle } from "../artifacts/design-tools.ts"; import { bundleFiles, collectApp } from "../builds/app-snapshot.ts"; import { buildPackageLayout } from "../builds/package-layout.ts"; @@ -42,35 +45,49 @@ const report = (state: HostReport) => { last = line; process.stdout.write(line + "\n"); }; -const executor = await createPosixBuildExecutor( - workspace, - { - commandTimeoutMs: INSTALL_TIMEOUT_MS + 10_000, - maxToolResultBytes: 256 * 1024, - }, - await buildPackageLayout(), - // The person's running app installs its packages and reaches the services it was built against. - { openNetwork: true }, -).catch((error: unknown) => { - report({ - state: "failed", - error: error instanceof Error ? error.message : String(error), - log: [], - }); - process.exit(1); -}); +const message = (error: unknown) => + error instanceof Error ? error.message : String(error); + +const stderrLine = (phrase: string, error: unknown) => { + const { errorCode: code } = errorCode(error); + process.stderr.write( + code === undefined ? `${phrase}\n` : `${phrase} (${code})\n`, + ); +}; + +async function createExecutor() { + const layout = await buildPackageLayout(); + try { + return await createPosixBuildExecutor( + workspace, + { + commandTimeoutMs: INSTALL_TIMEOUT_MS + 10_000, + maxToolResultBytes: 256 * 1024, + }, + layout, + // The person's running app installs its packages and reaches the services it was built against. + { openNetwork: true }, + ); + } catch (error) { + report({ state: "failed", error: message(error), log: [] }); + process.exit(1); + } +} +const executor = await createExecutor(); // What this host last copied from a source, by digest, kept beside the workspace so a later host // removes a file the source dropped while no one looked, and never a file the app wrote itself. const manifest = join(dirname(workspace), "mirrored.json"); -const mirrored = new Map( - Object.entries( - await readFile(manifest, "utf8").then( - (text) => JSON.parse(text) as Record, - () => ({}), - ), - ), -); +async function readMirrored(): Promise> { + if (!(await Bun.file(manifest).exists())) return {}; + const recorded = parseJson(await readFile(manifest, "utf8")); + if (recorded !== undefined) return recorded as Record; + process.stderr.write( + "The mirror record is not JSON; mirroring from scratch.\n", + ); + return {}; +} +const mirrored = new Map(Object.entries(await readMirrored())); const contained = (path: string) => join(workspace, path).startsWith(workspace + sep); @@ -94,15 +111,16 @@ async function mirror(files: Readonly>) { mirrored.delete(path); changed.push(path); } - if (changed.length) - await writeFile(manifest, JSON.stringify(Object.fromEntries(mirrored))); + if (changed.length) { + // Renamed into place, so a host stopped mid-write never leaves a half-written record. + const staged = manifest + ".tmp"; + await writeFile(staged, JSON.stringify(Object.fromEntries(mirrored))); + await rename(staged, manifest); + } return changed; } -let installed = await access(join(workspace, "node_modules")).then( - () => true, - () => false, -); +let installed = existsSync(join(workspace, "node_modules")); let failure: { error: string; log: string[] } | undefined; let installing = false; // A crashed app is started again once; after that it waits for the next change to its files. @@ -120,16 +138,19 @@ async function apply(files: Readonly>) { if (changed.length) restartedSinceChange = false; if (!installed) { installing = true; - const result = await executor - .run( + let result; + try { + result = await executor.run( { id: "install", name: "run_shell", arguments: { command: "bun install", timeout: INSTALL_TIMEOUT_MS }, }, AbortSignal.timeout(INSTALL_TIMEOUT_MS + 5_000), - ) - .finally(() => (installing = false)); + ); + } finally { + installing = false; + } const output = String(result.content); if (result.isError || output.startsWith("exit code ")) { failure = { @@ -145,14 +166,17 @@ async function apply(files: Readonly>) { if (!app || !app.running || app.stopped !== undefined) await executor.app(); } -let queue: Promise = Promise.resolve(); +async function runAfter(previous: Promise, work: () => Promise) { + await previous; + try { + await work(); + } catch (error) { + failure = { error: message(error), log: [] }; + } +} +let queue: Promise = Promise.resolve(); const enqueue = (work: () => Promise) => { - queue = queue.then(work).catch((error: unknown) => { - failure = { - error: error instanceof Error ? error.message : String(error), - log: [], - }; - }); + queue = runAfter(queue, work); }; let watcher: FSWatcher | undefined; @@ -171,9 +195,15 @@ function follow(source: HostSource) { pending = setTimeout(sync, SYNC_DEBOUNCE_MS); }); // A build workspace goes away when its run ends; the hub then names the next source. - watcher.on("error", () => watcher?.close()); - } catch { + watcher.on("error", (error) => { + stderrLine("Watching the app's files stopped", error); + watcher?.close(); + }); + } catch (error) { watcher = undefined; + // A directory that is already gone fails the sync below, which reports it. + if (!(hasCode(error) && error.code === "ENOENT")) + stderrLine("Watching the app's files failed; the app syncs once", error); } sync(); } @@ -183,22 +213,66 @@ const close = async (code: number) => { if (closing) return; closing = true; watcher?.close(); - await executor.dispose().catch(() => undefined); + try { + await executor.dispose(); + } catch (error) { + // The hub keeps the host's stderr as the app's log. + stderrLine("Disposing the app sandbox failed", error); + } process.exit(code); }; process.once("SIGTERM", () => void close(0)); -void (async () => { + +function followLine(line: string) { + const json = parseJson(line); + const source = json === undefined ? undefined : HostSource(json); + if (source === undefined || source instanceof type.errors) { + process.stderr.write( + `Skipped a source line: ${source === undefined ? "not JSON" : "not a valid source"}\n`, + ); + return; + } + follow(source); +} + +async function followStdin() { const decoder = new TextDecoder(); let buffered = ""; - for await (const chunk of Bun.stdin.stream()) { - buffered += decoder.decode(chunk, { stream: true }); - const lines = buffered.split("\n"); - buffered = lines.pop() ?? ""; - for (const line of lines) - if (line.trim()) follow(HostSource.assert(JSON.parse(line))); + try { + for await (const chunk of Bun.stdin.stream()) { + buffered += decoder.decode(chunk, { stream: true }); + for ( + let end = buffered.indexOf("\n"); + end !== -1; + end = buffered.indexOf("\n") + ) { + const line = buffered.slice(0, end); + buffered = buffered.slice(end + 1); + if (line.trim()) followLine(line); + } + } + } catch (error) { + stderrLine("The host stopped", error); + return close(1); } await close(0); -})(); +} +void followStdin(); + +// Whether the app answers at all: any HTTP response means it is up. +async function answers(url: string) { + try { + await fetch(url, { signal: AbortSignal.timeout(STATE_MS) }); + return true; + } catch (error) { + if ( + error instanceof TypeError || + (error instanceof DOMException && error.name === "TimeoutError") + ) + return false; + throw error; + } +} setInterval(async () => { if (installing) @@ -218,14 +292,8 @@ setInterval(async () => { log: app.log, }); } - const answers = await fetch(app.url, { - signal: AbortSignal.timeout(STATE_MS), - }).then( - () => true, - () => false, - ); report( - answers + (await answers(app.url)) ? { state: "running", url: app.url } : { state: "starting", phase: "starting", log: app.log }, ); diff --git a/apps/hub/src/live-app/proxy.ts b/apps/hub/src/live-app/proxy.ts index 56097451..9ca27c79 100644 --- a/apps/hub/src/live-app/proxy.ts +++ b/apps/hub/src/live-app/proxy.ts @@ -1,5 +1,9 @@ +import { getLogger } from "@intx/log"; import type { MiddlewareHandler } from "hono"; import { liveAppWording } from "../config/live-app.ts"; +import { errorCode, requestErrorMetadata } from "../request-error.ts"; + +const log = getLogger(["corbits-builder", "hub", "live-app"]); export type LiveAppScope = { tenantId: string; projectId: string }; @@ -74,7 +78,9 @@ export function createLiveAppProxy(options: { redirect: "manual", decompress: false, }); - } catch { + } catch (error) { + // fetch rejects with a TypeError when the app is not listening. + if (!(error instanceof TypeError)) throw error; return new Response(liveAppWording.unreachable, { status: 502 }); } const out = new Headers(response.headers); @@ -98,6 +104,13 @@ export function createLiveAppProxy(options: { entry.server = Bun.serve({ hostname: options.hostname, port: 0, + error(error) { + log.warn("Live app proxy request failed", { + ...errorCode(error), + ...requestErrorMetadata(error), + }); + return new Response(liveAppWording.unreachable, { status: 502 }); + }, fetch: async (request, server) => { if (!(await allows(request, entry))) return new Response(liveAppWording.signIn, { status: 401 }); diff --git a/apps/hub/src/live-app/runner.ts b/apps/hub/src/live-app/runner.ts index 8c0e9083..78d89f8e 100644 --- a/apps/hub/src/live-app/runner.ts +++ b/apps/hub/src/live-app/runner.ts @@ -2,7 +2,7 @@ import { mkdirSync } from "node:fs"; import { dirname, resolve } from "node:path"; import { fileURLToPath } from "node:url"; import { getLogger } from "@intx/log"; -import { requestErrorMetadata } from "../request-error.ts"; +import { errorCode, requestErrorMetadata } from "../request-error.ts"; import { liveAppWording } from "../config/live-app.ts"; import type { HostReport, HostSource } from "./host.ts"; import type { createLiveAppProxy } from "./proxy.ts"; @@ -55,22 +55,53 @@ export function createLiveAppRunner(options: { }, SWEEP_MS); sweep.unref(); - async function follow(run: Run) { + async function follow(projectId: string, run: Run) { const decoder = new TextDecoder(); let pending = ""; - for await (const chunk of run.child.stdout) { - pending += decoder.decode(chunk, { stream: true }); - const lines = pending.split("\n"); - pending = lines.pop() ?? ""; - for (const line of lines) run.state = JSON.parse(line) as HostReport; + try { + for await (const chunk of run.child.stdout) { + pending += decoder.decode(chunk, { stream: true }); + for ( + let end = pending.indexOf("\n"); + end !== -1; + end = pending.indexOf("\n") + ) { + run.state = JSON.parse(pending.slice(0, end)) as HostReport; + pending = pending.slice(end + 1); + } + } + } catch (error) { + log.warn("Reading the live app host's state failed for {projectId}", { + projectId, + ...errorCode(error), + ...requestErrorMetadata(error), + }); } } - async function collect(run: Run) { + async function collect(projectId: string, run: Run) { const decoder = new TextDecoder(); - for await (const chunk of run.child.stderr) - for (const line of decoder.decode(chunk, { stream: true }).split("\n")) - if (line.trim() && run.stderr.push(line) > 40) run.stderr.shift(); + try { + for await (const chunk of run.child.stderr) + for (const line of decoder.decode(chunk, { stream: true }).split("\n")) + if (line.trim() && run.stderr.push(line) > 40) run.stderr.shift(); + } catch (error) { + log.warn("Reading the live app host's stderr failed for {projectId}", { + projectId, + ...errorCode(error), + ...requestErrorMetadata(error), + }); + } + } + + async function failOnExit(projectId: string, run: Run) { + const code = await run.child.exited; + if (runs.get(projectId) === run && run.state.state !== "failed") + run.state = { + state: "failed", + error: liveAppWording.hostExited(code), + log: run.stderr, + }; } function spawn(projectId: string, key: string) { @@ -103,26 +134,9 @@ export function createLiveAppRunner(options: { lastSeen: Date.now(), }; runs.set(projectId, run); - void follow(run).catch((error: unknown) => - log.warn("Reading the live app host's state failed for {projectId}", { - projectId, - ...requestErrorMetadata(error), - }), - ); - void collect(run).catch((error: unknown) => - log.warn("Reading the live app host's stderr failed for {projectId}", { - projectId, - ...requestErrorMetadata(error), - }), - ); - void child.exited.then((code) => { - if (runs.get(projectId) === run && run.state.state !== "failed") - run.state = { - state: "failed", - error: liveAppWording.hostExited(code), - log: run.stderr, - }; - }); + void follow(projectId, run); + void collect(projectId, run); + void failOnExit(projectId, run); return run; } diff --git a/apps/hub/src/mail/caller-mail.ts b/apps/hub/src/mail/caller-mail.ts index 788baa3c..ed29b5f1 100644 --- a/apps/hub/src/mail/caller-mail.ts +++ b/apps/hub/src/mail/caller-mail.ts @@ -1,6 +1,7 @@ import type { Context } from "hono"; import type { TenantEnv } from "@intx/hub-api"; import { ProjectError } from "../projects/errors.ts"; +import { readJson } from "../lib/json.ts"; // Mail the hub sends on the person's behalf goes through the native run mail route with the // person's own credentials, exactly as the UI sends a message. @@ -29,7 +30,7 @@ export async function mailRunAsCaller( }), ); if (response.ok) return; - const payload: unknown = await response.json().catch(() => null); + const payload: unknown = await readJson(response); const upstream = payload !== null && typeof payload === "object" && "error" in payload ? payload.error diff --git a/apps/hub/src/mail/committed-agent-state.ts b/apps/hub/src/mail/committed-agent-state.ts index c328d6f6..82260544 100644 --- a/apps/hub/src/mail/committed-agent-state.ts +++ b/apps/hub/src/mail/committed-agent-state.ts @@ -3,11 +3,14 @@ import { mkdtemp, rm } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { pipeline } from "node:stream"; +import { getLogger } from "@intx/log"; +import { errorCode, requestErrorMetadata } from "../request-error.ts"; // Same ref the workflow-run pack receive updates. That receive stores objects // and advances the ref; it does not check the worktree out. A reader that // walks the working tree sees an empty mirror after a successful pack. const MIRROR_REF = "refs/heads/main"; +const log = getLogger(["corbits-builder", "hub", "agent-state"]); function capture(args: string[]): Promise<{ code: number; stdout: string }> { return new Promise((resolve, reject) => { @@ -38,7 +41,13 @@ function extractArchive(repoDir: string, dest: string): Promise { extract.stderr.resume(); // A tar that exits early breaks the pipe; both exit codes report that, while an unhandled // EPIPE would take the hub down and a bare pipe() would leave git blocked on a full pipe. - pipeline(archive.stdout, extract.stdin, () => {}); + pipeline(archive.stdout, extract.stdin, function logPipeFailure(error) { + if (error) + log.warn("Copying the committed agent state failed", { + ...requestErrorMetadata(error), + ...errorCode(error), + }); + }); let settled = false; const fail = (error: Error) => { if (settled) return; diff --git a/apps/hub/src/mail/conversation-routes.ts b/apps/hub/src/mail/conversation-routes.ts index 8e57f99e..1991aab3 100644 --- a/apps/hub/src/mail/conversation-routes.ts +++ b/apps/hub/src/mail/conversation-routes.ts @@ -20,6 +20,7 @@ import { reportAuthorization } from "../projects/authorization.ts"; import { ProjectError, projectResource } from "../projects/errors.ts"; import { getLogger } from "@intx/log"; import { requestErrorMetadata } from "../request-error.ts"; +import { parseJson } from "../lib/json.ts"; import { createCoordinatorQuestionProseReader, type CoordinatorRepoDirSource, @@ -153,12 +154,8 @@ export function createConversationRoutes( status: response.status, headers: response.headers, }); - let payload: unknown; - try { - payload = JSON.parse(raw); - } catch { - return passThrough(raw); - } + const payload = parseJson(raw); + if (payload === undefined) return passThrough(raw); const { address } = selected; const events = (payload as { events?: unknown })["events"]; if (Array.isArray(events)) { diff --git a/apps/hub/src/mail/conversation-threads.ts b/apps/hub/src/mail/conversation-threads.ts index 7029d776..c3850d94 100644 --- a/apps/hub/src/mail/conversation-threads.ts +++ b/apps/hub/src/mail/conversation-threads.ts @@ -47,8 +47,11 @@ const addrSpec = (line: string | undefined): string | undefined => { if (line === undefined || line.trim() === "") return undefined; try { return extractAddrSpec(line).toLowerCase(); - } catch { - return undefined; + } catch (error) { + // A header that names no single well-formed address names no participant. + if (error instanceof Error && error.message.startsWith("extractAddrSpec:")) + return undefined; + throw error; } }; diff --git a/apps/hub/src/mail/conversation-turns.ts b/apps/hub/src/mail/conversation-turns.ts index 3a4ff9dc..30c1e911 100644 --- a/apps/hub/src/mail/conversation-turns.ts +++ b/apps/hub/src/mail/conversation-turns.ts @@ -21,7 +21,7 @@ import { import { withCommittedAgentState } from "./committed-agent-state.ts"; import type { CoordinatorRepoDirSource } from "./coordinator-question.ts"; import { getLogger } from "@intx/log"; -import { requestErrorMetadata } from "../request-error.ts"; +import { errorCode, requestErrorMetadata } from "../request-error.ts"; const log = getLogger(["corbits-builder", "hub", "conversation-turns"]); @@ -119,17 +119,21 @@ export async function readDurableConversationTurns(input: { for (const turn of snapshot.turns) collected.push(turn); } catch (error) { // One damaged agent mirror must not drop the other agents' turns. - log.error( - "Agent conversation mirror unreadable", - requestErrorMetadata(error), - ); + log.error("Agent conversation mirror unreadable", { + ...errorCode(error), + ...requestErrorMetadata(error), + }); } } return collected; }, ); return mapMirrorConversationTurns(turns ?? []); - } catch { + } catch (error) { + log.warn("Durable conversation unreadable", { + ...errorCode(error), + ...requestErrorMetadata(error), + }); return []; } } diff --git a/apps/hub/src/mail/coordinator-question.ts b/apps/hub/src/mail/coordinator-question.ts index 96d42cf3..85797dcf 100644 --- a/apps/hub/src/mail/coordinator-question.ts +++ b/apps/hub/src/mail/coordinator-question.ts @@ -1,11 +1,15 @@ import { readdir } from "node:fs/promises"; import { join } from "node:path"; +import { getLogger } from "@intx/log"; import { reconstructDurableConversation } from "@intx/workflow-host"; import { deriveWorkflowRunRepoId } from "@intx/workflow-deploy"; +import { errorCode, requestErrorMetadata } from "../request-error.ts"; import { withCommittedAgentState } from "./committed-agent-state.ts"; +const log = getLogger(["corbits-builder", "hub", "coordinator-question"]); + // The input park (`SignalAwaited` with `parkKind: "input"`) carries no wording: // the coordinator's turn reply is what reaches the person as mail, and its // durable copy is the warm agent's conversation mirror under @@ -28,15 +32,10 @@ export function createCoordinatorQuestionProseReader(options: { address: string; park: Record; }): Promise { - let repoDir: string; - try { - repoDir = options.repoStore.getRepoDir({ - kind: "workflow-run", - id: deriveWorkflowRunRepoId(input.address), - }); - } catch { - return undefined; - } + const repoDir = options.repoStore.getRepoDir({ + kind: "workflow-run", + id: deriveWorkflowRunRepoId(input.address), + }); // The asking turn completed before the runtime committed the re-arm park, // so the park's own commit time bounds which turns may have asked it. const parkedAt = input.park["at"]; @@ -49,19 +48,10 @@ export function createCoordinatorQuestionProseReader(options: { | undefined; try { best = await withCommittedAgentState(repoDir, async (agentStateRoot) => { - let keys: string[]; - try { - const entries = await readdir(agentStateRoot, { - withFileTypes: true, - }); - keys = entries - .filter((entry) => entry.isDirectory()) - .map((entry) => entry.name); - } catch { - // No durable mirror yet (the run never completed a turn): there is no - // committed prose to expose, and the park renders verbatim. - return undefined; - } + const entries = await readdir(agentStateRoot, { withFileTypes: true }); + const keys = entries + .filter((entry) => entry.isDirectory()) + .map((entry) => entry.name); let found: | { prose: string; agentKey: string; timestamp: number } | undefined; @@ -74,11 +64,16 @@ export function createCoordinatorQuestionProseReader(options: { ); if (snapshot === null) continue; turns = snapshot.turns; - } catch { + } catch (error) { // A corrupt mirror leaves the rest of the conversation projection // intact; the card falls back to the verbatim park rather than // guessing at wording. The reader still throws on the restore path, // where a lost conversation is a correctness failure. + log.warn("Agent conversation mirror unreadable", { + agentKey, + ...errorCode(error), + ...requestErrorMetadata(error), + }); continue; } for (const turn of turns) { @@ -96,7 +91,11 @@ export function createCoordinatorQuestionProseReader(options: { } return found; }); - } catch { + } catch (error) { + log.warn("Coordinator question prose unreadable", { + ...errorCode(error), + ...requestErrorMetadata(error), + }); return undefined; } if (best === undefined || best.prose.trim() === "") return undefined; diff --git a/apps/hub/src/mail/hub-mail.ts b/apps/hub/src/mail/hub-mail.ts index 3568f13a..bb1c01ce 100644 --- a/apps/hub/src/mail/hub-mail.ts +++ b/apps/hub/src/mail/hub-mail.ts @@ -2,6 +2,7 @@ import { and, eq } from "drizzle-orm"; import type { DB } from "@intx/db"; import { principal, user } from "@intx/db/schema"; import type { GetSession } from "@intx/hub-api"; +import { readJson } from "../lib/json.ts"; type Session = NonNullable>>; @@ -61,7 +62,7 @@ export function createHubMail( }); const response = await fetch(request); if (response.ok) return; - const payload: unknown = await response.json().catch(() => null); + const payload: unknown = await readJson(response); const upstream = payload !== null && typeof payload === "object" && "error" in payload ? payload.error diff --git a/apps/hub/src/projects/coordinator-resume.ts b/apps/hub/src/projects/coordinator-resume.ts index 68bcc49a..6cc4ad79 100644 --- a/apps/hub/src/projects/coordinator-resume.ts +++ b/apps/hub/src/projects/coordinator-resume.ts @@ -11,6 +11,7 @@ import type { CoordinatorRepoDirSource } from "../mail/coordinator-question.ts"; import { awaitingApproval } from "./deploy-tools.ts"; import { getLogger } from "@intx/log"; import { requestErrorMetadata } from "../request-error.ts"; +import { readJson } from "../lib/json.ts"; const log = getLogger(["corbits-builder", "hub", "coordinator-resume"]); @@ -98,20 +99,17 @@ export function createCoordinatorResumeContext(deps: { const runId = c.req.param("runId"); if (c.req.method !== "POST" || !tenantId || !runId) return next(); // A context that cannot be read must not cost the person their message. - const context = await heldContext(tenantId, runId).catch( - (error: unknown) => { - log.error("Resume context unavailable", { - runId, - ...requestErrorMetadata(error), - }); - return undefined; - }, - ); + let context: Awaited>; + try { + context = await heldContext(tenantId, runId); + } catch (error) { + log.error("Resume context unavailable", { + runId, + ...requestErrorMetadata(error), + }); + } if (context === undefined) return next(); - const body: unknown = await c.req.raw - .clone() - .json() - .catch(() => undefined); + const body = await readJson(c.req.raw.clone()); if (!isRecord(body) || typeof body["content"] !== "string") return next(); const headers = new Headers(c.req.raw.headers); headers.delete("content-length"); diff --git a/apps/hub/src/projects/coordinator-startup.ts b/apps/hub/src/projects/coordinator-startup.ts index 098e6e3c..0958817d 100644 --- a/apps/hub/src/projects/coordinator-startup.ts +++ b/apps/hub/src/projects/coordinator-startup.ts @@ -114,11 +114,14 @@ function createProjectCoordinatorStartup(ports: CoordinatorStartupPorts) { const key = startKey(actor.tenantId, projectId); const pending = inflight.get(key); if (pending) return pending; - const work = attach(ports, deployed, key, actor, projectId).finally( - () => { + async function attachOnce() { + try { + return await attach(ports, deployed, key, actor, projectId); + } finally { inflight.delete(key); - }, - ); + } + } + const work = attachOnce(); inflight.set(key, work); return work; }, @@ -128,10 +131,9 @@ function createProjectCoordinatorStartup(ports: CoordinatorStartupPorts) { deployed.delete(startKey(actor.tenantId, projectId)); return ports.unbind(actor.tenantId, projectId); }, - attached(actor: Actor, projectId: string): Promise { - return ports - .boundRun(actor.tenantId, projectId) - .then((row) => row !== null && !row.ended); + async attached(actor: Actor, projectId: string): Promise { + const row = await ports.boundRun(actor.tenantId, projectId); + return row !== null && !row.ended; }, }; } diff --git a/apps/hub/src/projects/coordinator-upgrade.ts b/apps/hub/src/projects/coordinator-upgrade.ts index 4c4bd24d..d631c195 100644 --- a/apps/hub/src/projects/coordinator-upgrade.ts +++ b/apps/hub/src/projects/coordinator-upgrade.ts @@ -18,6 +18,7 @@ import { spawnedWorkflowRole } from "../workflows/spawn-role.ts"; import { MODEL_CHOICE_REFUSALS } from "../config/models.ts"; import { project, projectCoordinator } from "../db/schema.ts"; import type { Actor } from "../lib/actor.ts"; +import { settle } from "../lib/serial-queue.ts"; import { requestErrorMetadata } from "../request-error.ts"; import { ProjectError } from "./errors.ts"; @@ -215,12 +216,15 @@ export function createCoordinatorUpgrades(deps: { function upgradeOnce(projectId: string, tenantId: string, opener?: Actor) { const pending = inflight.get(projectId); if (pending && (pending.opened || !opener)) return pending.work; - const work: Promise = (pending?.work ?? Promise.resolve()) - .catch(() => undefined) - .then(() => upgrade(projectId, tenantId, opener)) - .finally(() => { + async function upgradeAfterPending(): Promise { + try { + await settle(pending?.work); + await upgrade(projectId, tenantId, opener); + } finally { if (inflight.get(projectId)?.work === work) inflight.delete(projectId); - }); + } + } + const work = upgradeAfterPending(); inflight.set(projectId, { opened: opener !== undefined, work }); return work; } diff --git a/apps/hub/src/projects/next-step.ts b/apps/hub/src/projects/next-step.ts index 26bee82f..99a77237 100644 --- a/apps/hub/src/projects/next-step.ts +++ b/apps/hub/src/projects/next-step.ts @@ -29,7 +29,7 @@ import { import { projectStatusWording as text } from "../config/project-status.ts"; import type { Actor } from "../lib/actor.ts"; import type { LiveApps } from "../live-app/runner.ts"; -import { requestErrorMetadata } from "../request-error.ts"; +import { errorCode, requestErrorMetadata } from "../request-error.ts"; import { actorOf, authorizeProject } from "./authorization.ts"; import { ProjectError } from "./errors.ts"; import { workflowRoleSql } from "../workflows/spawn-role.ts"; @@ -39,15 +39,6 @@ const log = getLogger(["corbits-builder", "hub", "next-step"]); // The web client polls run states every few seconds, so a run whose state cannot be read is logged // once per this interval rather than on every poll. const UNREADABLE_LOG_MS = 5 * 60 * 1000; -// A thrown error's code, such as ENOENT or isomorphic-git's NotFoundError, names what failed -// without the message's paths; anything else it carries is left out. -const errorCode = (error: unknown) => - error instanceof Error && - "code" in error && - typeof error.code === "string" && - /^[A-Za-z][A-Za-z0-9_]{0,63}$/.test(error.code) - ? { errorCode: error.code } - : {}; type ProjectRun = { runId: string; @@ -126,22 +117,36 @@ export function createProjectNextStep( const kind = event.body["parkKind"]; return kind === "input" || kind === "signal-relay" ? "idle" : "approval"; } catch (error) { - const now = Date.now(); - if (now - (unreadableLogged.get(run.runId) ?? 0) >= UNREADABLE_LOG_MS) { - unreadableLogged.set(run.runId, now); - log.error("Reading a run's latest event failed", { - runId: run.runId, - ...errorCode(error), - ...requestErrorMetadata(error), - }); - } + logUnreadable(run, error); return "unknown"; } } + function logUnreadable(run: ProjectRun, error: unknown) { + const now = Date.now(); + if (now - (unreadableLogged.get(run.runId) ?? 0) < UNREADABLE_LOG_MS) + return; + unreadableLogged.set(run.runId, now); + log.error("Reading a run's latest event failed", { + runId: run.runId, + ...errorCode(error), + ...requestErrorMetadata(error), + }); + } + + // A failed run's own failure message stands in for an event that cannot be read. + async function failureEvent(run: ProjectRun) { + if (run.address === null) return null; + try { + return await latestEvent(run); + } catch (error) { + logUnreadable(run, error); + return null; + } + } + async function failureReason(run: ProjectRun) { - const event = - run.address === null ? null : await latestEvent(run).catch(() => null); + const event = await failureEvent(run); const error = event?.body["error"]; const message = typeof error === "object" && error !== null && "message" in error diff --git a/apps/hub/src/projects/routes.ts b/apps/hub/src/projects/routes.ts index 5aeb0753..4eea6f6f 100644 --- a/apps/hub/src/projects/routes.ts +++ b/apps/hub/src/projects/routes.ts @@ -14,6 +14,7 @@ import { type TenantEnv, } from "@intx/hub-api"; import type { DB } from "@intx/db"; +import { getLogger } from "@intx/log"; import type { RepoStore, WorkflowRunReader } from "@intx/hub-sessions"; import { TenantResponse } from "@intx/types"; import { @@ -43,6 +44,10 @@ import type { LiveApps } from "../live-app/runner.ts"; import { importedProjectMessage } from "../config/import.ts"; import { MODEL_CHOICE_REFUSALS } from "../config/models.ts"; import { mailRunAsCaller } from "../mail/caller-mail.ts"; +import { readJson } from "../lib/json.ts"; +import { errorCode, requestErrorMetadata } from "../request-error.ts"; + +const log = getLogger(["corbits-builder", "hub", "projects"]); type CoordinatorStart = { ensure: (actor: Actor, projectId: string) => Promise; @@ -102,7 +107,7 @@ export function nativeTenantCreation( body: JSON.stringify(body), }), ); - const payload: unknown = await response.json().catch(() => null); + const payload: unknown = await readJson(response); if (response.status === 201) return { id: TenantResponse.assert(payload).id }; const upstream = @@ -131,6 +136,23 @@ function caller(c: Context): ProjectCaller { }; } +// The create's own failure is what the caller sees; a project left unarchived only keeps its card. +async function archiveFailedCreate( + archive: (actor: Actor, projectId: string) => Promise, + actor: Actor, + projectId: string, +) { + try { + await archive(actor, projectId); + } catch (error) { + log.warn("Archiving failed project create {projectId} failed", { + projectId, + ...errorCode(error), + ...requestErrorMetadata(error), + }); + } +} + // A project whose coordinator never started is archived, so the failed create // leaves no card behind; a retry replays the receipt and starts it again. One // refused only for want of a model profile stays, and starts once one resolves. @@ -145,7 +167,7 @@ async function startNewProject( await start.ensure(actor, projectId); } catch (error) { if (!(error instanceof ProjectError && error.code in MODEL_CHOICE_REFUSALS)) - await archive(actor, projectId).catch(() => undefined); + await archiveFailedCreate(archive, actor, projectId); throw error; } } @@ -244,12 +266,8 @@ export function createProjectRoutes( ) ) return c.json({ code: "json_content_type_required" }, 415); - let input: unknown; - try { - input = await c.req.json(); - } catch { - throw new ProjectError(400, "invalid_json"); - } + const input = await readJson(c.req); + if (input === undefined) throw new ProjectError(400, "invalid_json"); const read = readProjectImport(input); if (read instanceof type.errors) throw new ProjectError(400, "invalid_project_import", { @@ -289,7 +307,7 @@ export function createProjectRoutes( for (const material of materials) await artifacts.fileMaterial(created.actor, projectId, material); } catch (error) { - await archive(created.actor, projectId).catch(() => undefined); + await archiveFailedCreate(archive, created.actor, projectId); throw error; } // The coordinator starts after filing, so its first look at the @@ -313,7 +331,7 @@ export function createProjectRoutes( ), ); } catch (error) { - await archive(created.actor, projectId).catch(() => undefined); + await archiveFailedCreate(archive, created.actor, projectId); throw error; } return c.json(ProjectImportReceipt.assert({ projectId, omitted })); @@ -488,12 +506,8 @@ export function createProjectRoutes( ) ) return c.json({ code: "json_content_type_required" }, 415); - let input: unknown; - try { - input = await c.req.json(); - } catch { - throw new ProjectError(400, "invalid_json"); - } + const input = await readJson(c.req); + if (input === undefined) throw new ProjectError(400, "invalid_json"); const result = await service.mutate( caller(c), nativeTenantCreation(c, nativeFetch), @@ -523,8 +537,10 @@ export function createProjectRoutes( result.opening, ); } catch (error) { - await archive(result.actor, result.projectId).catch( - () => undefined, + await archiveFailedCreate( + archive, + result.actor, + result.projectId, ); throw error; } diff --git a/apps/hub/src/projects/specialist-ended.ts b/apps/hub/src/projects/specialist-ended.ts index bface72e..12ee1fbb 100644 --- a/apps/hub/src/projects/specialist-ended.ts +++ b/apps/hub/src/projects/specialist-ended.ts @@ -10,7 +10,7 @@ import { runFailure } from "../artifacts/service.ts"; import { specialistEndedMessage } from "../config/specialist-ended.ts"; import { endedRun, project, projectCoordinator } from "../db/schema.ts"; import { HubMailError, type HubMail } from "../mail/hub-mail.ts"; -import { requestErrorMetadata } from "../request-error.ts"; +import { errorCode, requestErrorMetadata } from "../request-error.ts"; import { spawnedWorkflowRole } from "../workflows/spawn-role.ts"; const log = getLogger(["corbits-builder", "hub", "specialist-ended"]); @@ -18,12 +18,19 @@ const log = getLogger(["corbits-builder", "hub", "specialist-ended"]); // The correlation id of the latest assignment the run received after its last outgoing mail; none // when it answered every assignment it was given. async function unansweredAssignment( - directory: string, + mailDirectory: string, + runId: string, ): Promise { let entries: Awaited>; try { - entries = await listMail(directory); - } catch { + entries = await listMail(join(mailDirectory, runId)); + } catch (error) { + // An unreadable mail store reads as answered, so the ended run is still recorded. + log.warn("Reading ended specialist {runId}'s mail failed", { + runId, + ...errorCode(error), + ...requestErrorMetadata(error), + }); return undefined; } let assigned: { at: number; correlationId: string } | undefined; @@ -115,7 +122,7 @@ export function createSpecialistEndNotices(deps: { .returning({ runId: endedRun.runId }); const role = spawnedWorkflowRole(ended.bundle); const correlationId = role - ? await unansweredAssignment(join(deps.mailDirectory, ended.runId)) + ? await unansweredAssignment(deps.mailDirectory, ended.runId) : undefined; const [coordinator] = correlationId === undefined diff --git a/apps/hub/src/request-error.ts b/apps/hub/src/request-error.ts index 5076d33f..bddce222 100644 --- a/apps/hub/src/request-error.ts +++ b/apps/hub/src/request-error.ts @@ -1,4 +1,5 @@ import { fileURLToPath } from "node:url"; +import { hasCode } from "@intx/types"; const sourceFiles = [ "apps/hub/src/app.ts", @@ -8,6 +9,13 @@ const sourceFiles = [ ] as const; const repositoryRoot = fileURLToPath(new URL("../../../", import.meta.url)); +// A thrown error's code, such as ENOENT, EAUTH, a Postgres SQLSTATE or isomorphic-git's +// NotFoundError, names what failed without the message's paths; anything else it carries is left out. +export const errorCode = (error: unknown) => + hasCode(error) && /^[A-Za-z0-9][A-Za-z0-9_]{0,63}$/.test(error.code) + ? { errorCode: error.code } + : {}; + // Never emit error-supplied filenames, messages, causes, or custom names. export function requestErrorMetadata(error: unknown) { const errorType = diff --git a/apps/hub/src/run-stream.ts b/apps/hub/src/run-stream.ts index 2baae1e8..a14e0bd8 100644 --- a/apps/hub/src/run-stream.ts +++ b/apps/hub/src/run-stream.ts @@ -10,6 +10,7 @@ import { idResource, type RequireGrant, type TenantEnv } from "@intx/hub-api"; import { findRoutableById, type SidecarRouter } from "@intx/hub-sessions"; import { getLogger } from "@intx/log"; import { parseInferenceEvent } from "@intx/types/runtime"; +import { errorCode, requestErrorMetadata } from "./request-error.ts"; const HEARTBEAT_MS = 5_000; const MailDelivered = type({ @@ -44,11 +45,20 @@ export function createWorkflowRunStreamRoutes(deps: { c.header("X-Accel-Buffering", "no"); return streamSSE(c, async (stream) => { - const write = (event: string, data: unknown) => { - stream.writeSSE({ event, data: JSON.stringify(data) }).catch(() => { - log.error(`Workflow run stream write failed for ${runId}`); - }); - }; + async function send(message: { event: string; data: string }) { + try { + await stream.writeSSE(message); + } catch (error) { + log.error("Workflow run stream {event} write failed for {runId}", { + event: message.event, + runId, + ...errorCode(error), + ...requestErrorMetadata(error), + }); + } + } + const write = (event: string, data: unknown) => + void send({ event, data: JSON.stringify(data) }); const unsubscribe = deps.sidecarRouter.subscribeAgent( record.address, (raw) => { @@ -72,11 +82,10 @@ export function createWorkflowRunStreamRoutes(deps: { write("status", { routable: false }); }, ); - const heartbeat = setInterval(() => { - stream.writeSSE({ event: "ping", data: "" }).catch(() => { - log.error(`Workflow run stream heartbeat failed for ${runId}`); - }); - }, HEARTBEAT_MS); + const heartbeat = setInterval( + () => void send({ event: "ping", data: "" }), + HEARTBEAT_MS, + ); try { await new Promise((resolve) => { diff --git a/apps/hub/src/sidecar/local-provisioner.ts b/apps/hub/src/sidecar/local-provisioner.ts index 01fe5e88..981cd82f 100644 --- a/apps/hub/src/sidecar/local-provisioner.ts +++ b/apps/hub/src/sidecar/local-provisioner.ts @@ -90,6 +90,11 @@ type ManagedProcess = { exited: boolean; }; +async function exits(handle: LocalSidecarProcess): Promise { + await handle.exited; + return true; +} + async function exitedWithin( managed: ManagedProcess, timeoutMs: number, @@ -100,10 +105,7 @@ async function exitedWithin( timer = setTimeout(resolve, timeoutMs, false); }); try { - return await Promise.race([ - managed.handle.exited.then(() => true), - timedOut, - ]); + return await Promise.race([exits(managed.handle), timedOut]); } finally { clearTimeout(timer); } diff --git a/apps/hub/src/sidecar/manifest-store.ts b/apps/hub/src/sidecar/manifest-store.ts index a8228c01..20b52b6e 100644 --- a/apps/hub/src/sidecar/manifest-store.ts +++ b/apps/hub/src/sidecar/manifest-store.ts @@ -3,7 +3,7 @@ import fs from "node:fs/promises"; import path from "node:path"; import { type } from "arktype"; import { errorText } from "./error-text.ts"; -import type { CredentialCipher } from "@intx/types"; +import { hasCode, type CredentialCipher } from "@intx/types"; /** The hub's one local sidecar: the identity it authenticates with and where it keeps its state. */ export type LocalSidecarManifest = { @@ -59,15 +59,11 @@ function manifestTokenAad(sidecarId: string): string { return JSON.stringify(["sidecar-manifest-token", sidecarId]); } -function isMissing(error: unknown): boolean { - return error instanceof Error && "code" in error && error.code === "ENOENT"; -} - async function readOptional(file: string): Promise { try { return await fs.readFile(file, "utf8"); } catch (error) { - if (isMissing(error)) return null; + if (hasCode(error) && error.code === "ENOENT") return null; throw error; } } diff --git a/apps/hub/src/sidecar/spawn.ts b/apps/hub/src/sidecar/spawn.ts index 804f1a2d..cbb6f0b9 100644 --- a/apps/hub/src/sidecar/spawn.ts +++ b/apps/hub/src/sidecar/spawn.ts @@ -1,3 +1,4 @@ +import { hasCode } from "@intx/types"; import type { LocalSidecarExit, SpawnLocalSidecar, @@ -28,22 +29,20 @@ const SIDECAR_ENTRY = "apps/sidecar/src/index.ts"; const EXIT_POLL_MS = 100; function commandOf(pid: number): string | undefined { - try { - const ps = Bun.spawnSync(["ps", "-o", "command=", "-p", String(pid)], { - stderr: "ignore", - }); - return ps.exitCode === 0 ? ps.stdout.toString() : undefined; - } catch { - return undefined; - } + const ps = Bun.spawnSync(["ps", "-o", "command=", "-p", String(pid)], { + stderr: "ignore", + }); + return ps.exitCode === 0 ? ps.stdout.toString() : undefined; } -function isAlive(pid: number): boolean { +// Signals a process that is still running; false when it no longer exists. +function sendSignal(pid: number, name: NodeJS.Signals | 0): boolean { try { - process.kill(pid, 0); + process.kill(pid, name); return true; - } catch { - return false; + } catch (error) { + if (hasCode(error) && error.code === "ESRCH") return false; + throw error; } } @@ -57,14 +56,10 @@ export async function stopStaleLocalSidecar( timeoutMs: number, ): Promise { if (!commandOf(pid)?.includes(SIDECAR_ENTRY)) return false; - for (const signal of ["SIGTERM", "SIGKILL"] as const) { - try { - process.kill(pid, signal); - } catch { - return true; - } + for (const name of ["SIGTERM", "SIGKILL"] as const) { + if (!sendSignal(pid, name)) return true; for (const deadline = Date.now() + timeoutMs; Date.now() < deadline;) { - if (!isAlive(pid)) return true; + if (!sendSignal(pid, 0)) return true; await Bun.sleep(EXIT_POLL_MS); } }