Surface autonomous provider errors in chat

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();
     };

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<void> } {
   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<void>; 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 {

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]);

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,14 +1030,26 @@ 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`
       );
     }
 
     // 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);

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 {