fix(appkit): address pkosiec feedback

hubertzub-db · hubertzub-db · commit 6ee2efb8f4e3 · 2026-07-27T10:55:55.000+02:00
Signed-off-by: Hubert Zub &lt;hubert.zub@databricks.com&gt;
diff --git a/packages/appkit/src/agents/supervisor-api.ts b/packages/appkit/src/agents/supervisor-api.ts
@@ -30,7 +30,15 @@ type SupervisorErrorCode =
  * never sees raw upstream text.
  */
 function emitError(code: SupervisorErrorCode, detail: unknown): AgentEvent {
-  logger.warn("supervisor-api error code=%s detail=%O", code, detail);
+  // Summarise at `warn` (CWE-532: never dump the full upstream payload to
+  // the default log level); the verbose object is only available via
+  // `DEBUG=appkit:agents:supervisor-api`.
+  logger.warn(
+    "supervisor-api error code=%s detail=%s",
+    code,
+    summariseErrorPayload(detail),
+  );
+  logger.debug("supervisor-api error code=%s detail=%O", code, detail);
   return {
     type: "status",
     status: "error",
@@ -353,51 +361,64 @@ export class SupervisorApiAdapter implements AgentAdapter {
         }
       | undefined;
 
-    for await (const { event, data } of readSseEvents(stream, signal)) {
-      if (data === "[DONE]") continue;
-
-      let parsed: Record<string, unknown>;
-      try {
-        parsed = JSON.parse(data);
-      } catch (err) {
-        logger.debug(
-          "Failed to parse SSE data line: %s (%O)",
-          data.slice(0, 200),
-          err,
-        );
-        continue;
-      }
+    // `readSseEvents` throws on transport errors and on the DoS caps
+    // (maxLineChars / maxBufferChars). Without this guard the rejection
+    // propagates out of `run()` and tears down the request. Treat a
+    // consumer-initiated abort as a clean stop; everything else becomes a
+    // sanitised terminal `transport` error.
+    try {
+      for await (const { event, data } of readSseEvents(stream, signal)) {
+        if (data === "[DONE]") continue;
+
+        let parsed: Record<string, unknown>;
+        try {
+          parsed = JSON.parse(data);
+        } catch (err) {
+          logger.debug(
+            "Failed to parse SSE data line: %s (%O)",
+            data.slice(0, 200),
+            err,
+          );
+          continue;
+        }
 
-      const eventType = event || (parsed.type as string) || "";
-      eventCounts.set(eventType, (eventCounts.get(eventType) ?? 0) + 1);
-
-      // `response.completed` is held back until after the loop so we can
-      // synthesise a `message_delta` from `response.output[]` when the
-      // stream produced no incremental deltas (intermittent SA behaviour).
-      // Emitting `complete` first would let UIs finalise the turn before the
-      // recovered text arrives.
-      if (eventType === "response.completed") {
-        lastCompleted = parsed.response as typeof lastCompleted;
-        continue;
-      }
+        const eventType =
+          event || (typeof parsed.type === "string" ? parsed.type : "");
+        eventCounts.set(eventType, (eventCounts.get(eventType) ?? 0) + 1);
+
+        // `response.completed` is held back until after the loop so we can
+        // synthesise a `message_delta` from `response.output[]` when the
+        // stream produced no incremental deltas (intermittent SA behaviour).
+        // Emitting `complete` first would let UIs finalise the turn before the
+        // recovered text arrives.
+        if (eventType === "response.completed") {
+          lastCompleted = parsed.response as typeof lastCompleted;
+          continue;
+        }
 
-      const out = mapEvent(eventType, parsed, streamedItemIds);
-      if (out) {
-        if (out.type === "message_delta") receivedAnyDelta = true;
-        yield out;
-        if (out.type === "status" && out.status === "error") {
-          terminated = true;
-          break;
+        const out = mapEvent(eventType, parsed, streamedItemIds);
+        if (out) {
+          if (out.type === "message_delta") receivedAnyDelta = true;
+          yield out;
+          if (out.type === "status" && out.status === "error") {
+            terminated = true;
+            break;
+          }
         }
       }
+    } catch (err) {
+      if (signal?.aborted) return;
+      yield emitError("transport", err);
+      return;
     }
 
     if (signal?.aborted) return;
 
     if (eventCounts.size === 0) {
-      logger.warn(
-        "Supervisor API stream closed without emitting any SSE events.",
-      );
+      // A stream that closes without a single event leaves the consumer
+      // stuck in `running`. Surface a terminal `transport` error so the
+      // turn ends.
+      yield emitError("transport", "stream closed without events");
       return;
     }
 
@@ -419,14 +440,15 @@ export class SupervisorApiAdapter implements AgentAdapter {
 
     if (eventCounts.has("response.completed")) {
       // SA sometimes signals a failed turn via `response.completed` with a
-      // nested `status: "failed"` (or a populated `error`/`incomplete_details`)
-      // rather than emitting `response.failed`. Without this gate the
-      // adapter would silently yield `complete` on a server-side failure.
-      if (
-        lastCompleted?.status === "failed" ||
-        lastCompleted?.error != null ||
-        lastCompleted?.incomplete_details != null
-      ) {
+      // nested `status: "failed"` (or a populated `error`) rather than
+      // emitting `response.failed`. Without this gate the adapter would
+      // silently yield `complete` on a server-side failure.
+      //
+      // `incomplete_details` on its own is NOT fatal: a benign
+      // `max_output_tokens` truncation populates it while still producing
+      // usable partial output. In that case we fall through to `complete`
+      // and let the recovered text above stand as the turn result.
+      if (lastCompleted?.status === "failed" || lastCompleted?.error != null) {
         yield emitError("upstream_failed", {
           status: lastCompleted?.status,
           error: lastCompleted?.error,
@@ -539,9 +561,13 @@ function mapEvent(
   // than we care to map.
   switch (eventType as ResponseStreamEvent["type"]) {
     case "response.output_text.delta": {
-      const itemId = data.item_id as string | undefined;
+      const itemId =
+        typeof data.item_id === "string" ? data.item_id : undefined;
       if (itemId) streamedItemIds.add(itemId);
-      return { type: "message_delta", content: (data.delta as string) ?? "" };
+      return {
+        type: "message_delta",
+        content: typeof data.delta === "string" ? data.delta : "",
+      };
     }
 
     // `response.completed` is intentionally absent: `streamResponse` holds
diff --git a/packages/appkit/src/agents/tests/supervisor-api.test.ts b/packages/appkit/src/agents/tests/supervisor-api.test.ts
@@ -770,6 +770,114 @@ describe("SupervisorApiAdapter", () => {
     });
     expect(events).toContainEqual({ type: "status", status: "complete" });
   });
+
+  test("yields a sanitised transport error when readSseEvents throws mid-stream", async () => {
+    // `readSseEvents` enforces a DoS cap (maxLineChars) and throws when an
+    // SSE block exceeds it. Without a try/catch around the consuming loop
+    // that rejection would propagate out of run() and tear down the request.
+    // The adapter must catch it and surface a terminal `transport` error.
+    const oversizedBlock = `data: ${"x".repeat(1024 * 1024 + 100)}\n\n`;
+    const { streamBody } = makeStreamBody([oversizedBlock]);
+    const adapter = new SupervisorApiAdapter({
+      streamBody,
+      model: "databricks-claude-sonnet-4",
+    });
+    const events = await collect(
+      adapter.run(createInput(), { executeTool: vi.fn() }),
+    );
+    expect(events).toEqual([
+      { type: "status", status: "running" },
+      {
+        type: "status",
+        status: "error",
+        error: "Supervisor API error (transport)",
+      },
+    ]);
+    // The raw error text (which echoes the cap detail) never reaches the client.
+    for (const e of events) {
+      if (e.type === "status" && "error" in e) {
+        expect(e.error).not.toContain("maxLineChars");
+      }
+    }
+  });
+
+  test("emits a terminal transport error when the stream closes without any events", async () => {
+    // A stream that closes with zero SSE events would otherwise leave the
+    // consumer stuck in `running`. The adapter must end the turn with a
+    // terminal `transport` error.
+    const { streamBody } = makeStreamBody([]);
+    const adapter = new SupervisorApiAdapter({
+      streamBody,
+      model: "databricks-claude-sonnet-4",
+    });
+    const events = await collect(
+      adapter.run(createInput(), { executeTool: vi.fn() }),
+    );
+    expect(events).toEqual([
+      { type: "status", status: "running" },
+      {
+        type: "status",
+        status: "error",
+        error: "Supervisor API error (transport)",
+      },
+    ]);
+  });
+
+  test("treats incomplete_details alone as benign truncation: recovers text and completes", async () => {
+    // A `max_output_tokens` truncation populates `incomplete_details` while
+    // still producing usable partial output. The adapter must NOT treat this
+    // as a failure — it recovers the partial text and yields `complete`.
+    const { streamBody } = makeStreamBody([
+      sseEvent("response.completed", {
+        response: {
+          status: "incomplete",
+          incomplete_details: { reason: "max_output_tokens" },
+          output: [
+            {
+              type: "message",
+              id: "msg-trunc",
+              role: "assistant",
+              content: [{ type: "output_text", text: "Partial answer" }],
+            },
+          ],
+        },
+      }),
+    ]);
+    const adapter = new SupervisorApiAdapter({
+      streamBody,
+      model: "databricks-claude-sonnet-4",
+    });
+    const events = await collect(
+      adapter.run(createInput(), { executeTool: vi.fn() }),
+    );
+    expect(events).toEqual([
+      { type: "status", status: "running" },
+      { type: "message_delta", content: "Partial answer" },
+      { type: "status", status: "complete" },
+    ]);
+  });
+
+  test("coerces a non-string output_text delta to an empty string", async () => {
+    // Hardening for the previous `(data.delta as string) ?? ""` cast: a
+    // non-string `delta` (or `item_id`) must not leak through as a non-string
+    // or `"undefined"` — it is coerced to "".
+    const { streamBody } = makeStreamBody([
+      sseEvent("response.output_text.delta", { item_id: 42, delta: 123 }),
+      sseEvent("response.completed", {}),
+    ]);
+    const adapter = new SupervisorApiAdapter({
+      streamBody,
+      model: "databricks-claude-sonnet-4",
+    });
+    const events = await collect(
+      adapter.run(createInput(), { executeTool: vi.fn() }),
+    );
+    expect(events).toEqual([
+      { type: "status", status: "running" },
+      { type: "message_delta", content: "" },
+      { type: "status", status: "complete" },
+    ]);
+  });
 });
 
 describe("fromSupervisorApi", () => {
