diff --git a/apps/web/app/lib/useHomeStream.ts b/apps/web/app/lib/useHomeStream.ts index 2e3d162b..14b56feb 100644 --- a/apps/web/app/lib/useHomeStream.ts +++ b/apps/web/app/lib/useHomeStream.ts @@ -75,10 +75,12 @@ export function useHomeStream(): { }); const onAny = () => refresh(); es.addEventListener("session_state", onAny); + es.addEventListener("session_title", onAny); es.addEventListener("task_event", onAny); // ui_message_part is per-token noise — don't refetch on each. return () => { es.removeEventListener("session_state", onAny); + es.removeEventListener("session_title", onAny); es.removeEventListener("task_event", onAny); es.close(); }; diff --git a/apps/web/app/lib/useLiveSession.ts b/apps/web/app/lib/useLiveSession.ts index 1ccc9b2d..ecd94bd6 100644 --- a/apps/web/app/lib/useLiveSession.ts +++ b/apps/web/app/lib/useLiveSession.ts @@ -7,7 +7,8 @@ import { API_BASE } from "./api"; * Subscribe to a per-session SSE channel. Feeds `ui_message_part` chunks * into `readUIMessageStream` for live assembly; accepts pre-assembled * UIMessages via `messages_appended`. Both upsert by id so chunks and - * the end-of-turn canonical broadcast converge. + * the end-of-turn canonical broadcast converge. Metadata updates such as + * `session_title` are delivered as callbacks. * * `whenConnected()` resolves on the EventSource's `open` event for the * current sessionId — callers await it before posting /api/chat so the @@ -36,6 +37,8 @@ export function useLiveSession( /** A queued user message was cancelled (via DELETE /queued/:id). * Other tabs drop their chip. */ onInboxCancelled?: (item: { messageId: string }) => void; + /** Session title was generated after the post-turn metadata write. */ + onSessionTitle?: (title: string) => void; } ): { state: "running" | "idle"; whenConnected: () => Promise } { const [state, setState] = useState<"running" | "idle">("idle"); @@ -46,11 +49,13 @@ export function useLiveSession( const onDataPartRef = useRef(opts?.onDataPart); const onInboxQueuedRef = useRef(opts?.onInboxQueued); const onInboxCancelledRef = useRef(opts?.onInboxCancelled); + const onSessionTitleRef = useRef(opts?.onSessionTitle); setMessagesRef.current = setMessages; onTaskEventRef.current = opts?.onTaskEvent; onDataPartRef.current = opts?.onDataPart; onInboxQueuedRef.current = opts?.onInboxQueued; onInboxCancelledRef.current = opts?.onInboxCancelled; + onSessionTitleRef.current = opts?.onSessionTitle; // Promise that resolves on the current EventSource's `open`. Replaced // on every sessionId change so callers always await the live one. const connectedRef = useRef<{ promise: Promise; resolve: () => void }>( @@ -223,6 +228,16 @@ export function useLiveSession( /* ignore */ } }, + session_title: (e) => { + try { + const env = JSON.parse(e.data) as { title?: unknown }; + if (typeof env.title === "string" && env.title.trim()) { + onSessionTitleRef.current?.(env.title); + } + } catch { + /* ignore */ + } + }, task_event: (e) => { try { const env = JSON.parse(e.data) as { diff --git a/apps/web/app/routes/index.tsx b/apps/web/app/routes/index.tsx index 91030c9b..cceef60a 100644 --- a/apps/web/app/routes/index.tsx +++ b/apps/web/app/routes/index.tsx @@ -307,6 +307,9 @@ function ChatPage() { onInboxCancelled: ({ messageId }) => { setQueuedMessages((q) => q.filter((m) => m.id !== messageId)); }, + onSessionTitle: (title) => { + setActiveSessionTitle(title); + }, } ); const isLiveRunning = liveSession.state === "running"; @@ -330,7 +333,7 @@ function ChatPage() { fetch(`${API_BASE}/api/sessions/${sid}`) .then((r) => (r.ok ? r.json() : null)) .then((data: { title?: string | null } | null) => { - if (data) setActiveSessionTitle(data.title ?? null); + if (data?.title) setActiveSessionTitle(data.title); }) .catch(() => {}); }, [isLiveRunning, activeSessionId]); diff --git a/packages/agent-core/src/agent.ts b/packages/agent-core/src/agent.ts index 43aec8a4..35501c5b 100644 --- a/packages/agent-core/src/agent.ts +++ b/packages/agent-core/src/agent.ts @@ -54,6 +54,10 @@ import { ensureStepBoundaries, finalizeOrphanToolParts, } from "./messages.js"; +import { + extractErrorText, + extractStatusCode, +} from "./error-classifier.js"; import type { AgentConfig, MessageMetadata, @@ -64,6 +68,20 @@ import type { const log = createLogger("agent-core.agent"); const DEFAULT_AUTONOMOUS_TIMEOUT_MS = 5 * 60 * 1000; +const UPSTREAM_ERROR_MAX_CHARS = 4096; + +function buildUpstreamErrorPart(err: unknown, provider?: string) { + const statusCode = extractStatusCode(err); + const raw = extractErrorText(err); + const message = + raw.length > UPSTREAM_ERROR_MAX_CHARS + ? raw.slice(0, UPSTREAM_ERROR_MAX_CHARS) + : raw; + return { + type: "data-upstream-error" as const, + data: { provider, statusCode, message }, + }; +} /** * Sum per-step provider-reported cost (OpenRouter usage accounting puts @@ -259,9 +277,8 @@ function lastIsUnansweredAutonomousWake( * the stream and persistence. */ /** Structural subset of the server's SessionBroadcaster used by - * `runAutonomous` to push UIMessage stream chunks + appended messages - * to SSE subscribers. Kept in agent-core so the package stays free - * of a runtime dep on @openacme/server. */ + * agent-core to push live session updates to SSE subscribers. Kept in + * agent-core so the package stays free of a runtime dep on @openacme/server. */ export interface AutonomousBroadcaster { broadcast( sessionId: string, @@ -276,6 +293,7 @@ export interface AutonomousBroadcaster { metadata?: unknown; }>; } + | { kind: "session_title"; title: string } ): void; } @@ -365,6 +383,26 @@ export class Agent { } } + private surfaceAutonomousError(sessionId: string, err: unknown): void { + const msg = { + id: randomUUID(), + role: "assistant" as const, + parts: [buildUpstreamErrorPart(err, this.config.model.provider)], + }; + try { + this.messageStore.append(sessionId, msg); + this.broadcaster?.broadcast(sessionId, { + kind: "messages_appended", + messages: [msg], + }); + } catch (e) { + log.warn( + { err: e, sessionId, agentId: this.config.id }, + "runAutonomous: failed to surface upstream error" + ); + } + } + /** `history` MUST end in the new user message. Caller drives the returned stream. */ async runStream(opts: { sessionId: string; @@ -739,6 +777,7 @@ export class Agent { let timedOut = false; let usage: TokenUsage | undefined; let assistantMessage: UIMessage | null = null; + let capturedError: unknown = null; const recall = inProgress ? await this.applyMemoryRecall({ @@ -835,6 +874,9 @@ export class Agent { history, signal: timeoutAbort.signal, prepareStep, + onError: ({ error }) => { + capturedError = error; + }, usage: { kind: "autonomous", taskId: usageTask?.id }, }); @@ -972,6 +1014,7 @@ export class Agent { if (externalAbort) { externalAbort.removeEventListener("abort", onExternalAbort); } + this.surfaceAutonomousError(sessionId, capturedError ?? e); throw e; } } finally { @@ -987,6 +1030,13 @@ export class Agent { ); } if (!assistantMessage) { + this.surfaceAutonomousError( + sessionId, + capturedError ?? + new Error( + `Autonomous turn in session ${sessionId} produced no assistant message` + ) + ); throw new Error( `Autonomous turn in session ${sessionId} produced no assistant message` ); @@ -994,7 +1044,12 @@ export class Agent { // User message was pre-persisted + pre-broadcast above so any // `ping_user` events fired during the turn aren't auto-resolved. - const assistantParts = assistantMessage.parts as UIMessage["parts"]; + const assistantParts = capturedError + ? [ + ...(assistantMessage.parts as UIMessage["parts"]), + buildUpstreamErrorPart(capturedError, this.config.model.provider), + ] + : (assistantMessage.parts as UIMessage["parts"]); if (assistantParts.length > 0) { const sanitized = ensureStepBoundaries( finalizeOrphanToolParts(assistantParts) @@ -1020,11 +1075,19 @@ export class Agent { }, ], }); - const stored = this.messageStore.getHistory(sessionId); - this.fireExtractor({ - sessionId, - sessionMessages: stored as unknown as UIMessage[], - }); + if (!capturedError) { + const stored = this.messageStore.getHistory(sessionId); + this.fireExtractor({ + sessionId, + sessionMessages: stored as unknown as UIMessage[], + }); + } + } + + if (capturedError) { + throw new Error( + extractErrorText(capturedError) || "Autonomous upstream provider error" + ); } // No cursor advance — inbox-drain-and-delete is the new @@ -1610,7 +1673,7 @@ export class Agent { const fallback = sliceFallbackTitle(opts.sessionMessages); if (!fallback) return; try { - this.sessionStore.updateTitle(opts.sessionId, fallback); + this.writeSessionTitle(opts.sessionId, fallback); } catch (e) { log.warn( { err: e, agentId: this.config.id, sessionId: opts.sessionId }, @@ -1629,7 +1692,7 @@ export class Agent { }) .then((title) => { if (title) { - this.sessionStore.updateTitle(opts.sessionId, title); + this.writeSessionTitle(opts.sessionId, title); return; } writeFallback(); @@ -1646,6 +1709,14 @@ export class Agent { }); } + private writeSessionTitle(sessionId: string, title: string): void { + this.sessionStore.updateTitle(sessionId, title); + this.broadcaster?.broadcast(sessionId, { + kind: "session_title", + title, + }); + } + /** Get conversation history for a session as persisted UIMessages. */ getHistory(sessionId: string): StoredUIMessage[] { return this.messageStore.getHistory(sessionId); diff --git a/packages/agent-core/src/compression.ts b/packages/agent-core/src/compression.ts index 097b0f86..8fbc0026 100644 --- a/packages/agent-core/src/compression.ts +++ b/packages/agent-core/src/compression.ts @@ -2,6 +2,7 @@ import { generateText, type UIMessage, type UIMessagePart } from "ai"; import { createHash, randomUUID } from "node:crypto"; import { getModel } from "@openacme/llm-provider"; import type { ModelConfig } from "@openacme/config"; +import { extractErrorText } from "./error-classifier.js"; import type { CompressionConfig } from "./types.js"; /** @@ -990,8 +991,7 @@ function modelLabel(m: ModelConfig): string { } function errorMessage(e: unknown): string { - if (e instanceof Error) return e.message; - return String(e); + return extractErrorText(e); } export interface CompressOpts { diff --git a/packages/agent-core/src/error-classifier.ts b/packages/agent-core/src/error-classifier.ts index a8356293..7949eb74 100644 --- a/packages/agent-core/src/error-classifier.ts +++ b/packages/agent-core/src/error-classifier.ts @@ -1,3 +1,4 @@ +import { inspect } from "node:util"; import { APICallError } from "ai"; /** @@ -80,27 +81,168 @@ export function extractStatusCode(err: unknown): number | undefined { return undefined; } -/** Original-case error text for surfacing to humans. Concatenates - * `message` + `responseBody` so OpenRouter's `metadata.raw` wrapping - * of upstream provider errors is included verbatim. Walks one-hop - * `.cause` chain to find a wrapped APICallError — `streamText` often - * hands `onError` a generic Error whose `cause` is the real - * APICallError with the response body attached. */ -export function extractErrorText(err: unknown): string { +/** Original-case error text for surfacing to humans. Handles SDK Error / + * APICallError instances and nested provider error objects. If no semantic + * error text can be found, falls back to `dumpUnknown()` so plain objects + * never surface as `[object Object]`. */ +const DUMP_MAX_CHARS = 4096; + +function isUsefulErrorText(text: string): boolean { + const trimmed = text.trim(); + return trimmed.length > 0 && trimmed !== "[object Object]"; +} + +function truncateDump(text: string): string { + return text.length > DUMP_MAX_CHARS + ? `${text.slice(0, DUMP_MAX_CHARS)}...[truncated]` + : text; +} + +function objectEntries(value: object): Array<[string, unknown]> { + return Object.keys(value).map((key) => { + try { + return [key, (value as Record)[key]]; + } catch (e) { + return [key, `[threw while reading property: ${String(e)}]`]; + } + }); +} + +function normalizeForDump(value: unknown, seen: WeakSet): unknown { + if ( + value == null || + typeof value === "string" || + typeof value === "number" || + typeof value === "boolean" + ) { + return value; + } + if (typeof value === "bigint") return value.toString(); + if (typeof value === "symbol") return value.toString(); + if (typeof value === "function") return `[Function ${value.name || "anonymous"}]`; + if (typeof value !== "object") return String(value); + if (seen.has(value)) return "[Circular]"; + seen.add(value); + + if (value instanceof Error) { + const errorDump: Record = { + name: value.name || "Error", + }; + if (isUsefulErrorText(value.message)) errorDump.message = value.message; + if (value.cause && value.cause !== value) { + errorDump.cause = normalizeForDump(value.cause, seen); + } + for (const [key, child] of objectEntries(value)) { + if (!(key in errorDump)) errorDump[key] = normalizeForDump(child, seen); + } + return errorDump; + } + + if (Array.isArray(value)) { + return value.map((item) => normalizeForDump(item, seen)); + } + + const dump: Record = {}; + for (const [key, child] of objectEntries(value)) { + dump[key] = normalizeForDump(child, seen); + } + return dump; +} + +function dumpUnknown(value: unknown): string { + try { + const normalized = normalizeForDump(value, new WeakSet()); + const json = JSON.stringify(normalized); + if (json && isUsefulErrorText(json)) return truncateDump(json); + } catch { + // Fall through to inspect/String. + } + + try { + const inspected = inspect(value, { + breakLength: 120, + colors: false, + depth: 6, + getters: false, + maxArrayLength: 50, + maxStringLength: DUMP_MAX_CHARS, + }); + if (isUsefulErrorText(inspected)) return truncateDump(inspected); + } catch { + // Fall through to String(value). + } + + const text = String(value); + return isUsefulErrorText(text) ? truncateDump(text) : "Unknown error"; +} + +function extractNestedErrorText( + value: unknown, + seen: WeakSet +): string { + if (value == null) return ""; + const inner = extractErrorTextInner(value, seen); + return isUsefulErrorText(inner) ? inner : ""; +} + +function extractErrorTextInner( + err: unknown, + seen: WeakSet +): string { if (typeof err === "string") return err; + if ( + typeof err === "number" || + typeof err === "boolean" || + typeof err === "bigint" + ) { + return String(err); + } + if (err == null) return String(err); if (APICallError.isInstance(err)) { - const parts = [err.message ?? "", err.responseBody ?? ""].filter(Boolean); + const parts = [ + err.message ?? "", + typeof err.responseBody === "string" + ? err.responseBody + : extractNestedErrorText(err.responseBody, seen), + ].filter((part) => isUsefulErrorText(part)); return parts.join("\n").trim(); } + if (err instanceof Error) { + if (err.cause && err.cause !== err) { + const inner = extractErrorTextInner(err.cause, seen); + if (isUsefulErrorText(inner)) return inner; + } + if (isUsefulErrorText(err.message)) return err.message; + return dumpUnknown(err); + } if (err && typeof err === "object") { - const e = err as { message?: string; cause?: unknown }; + if (seen.has(err)) return ""; + seen.add(err); + const e = err as Record; if (e.cause && e.cause !== err) { - const inner = extractErrorText(e.cause); + const inner = extractErrorTextInner(e.cause, seen); + if (isUsefulErrorText(inner)) return inner; + } + for (const key of [ + "message", + "error", + "responseBody", + "body", + "data", + "detail", + "details", + "errors", + ]) { + const inner = extractNestedErrorText(e[key], seen); if (inner) return inner; } - return e.message ?? String(err); + return dumpUnknown(err); } - return String(err); + return dumpUnknown(err); +} + +export function extractErrorText(err: unknown): string { + return extractErrorTextInner(err, new WeakSet()).trim(); } function extractText(err: unknown): string { diff --git a/packages/agent-core/src/extractor.ts b/packages/agent-core/src/extractor.ts index 47179e72..4a953e96 100644 --- a/packages/agent-core/src/extractor.ts +++ b/packages/agent-core/src/extractor.ts @@ -11,6 +11,7 @@ import { scanMemoryFiles, } from "@openacme/memory"; import type { Agent } from "./agent.js"; +import { extractErrorText } from "./error-classifier.js"; import { runSubagent, type ForkedSubagentResult } from "./subagent.js"; const EXTRACTOR_STEP_CAP = 10; @@ -143,7 +144,7 @@ export async function runExtractor( } catch (e) { return { status: "failed", - error: e instanceof Error ? e.message : String(e), + error: extractErrorText(e), }; } diff --git a/packages/agent-core/src/subagent.ts b/packages/agent-core/src/subagent.ts index f03a02d9..b1dd0bde 100644 --- a/packages/agent-core/src/subagent.ts +++ b/packages/agent-core/src/subagent.ts @@ -22,6 +22,7 @@ import { z, type ZodTypeAny } from "zod"; import { resolveSubagentModel } from "@openacme/llm-provider"; import type { UsageKind } from "@openacme/db"; import type { Agent } from "./agent.js"; +import { extractErrorText } from "./error-classifier.js"; import type { TokenUsage } from "./types.js"; const DEFAULT_TIMEOUT_MS = 120_000; @@ -201,7 +202,7 @@ async function runForked( mode: "forked", status: "failed", message: assembled, - error: e instanceof Error ? e.message : String(e), + error: extractErrorText(e), }; } } @@ -286,7 +287,7 @@ async function runStructured( mode: "structured", status: "failed", object: null, - error: e instanceof Error ? e.message : String(e), + error: extractErrorText(e), }; } } diff --git a/packages/agent-core/test/agent-fire-title.test.ts b/packages/agent-core/test/agent-fire-title.test.ts new file mode 100644 index 00000000..5757fb38 --- /dev/null +++ b/packages/agent-core/test/agent-fire-title.test.ts @@ -0,0 +1,147 @@ +import { describe, it, expect, vi, beforeEach } from "vitest"; +import * as fs from "node:fs"; +import * as os from "node:os"; +import * as path from "node:path"; +import { + applySchema, + WasmDatabase, + createSessionStore, + createMessageStore, + createInboxStore, +} from "@openacme/db"; +import { MemoryStore } from "@openacme/memory"; +import { TaskStore } from "@openacme/tasks"; +import type { ToolRegistry } from "@openacme/tools"; +import type { UIMessage } from "ai"; +import { Agent, type AutonomousBroadcaster } from "../src/agent.js"; +import type { AgentConfig } from "../src/types.js"; +import * as titleModule from "../src/title.js"; + +const stubToolRegistry = { + get: () => undefined, + getVercelTools: () => ({}), +} as unknown as ToolRegistry; + +function freshDb() { + const db = new WasmDatabase(":memory:"); + db.pragma("foreign_keys = ON"); + applySchema(db); + return db; +} + +function makeAgent(): { + agent: Agent; + broadcasts: Array<{ + sessionId: string; + event: Parameters[1]; + }>; +} { + const tmpRoot = fs.mkdtempSync(path.join(os.tmpdir(), "openacme-title-")); + const db = freshDb(); + const sessionStore = createSessionStore(db); + const messageStore = createMessageStore(db); + const broadcasts: Array<{ + sessionId: string; + event: Parameters[1]; + }> = []; + const broadcaster: AutonomousBroadcaster = { + broadcast(sessionId, event) { + broadcasts.push({ sessionId, event }); + }, + }; + const config: AgentConfig = { + id: "a1", + name: "A1", + model: { + provider: "openai", + model: "test", + apiKey: "x", + auth: "api_key", + }, + persona: "test", + tools: [], + maxSteps: 1, + }; + const agent = new Agent(config, { + sessionStore, + messageStore, + toolRegistry: stubToolRegistry, + attachmentsRoot: path.join(tmpRoot, "att"), + memoryStore: new MemoryStore(path.join(tmpRoot, "agents")), + taskStore: new TaskStore(path.join(tmpRoot, "tasks")), + inboxStore: createInboxStore(db), + broadcaster, + }); + return { agent, broadcasts }; +} + +function user(id: string, text: string): UIMessage { + return { id, role: "user", parts: [{ type: "text", text }] }; +} + +function asst(id: string, text: string): UIMessage { + return { id, role: "assistant", parts: [{ type: "text", text }] }; +} + +async function fireAndSettle( + agent: Agent, + args: Parameters[0], +): Promise { + agent.fireTitle(args); + for (let i = 0; i < 20; i++) await Promise.resolve(); +} + +describe("Agent.fireTitle", () => { + beforeEach(() => { + vi.restoreAllMocks(); + }); + + it("broadcasts the generated title after writing it", async () => { + const { agent, broadcasts } = makeAgent(); + agent.sessionStore.create(agent.config.id, { id: "s1" }); + vi.spyOn(titleModule, "runTitle").mockResolvedValue("OAuth refresh bug"); + + await fireAndSettle(agent, { + sessionId: "s1", + sessionMessages: [ + user("u1", "Why does OAuth refresh fail?"), + asst("a1", "The refresh token is stale."), + ], + }); + + expect(agent.sessionStore.get("s1")?.title).toBe("OAuth refresh bug"); + expect(broadcasts).toEqual([ + { + sessionId: "s1", + event: { kind: "session_title", title: "OAuth refresh bug" }, + }, + ]); + }); + + it("broadcasts the fallback title when generation returns empty", async () => { + const { agent, broadcasts } = makeAgent(); + agent.sessionStore.create(agent.config.id, { id: "s1" }); + vi.spyOn(titleModule, "runTitle").mockResolvedValue(null); + + await fireAndSettle(agent, { + sessionId: "s1", + sessionMessages: [ + user("u1", "Summarize the build failure."), + asst("a1", "The build failed because the title update was never broadcast."), + ], + }); + + expect(agent.sessionStore.get("s1")?.title).toBe( + "The build failed because the title update was never broadcast.", + ); + expect(broadcasts).toEqual([ + { + sessionId: "s1", + event: { + kind: "session_title", + title: "The build failed because the title update was never broadcast.", + }, + }, + ]); + }); +}); diff --git a/packages/agent-core/test/error-classifier.test.ts b/packages/agent-core/test/error-classifier.test.ts new file mode 100644 index 00000000..701bf71d --- /dev/null +++ b/packages/agent-core/test/error-classifier.test.ts @@ -0,0 +1,48 @@ +import { describe, expect, it } from "vitest"; +import { extractErrorText } from "../src/error-classifier.js"; + +describe("extractErrorText", () => { + it("extracts nested provider error messages from plain objects", () => { + const err = { + status: 400, + error: { + message: "Unsupported model gpt-5.2 for OpenAI OAuth", + type: "invalid_request_error", + code: "model_not_found", + }, + }; + + expect(extractErrorText(err)).toBe( + "Unsupported model gpt-5.2 for OpenAI OAuth" + ); + }); + + it("falls back to readable JSON for object errors without a message", () => { + const err: Record = { status: 500, code: "upstream_error" }; + err.self = err; + + expect(extractErrorText(err)).toBe( + '{"status":500,"code":"upstream_error","self":"[Circular]"}' + ); + }); + + it("does not surface object-stringified Error messages", () => { + const text = extractErrorText(new Error("[object Object]")); + + expect(text).toBe('{"name":"Error"}'); + expect(text).not.toContain("[object Object]"); + }); + + it("dumps fallback values that JSON.stringify would drop or reject", () => { + function retry() {} + const text = extractErrorText({ + code: 123n, + retry, + marker: Symbol("provider"), + }); + + expect(text).toContain('"code":"123"'); + expect(text).toContain('"retry":"[Function retry]"'); + expect(text).toContain('"marker":"Symbol(provider)"'); + }); +}); diff --git a/packages/server/src/app.ts b/packages/server/src/app.ts index f27456f2..22f5353f 100644 --- a/packages/server/src/app.ts +++ b/packages/server/src/app.ts @@ -2032,20 +2032,30 @@ async function runChatTurn(args: { // Error branch: stream failed mid-turn (provider 4xx/5xx, network // drop, etc.). Append an upstream-error part so the user sees what // failed; preserve whatever assembled before the failure. - const parts = !capturedError || signal.aborted - ? ensureStepBoundaries( - finalizeOrphanToolParts( - responseMessage.parts as UIMessage["parts"] - ) - ) - : [ - ...ensureStepBoundaries( - finalizeOrphanToolParts( - responseMessage.parts as UIMessage["parts"] - ) - ), - buildUpstreamErrorPart(capturedError, agentId, manager), - ]; + const baseParts = ensureStepBoundaries( + finalizeOrphanToolParts( + responseMessage.parts as UIMessage["parts"] + ) + ); + let parts = baseParts; + if (capturedError && !signal.aborted) { + const upstreamErrorPart = buildUpstreamErrorPart( + capturedError, + agentId, + manager + ); + log.warn( + { + sessionId, + agentId, + provider: upstreamErrorPart.data.provider, + statusCode: upstreamErrorPart.data.statusCode, + message: upstreamErrorPart.data.message, + }, + "chat turn upstream provider error" + ); + parts = [...baseParts, upstreamErrorPart]; + } manager.messageStore.append(sessionId, { id: responseMessage.id, role: responseMessage.role as "user" | "assistant", diff --git a/packages/server/src/broadcaster.ts b/packages/server/src/broadcaster.ts index 99dc0481..e5a5a11a 100644 --- a/packages/server/src/broadcaster.ts +++ b/packages/server/src/broadcaster.ts @@ -1,6 +1,7 @@ /** * In-memory per-session pub/sub for live updates to web clients. - * Shared by the scheduler (session_state), event store (task_event), + * Shared by the scheduler (session_state), title writer + * (session_title), event store (task_event), * both /api/chat and Agent.runAutonomous (ui_message_part chunks + * messages_appended for user/auto messages), and the SSE routes. * @@ -49,6 +50,11 @@ export type SessionBroadcastEvent = kind: "session_state"; state: "running" | "idle"; } + | { + /** Session metadata changed after a post-turn title write. */ + kind: "session_title"; + title: string; + } | { kind: "task_event"; /** A TaskEventRow from the EventStore — already serialized form diff --git a/packages/server/src/dispatcher.ts b/packages/server/src/dispatcher.ts index 0be3a692..582942a9 100644 --- a/packages/server/src/dispatcher.ts +++ b/packages/server/src/dispatcher.ts @@ -28,7 +28,7 @@ */ import type { TaskStore, Task } from "@openacme/tasks"; -import { AutonomousTurnTimeout } from "@openacme/agent-core"; +import { AutonomousTurnTimeout, extractErrorText } from "@openacme/agent-core"; import type { SessionStore, InboxStore } from "@openacme/db"; import { createLogger } from "@openacme/config/logger"; import type { AgentManager } from "./agent-manager.js"; @@ -420,7 +420,7 @@ export class Dispatcher { try { await agent.runAutonomous({ sessionId }); } catch (e) { - const message = e instanceof Error ? e.message : String(e); + const message = extractErrorText(e); const isTimeout = e instanceof AutonomousTurnTimeout; if (!isTimeout) { log.warn({ sessionId, message }, "autonomous turn failed"); diff --git a/packages/server/src/routes/streams.ts b/packages/server/src/routes/streams.ts index b716ca63..2c0913fc 100644 --- a/packages/server/src/routes/streams.ts +++ b/packages/server/src/routes/streams.ts @@ -2,8 +2,9 @@ * SSE routes for live workforce + per-session streams. * * - `GET /api/sessions/:id/stream` — per-session push channel. Emits - * `ui_message_part`, `messages_appended`, `session_state`, and - * `task_event` envelopes. Replays the broadcaster's ring buffer on + * `ui_message_part`, `messages_appended`, `session_state`, + * `session_title`, and `task_event` envelopes. Replays the + * broadcaster's ring buffer on * reconnect (Last-Event-ID present) so a brief disconnect doesn't * drop a streaming turn. Fresh connections are forward-only — past * messages come from DB history, not the buffer. diff --git a/packages/server/test/app-upstream-log.test.ts b/packages/server/test/app-upstream-log.test.ts new file mode 100644 index 00000000..b68213cd --- /dev/null +++ b/packages/server/test/app-upstream-log.test.ts @@ -0,0 +1,108 @@ +import { randomUUID } from "node:crypto"; +import { + existsSync, + mkdtempSync, + readFileSync, + rmSync, +} from "node:fs"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { describe, expect, it, vi } from "vitest"; + +describe("chat upstream error logging", () => { + it("persists the provider error and writes a meaningful log line", async () => { + const prevDataDir = process.env["OPENACME_DATA_DIR"]; + const prevLogFile = process.env["OPENACME_LOG_FILE"]; + const dataDir = mkdtempSync(path.join(tmpdir(), "openacme-log-test-")); + const logFile = path.join(dataDir, "openacme.log"); + let manager: { close: () => Promise } | undefined; + + try { + vi.resetModules(); + process.env["OPENACME_DATA_DIR"] = dataDir; + process.env["OPENACME_LOG_FILE"] = logFile; + + const { ConfigSchema } = await import("@openacme/config"); + const { createApp } = await import("../src/app.js"); + const { createStubModel } = await import( + "./e2e/support/stub-model.mjs" + ); + + const config = ConfigSchema.parse({ + dataDir, + model: { + provider: "custom", + model: "stub-1", + baseUrl: "http://127.0.0.1:9/v1", + apiKey: "stub", + }, + server: { host: "127.0.0.1", requireAuth: false }, + }); + const created = await createApp(config, { + resolveModel: () => createStubModel(), + }); + manager = created.manager; + + const headers = { + host: "127.0.0.1", + "content-type": "application/json", + }; + const createRes = await created.app.request("/api/agents", { + method: "POST", + headers, + body: JSON.stringify({ id: "helper", name: "Helper" }), + }); + expect(createRes.status).toBe(201); + + const sessionId = randomUUID(); + const failureText = "scripted failure object log check"; + const chatRes = await created.app.request("/api/chat", { + method: "POST", + headers, + body: JSON.stringify({ + agentId: "helper", + sessionId, + messages: [ + { + id: randomUUID(), + role: "user", + parts: [ + { type: "text", text: `break [[mock:error:${failureText}]]` }, + ], + }, + ], + }), + }); + expect(chatRes.status).toBe(200); + + for (let i = 0; i < 80; i++) { + const assistant = created.manager.messageStore + .getHistory(sessionId) + .find((m) => m.role === "assistant"); + if (assistant) break; + await new Promise((r) => setTimeout(r, 100)); + } + + const assistant = created.manager.messageStore + .getHistory(sessionId) + .find((m) => m.role === "assistant"); + const errorPart = assistant?.parts.find( + (p) => p?.type === "data-upstream-error" + ) as { data?: { message?: string } } | undefined; + expect(errorPart?.data?.message).toContain(failureText); + + const log = existsSync(logFile) ? readFileSync(logFile, "utf8") : ""; + expect(log).toContain("chat turn upstream provider error"); + expect(log).toContain(failureText); + expect(log).not.toContain("[object Object]"); + } finally { + await manager?.close(); + if (prevDataDir === undefined) delete process.env["OPENACME_DATA_DIR"]; + else process.env["OPENACME_DATA_DIR"] = prevDataDir; + if (prevLogFile === undefined) delete process.env["OPENACME_LOG_FILE"]; + else process.env["OPENACME_LOG_FILE"] = prevLogFile; + rmSync(dataDir, { recursive: true, force: true }); + vi.resetModules(); + } + }); +}); diff --git a/packages/server/test/dispatcher.test.ts b/packages/server/test/dispatcher.test.ts index c5cc41ea..f6cbedef 100644 --- a/packages/server/test/dispatcher.test.ts +++ b/packages/server/test/dispatcher.test.ts @@ -5,6 +5,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { ConfigSchema } from "@openacme/config"; import { createDatabase, + createCommentStore, createSessionStore, createInboxStore, } from "@openacme/db"; @@ -56,7 +57,9 @@ beforeEach(() => { db = createDatabase(config); sessionStore = createSessionStore(db); inboxStore = createInboxStore(db); - taskStore = new TaskStore(path.join(dataDir, "tasks")); + taskStore = new TaskStore(path.join(dataDir, "tasks"), { + commentStore: createCommentStore(db), + }); dispatcher = null; }); @@ -281,6 +284,33 @@ describe("Dispatcher failure handling", () => { expect(retryAt).toBeLessThan(Date.now() + 6 * 60_000); }); + it("parks plain object turn errors with a readable provider message", async () => { + const { manager } = fakeManager(["a1"], async () => { + throw { + status: 400, + error: { + message: "Unsupported model gpt-5.2 for OpenAI OAuth", + type: "invalid_request_error", + }, + }; + }); + const { task } = await makeBoundTask("a1", { + status: "in_progress", + }); + + const d = makeDispatcher(manager); + await d.start(); + await d.drain(5_000); + + const parked = taskStore.get(task.id); + expect(parked?.status).toBe("blocked"); + const comments = taskStore.listComments(task.id, { kinds: ["system"] }); + expect(comments.at(-1)?.body).toContain( + "Unsupported model gpt-5.2 for OpenAI OAuth" + ); + expect(comments.at(-1)?.body).not.toContain("[object Object]"); + }); + it("startup sweep resets stale in_progress tasks to open", async () => { // Assignee unknown to the manager: nothing can spawn, isolating // the sweep itself. diff --git a/packages/server/test/e2e/tasks.e2e.ts b/packages/server/test/e2e/tasks.e2e.ts index 27dd77d5..7d38b1ab 100644 --- a/packages/server/test/e2e/tasks.e2e.ts +++ b/packages/server/test/e2e/tasks.e2e.ts @@ -106,4 +106,52 @@ describe("autonomous dispatch (e2e)", () => { sse.close(); }); + + it("surfaces autonomous model errors into the session", async () => { + const session = srv.manager.sessionStore.create("worker"); + const sessionId = session.id; + const sse = await openSSE(`${srv.baseUrl}/api/sessions/${sessionId}/stream`); + + const messageId = randomUUID(); + srv.manager.inboxStore.deliver({ + agentId: "worker", + kind: "user_message", + source: "user", + sourceId: messageId, + relatedSession: sessionId, + payload: { + id: messageId, + role: "user", + parts: [ + { + type: "text", + text: "break autonomous turn [[mock:error:autonomous failure]]", + }, + ], + }, + }); + + try { + await sse.waitFor(isState("running"), 8_000); + await sse.waitFor(isState("idle"), 12_000); + + await waitUntil(async () => { + const messages = await c.messages(sessionId); + return messages.some((m) => + m.role === "assistant" && + m.parts.some((p) => p?.type === "data-upstream-error") + ); + }); + const assistant = (await c.messages(sessionId)).find((m) => + m.role === "assistant" && + m.parts.some((p) => p?.type === "data-upstream-error") + ); + const errorPart = assistant!.parts.find( + (p) => p?.type === "data-upstream-error" + ); + expect(errorPart.data.message).toContain("autonomous failure"); + } finally { + sse.close(); + } + }); });