fix(agent-sdk): show local command replies whose response never streamed

agent-sdk/src/bridge/activity.ts

@@ -7,6 +7,19 @@ export type MainAgentResponse = {
   thinkingBlock?: number;
 };
 
+/**
+ * Whether a completed top-level assistant message belongs to a response whose
+ * stream this session never saw begin. Its blocks arrive in order inside that
+ * response, before the next one starts, so the last started response is the
+ * only one a completed message can continue. Unlike the activity scope, a
+ * stop, retry or reset does not forget it. A message without a response ID
+ * cannot be correlated and counts as streamed.
+ */
+export function responseNeverStreamed(session: SessionState, responseId: unknown): boolean {
+  return typeof responseId === "string" && responseId.length > 0 &&
+    responseId !== session.lastStreamedResponseId;
+}
+
 export function resetMainAgentActivity(session: SessionState): void {
   if (session.mainAgentResponse?.thinkingBlock !== undefined) {
     emitSessionUpdate(session.sessionId, { type: "agent_activity_update", phase: "working" });
@@ -23,6 +36,7 @@ export function observeMainAgentStream(
   if (event.type === "message_start") {
     const message = event.message as Record<string, unknown> | undefined;
     if (typeof message?.id !== "string" || !message.id) return;
+    session.lastStreamedResponseId = message.id;
     if (session.mainAgentResponse?.messageId === message.id) return;
     resetMainAgentActivity(session);
     session.mainAgentResponse = { messageId: message.id };

agent-sdk/src/bridge/message_handlers.test.ts

@@ -0,0 +1,172 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+import type { SDKMessage } from "@anthropic-ai/claude-agent-sdk";
+import { replaceProtocolEventWriter } from "./events.js";
+import { handleSdkMessage } from "./message_handlers.js";
+import type { SessionState } from "./session_lifecycle.js";
+
+function makeSession(): SessionState {
+  return {
+    sessionId: "session-1",
+    cwd: "C:/work",
+    availableModels: [],
+    mode: null,
+    supportedModeIds: [],
+    runtimeUnavailableModeIds: [],
+    supportsBypassPermissionsMode: false,
+    fastModeState: "off",
+    connected: true,
+    closing: false,
+    connectEvent: "connected",
+    toolCalls: new Map(),
+    tasksById: new Map(),
+    taskOrder: [],
+    taskToolUseIds: new Map(),
+    taskIdsByToolUseId: new Map(),
+    pendingPermissions: new Map(),
+    pendingQuestions: new Map(),
+    pendingUserDialogs: new Map(),
+    pendingElicitations: new Map(),
+    informationalDedupKeys: new Set(),
+    knownConnectedMcpServers: new Set(),
+    mcpStatusRevalidatedAt: new Map(),
+    mcpAuthMonitors: new Map(),
+    hiddenToolUseIds: new Set(),
+    authHintSent: false,
+  } as unknown as SessionState;
+}
+
+/** The text of every `agent_message_chunk` the bridge wrote while `run` ran. */
+function replyChunks(run: () => void): Array<Record<string, unknown>> {
+  const writes: string[] = [];
+  const restore = replaceProtocolEventWriter((line) => {
+    writes.push(line);
+  });
+  try {
+    run();
+  } finally {
+    restore();
+  }
+  return writes
+    .map((line) => JSON.parse(line) as Record<string, unknown>)
+    .map((event) => event.update as Record<string, unknown> | undefined)
+    .filter((update): update is Record<string, unknown> => update?.type === "agent_message_chunk");
+}
+
+// Measured on Claude Code 2.1.288, the version the pinned SDK bundles (2.1.296
+// behaves the same): a local command (/rename, /color, /usage, /context)
+// replies with one complete top-level assistant message and no stream events,
+// and no system/local_command_output message, so its text exists only in that
+// message.
+test("a complete top-level reply whose response never streamed is shown once", () => {
+  const session = makeSession();
+  const reply = (uuid: string, responseId: string, text: string) => ({
+    type: "assistant",
+    uuid,
+    session_id: "session-1",
+    parent_tool_use_id: null,
+    message: {
+      id: responseId,
+      role: "assistant",
+      stop_reason: "end_turn",
+      content: [{ type: "text", text }],
+    },
+  });
+
+  const chunks = replyChunks(() => {
+    for (const message of [
+      reply("rename-reply", "a4cc84fe-7eb1-48bf-8fcd-0a11a7d5e251", "Session renamed to: probe"),
+      reply("color-reply", "5142ce8e-bb1c-4d99-a882-5752dd24280b", "Session color set to: blue"),
+    ]) {
+      handleSdkMessage(session, message as unknown as SDKMessage);
+    }
+  });
+
+  assert.deepEqual(chunks, [
+    {
+      type: "agent_message_chunk",
+      content: { type: "text", text: "Session renamed to: probe" },
+      source_message_uuid: "rename-reply",
+    },
+    {
+      type: "agent_message_chunk",
+      content: { type: "text", text: "Session color set to: blue" },
+      source_message_uuid: "color-reply",
+    },
+  ]);
+});
+
+test("a streamed, subagent, empty, error or uncorrelated reply is never shown again", () => {
+  const session = makeSession();
+  const emit = (message: Record<string, unknown>) =>
+    handleSdkMessage(session, {
+      session_id: "session-1",
+      parent_tool_use_id: null,
+      ...message,
+    } as unknown as SDKMessage);
+  const stream = (event: Record<string, unknown>) => emit({ type: "stream_event", uuid: "frame", event });
+  const complete = (uuid: string, message: Record<string, unknown>, extra = {}) =>
+    emit({ type: "assistant", uuid, message: { role: "assistant", ...message }, ...extra });
+
+  const texts = replyChunks(() => {
+    stream({ type: "message_start", message: { id: "msg_streamed" } });
+    stream({ type: "content_block_start", index: 0, content_block: { type: "text", text: "" } });
+    stream({ type: "content_block_delta", index: 0, delta: { type: "text_delta", text: "already streamed" } });
+    // The per-block message arrives inside the response; a terminal one may
+    // follow its stop. Neither is the reply's first appearance.
+    complete("streamed-block", { id: "msg_streamed", stop_reason: null, content: [{ type: "text", text: "already streamed" }] });
+    stream({ type: "content_block_stop", index: 0 });
+    stream({ type: "message_stop" });
+    complete("streamed-final", { id: "msg_streamed", stop_reason: "end_turn", content: [{ type: "text", text: "already streamed" }] });
+    complete("subagent", { id: "msg_subagent", content: [{ type: "text", text: "subagent text" }] }, { parent_tool_use_id: "tool-1" });
+    complete("empty", { id: "local-empty", content: [{ type: "text", text: "   " }] });
+    // A synthetic API error is reported through the turn result.
+    complete("synthetic-error", { id: "local-error", content: [{ type: "text", text: "API Error: rate limited" }] }, { error: "rate_limit" });
+    // Without a response ID nothing proves the text was not streamed.
+    complete("no-response-id", { content: [{ type: "text", text: "uncorrelated" }] });
+  }).map((chunk) => (chunk.content as Record<string, unknown>).text);
+
+  assert.deepEqual(texts, ["already streamed"]);
+});
+
+test("a non-streaming fallback after a partial stream is shown in full", () => {
+  const session = makeSession();
+  const emit = (message: Record<string, unknown>) =>
+    handleSdkMessage(session, {
+      session_id: "session-1",
+      parent_tool_use_id: null,
+      ...message,
+    } as unknown as SDKMessage);
+  const stream = (event: Record<string, unknown>) => emit({ type: "stream_event", uuid: "frame", event });
+
+  const texts = replyChunks(() => {
+    // The stream breaks off after a partial delta; the deltas already showed it.
+    stream({ type: "message_start", message: { id: "msg_partial" } });
+    stream({ type: "content_block_start", index: 0, content_block: { type: "text", text: "" } });
+    stream({ type: "content_block_delta", index: 0, delta: { type: "text_delta", text: "partial" } });
+    // The retried request answers without streaming, under a new response ID.
+    emit({
+      type: "assistant",
+      uuid: "fallback",
+      message: { id: "msg_fallback", role: "assistant", stop_reason: "end_turn", content: [{ type: "text", text: "the full reply" }] },
+    });
+  }).map((chunk) => (chunk.content as Record<string, unknown>).text);
+
+  assert.deepEqual(texts, ["partial", "the full reply"]);
+});
+
+test("SDK context usage Markdown is shown even without a response ID", () => {
+  const session = makeSession();
+  const texts = replyChunks(() => {
+    handleSdkMessage(session, {
+      type: "assistant",
+      uuid: "context-reply",
+      session_id: "session-1",
+      parent_tool_use_id: null,
+      context_usage: { categories: [] },
+      message: { role: "assistant", content: [{ type: "text", text: "## Context Usage" }] },
+    } as unknown as SDKMessage);
+  }).map((chunk) => (chunk.content as Record<string, unknown>).text);
+
+  assert.deepEqual(texts, ["## Context Usage"]);
+});

agent-sdk/src/bridge/message_handlers.ts

@@ -1,4 +1,4 @@
-import { observeMainAgentStream, resetMainAgentActivity } from "./activity.js";
+import { observeMainAgentStream, resetMainAgentActivity, responseNeverStreamed } from "./activity.js";
 import { nativeNotification } from "./notifications.js";
 import { elapsedNumber, messageMetadata, turnTiming } from "./presentation_metadata.js";
 import { refreshUltracode } from "./ultracode.js";
@@ -1274,7 +1274,16 @@ export function handleAssistantMessage(
   const content = Array.isArray(messageObject.content)
     ? messageObject.content
     : [];
-  if (asRecordOrNull(message.context_usage)) {
+  // A streamed reply was already shown from its deltas. SDK-owned context
+  // Markdown, and a complete top-level reply whose response never streamed (a
+  // local command such as /rename, /color or /usage), have no other carrier
+  // of their text. An error frame reports through the turn result instead.
+  if (
+    asRecordOrNull(message.context_usage) ||
+    (!message.parent_tool_use_id &&
+      assistantError.length === 0 &&
+      responseNeverStreamed(session, messageObject.id))
+  ) {
     const markdown = content
       .flatMap((block) => {
         const record = asRecordOrNull(block);

agent-sdk/src/bridge/session_lifecycle.ts

@@ -170,6 +170,8 @@ export type PendingWorkerShutdown = {
 
 export type SessionState = {
   mainAgentResponse?: MainAgentResponse;
+  /** The API response ID of the last top-level response whose stream began. */
+  lastStreamedResponseId?: string;
   ultracode?: import("../types.js").UltracodeSnapshot;
   sessionId: string;
   cwd: string;

docs/src/architecture.md

@@ -100,7 +100,7 @@ Shutdown redraws the live viewport without resetting committed history. Reading
 | Interpreting the live main-agent activity phase from SDK stream events | Bridge | `agent-sdk/src/bridge/activity.ts` |
 | Presentation decision from the typed `agent_activity_update` phase, turn liveness, pending interactions, compaction and cancellation | Rust | `src/app/activity.rs` |
 
-Only top-level `message_start` and indexed `thinking` or `redacted_thinking` block boundaries select the thinking phase, including blocks with empty or omitted text. A matching block stop, response stop or completion, retry or fallback, result or error, conversation replacement and session close reset it. SDK stream wrapper UUIDs identify individual frames; the API response ID scopes the ordered block stream and correlates completed assistant messages. A per-block assistant frame with a null stop reason keeps that scope open, and only a terminal, aborted or error frame closes its response. Stops carry no response ID and follow the SDK's ordered response boundaries. Subagent frames and completed history never start main-agent thinking. `system/thinking_tokens` estimates stay diagnostics, because they carry neither a response or block identity nor a matching stop.
+Only top-level `message_start` and indexed `thinking` or `redacted_thinking` block boundaries select the thinking phase, including blocks with empty or omitted text. A matching block stop, response stop or completion, retry or fallback, result or error, conversation replacement and session close reset it. SDK stream wrapper UUIDs identify individual frames; the API response ID scopes the ordered block stream and correlates completed assistant messages. A per-block assistant frame with a null stop reason keeps that scope open, and only a terminal, aborted or error frame closes its response. Stops carry no response ID and follow the SDK's ordered response boundaries. The same correlation decides where reply text comes from: a streamed response is shown from its deltas, and a completed top-level message whose response never streamed, such as a local command's reply, is shown from that message once; error messages report through the turn result. Subagent frames and completed history never start main-agent thinking. `system/thinking_tokens` estimates stay diagnostics, because they carry neither a response or block identity nor a matching stop.
 
 Rust consumes the typed phases without interpreting SDK content and never infers thinking from text, tools or silence. One derived decision assigns the activity heading to the active assistant, or to the composer for standalone work without an assistant message. SDK thinking appears separately in the active assistant output as a temporary live row that is never committed to history. Waiting or completing without output removes the temporary heading and activity; real assistant output keeps its heading. Composer-only headings and the composer activity block never enter transcript or history output.
 

docs/src/commands.md

@@ -66,6 +66,8 @@ Use:
 
 to inspect the current session's full command list. The output includes app-owned commands and SDK-advertised commands, with descriptions when the SDK provides them.
 
+Commands that Claude Code runs locally, such as `/rename`, `/color` or `/context`, reply in the chat under the command as they do in Claude Code. `/color` runs in Claude Code and its reply shows, but Claude Rust does not draw a session colour.
+
 ## Session Commands
 
 `/ultracode on` enables session-scoped dynamic-workflow orchestration while retaining the current thinking effort; `/ultracode off` disables it. Both require an idle turn. `/ultracode status` can be used during a turn and reports the latest verified SDK snapshot: on, available and off, requested but unavailable, unavailable and off, or unknown. Availability depends on SDK session capabilities and model support. The footer shows `Ultracode` only when it is verified as effective. Changing effort preserves active Ultracode, and changing models refreshes its state. This command does not persist a preference or control the one-turn `ultracode` keyword trigger.