fix(acp): stream agent-initiated turns

.changeset/stream-agent-initiated-acp-turns.md

@@ -0,0 +1,5 @@
+---
+"@moonshot-ai/kimi-code": patch
+---
+
+Stream agent-initiated turns to connected ACP sessions, preserve live updates while resuming, and return retryable errors instead of losing prompts during session handoff.

apps/vis/server/src/lib/blob-resolver.ts

@@ -46,7 +46,7 @@ function rehydrateRecord(
   baseUrl: string,
 ): void {
   const type = record['type'];
-  if (type === 'turn.prompt' || type === 'turn.steer') {
+  if (type === 'turn.prompt' || type === 'turn.steer' || type === 'turn.defer') {
     rehydrateParts(record['input'] as unknown as ContentPart[], sessionId, agentId, baseUrl);
     return;
   }

apps/vis/server/src/lib/context-projector.ts

@@ -505,6 +505,8 @@ export function projectContext(
       case 'forked':
       case 'turn.prompt':
       case 'turn.steer':
+      case 'turn.defer':
+      case 'turn.defer.consume':
       case 'turn.cancel':
       case 'permission.record_approval_result':
       case 'full_compaction.begin':

apps/vis/server/test/lib/blob-resolver.test.ts

@@ -130,6 +130,27 @@ describe('blob-resolver', () => {
       );
     });
 
+    it('resolves media held in a deferred turn input', () => {
+      const data: Record<string, unknown> = {
+        type: 'turn.defer',
+        id: 'deferred-1',
+        input: [
+          {
+            type: 'image_url',
+            imageUrl: { url: 'blobref:image/png;hashD' },
+          },
+        ],
+        origin: { kind: 'cron' },
+      };
+      const entries = [{ lineNo: 1, data: data as any, raw: {} }];
+
+      rehydrateWireEntries(entries, 'sess-4', 'main');
+
+      expect((entries[0]!.data as any).input[0].imageUrl.url).toBe(
+        '/api/sessions/sess-4/blobs/hashD?agent=main&mime=image%2Fpng',
+      );
+    });
+
     it('ignores records without media URLs', () => {
       const data = { type: 'config.update', cwd: '/tmp' };
       const entries = [{ lineNo: 1, data: data as any, raw: {} }];

apps/vis/server/test/lib/context-projector.test.ts

@@ -939,4 +939,28 @@ describe('context-projector', () => {
     expect(proj.messages[0]!.message.content[0]).toMatchObject({ text: bigText });
     expect(proj.messages[1]!.message.content[0]).toMatchObject({ text: bigText });
   });
+
+  it('does not project deferred turn bookkeeping before it enters context', () => {
+    const entries = [
+      {
+        lineNo: 1,
+        data: {
+          type: 'turn.defer' as const,
+          id: 'deferred-1',
+          input: [{ type: 'text' as const, text: 'pending' }],
+          origin: { kind: 'cron' as const },
+        },
+        raw: {},
+      },
+      {
+        lineNo: 2,
+        data: { type: 'turn.defer.consume' as const, id: 'deferred-1' },
+        raw: {},
+      },
+    ];
+
+    const proj = projectContext(entries as any);
+
+    expect(proj.messages).toEqual([]);
+  });
 });

apps/vis/web/src/components/wire/renderers.tsx

@@ -157,6 +157,56 @@ export const WIRE_RENDERERS: RendererMap = {
     ),
   },
 
+  'turn.defer': {
+    tone: 'warning',
+    label: 'defer',
+    headline: (r) => {
+      const text = firstText(r.input);
+      return {
+        main: (
+          <span className="flex items-center gap-2 min-w-0">
+            <Pill tone="turn" variant="soft">
+              {r.origin.kind}
+            </Pill>
+            <span className="truncate text-fg-1">pending → {truncate(text, 80)}</span>
+          </span>
+        ),
+        right: <Mono>#{r.id.slice(0, 8)}</Mono>,
+      };
+    },
+    detail: (r) => (
+      <div className="space-y-2">
+        <div className="grid grid-cols-[140px_1fr] gap-x-3 gap-y-[2px]">
+          <FieldRow label="id">
+            <Mono>{r.id}</Mono>
+          </FieldRow>
+          <FieldRow label="origin" wide>
+            <JsonViewer value={r.origin} defaultOpenDepth={2} />
+          </FieldRow>
+        </div>
+        <div>
+          <div className="mb-1 text-fg-2">
+            input ({r.input.length} part{r.input.length === 1 ? '' : 's'})
+          </div>
+          <div className="space-y-1">
+            {r.input.map((part, i) => (
+              <ContentPartView key={i} part={part} />
+            ))}
+          </div>
+        </div>
+      </div>
+    ),
+  },
+
+  'turn.defer.consume': {
+    tone: 'lifecycle',
+    label: 'defer·consume',
+    headline: (r) => ({
+      main: <Dim>deferred input entered context</Dim>,
+      right: <Mono>#{r.id.slice(0, 8)}</Mono>,
+    }),
+  },
+
   'turn.cancel': {
     tone: 'warning',
     label: 'cancel',

apps/vis/web/src/lib/analysis.ts

@@ -261,7 +261,13 @@ export function analyzeWire(entries: readonly WireEntry[]): Analysis {
           gapMs: t - prevTime,
           // A gap straddling a turn boundary is "waiting for the user"; a gap
           // inside a turn is the agent/tool being slow.
-          kind: rec.type === 'turn.prompt' || rec.type === 'turn.steer' ? 'between_turns' : 'in_turn',
+          kind:
+            rec.type === 'turn.prompt' ||
+            rec.type === 'turn.steer' ||
+            rec.type === 'turn.defer' ||
+            rec.type === 'turn.defer.consume'
+              ? 'between_turns'
+              : 'in_turn',
         });
       }
       prevTime = t;

apps/vis/web/test/analysis.test.ts

@@ -90,6 +90,31 @@ describe('analyzeWire', () => {
     expect(a.cache.hitRate).toBeNull();
   });
 
+  it('keeps deferred input bookkeeping outside turns and classifies its wait at the turn boundary', () => {
+    line = 0;
+    const a = analyzeWire([
+      e({ type: 'turn.prompt', input: [{ type: 'text', text: 'first' }], origin: { kind: 'user' } }, 0),
+      loop({ type: 'step.begin', uuid: 's1', turnId: 'T1', step: 0 }, 1),
+      loop({ type: 'step.end', uuid: 's1', turnId: 'T1', step: 0, finishReason: 'end_turn' }, 2),
+      e({
+        type: 'turn.defer',
+        id: 'deferred-1',
+        input: [{ type: 'text', text: 'pending' }],
+        origin: { kind: 'cron' },
+      }, 10_000),
+      e({ type: 'turn.defer.consume', id: 'deferred-1' }, 10_001),
+    ]);
+
+    expect(a.turns).toHaveLength(1);
+    expect(a.idleGaps).toEqual([
+      expect.objectContaining({
+        beforeLineNo: 4,
+        gapMs: 9998,
+        kind: 'between_turns',
+      }),
+    ]);
+  });
+
   it('computes cache hit rate from summed input usage', () => {
     line = 0;
     const a = analyzeWire([

apps/vis/web/test/renderers.test.ts

@@ -0,0 +1,30 @@
+import { describe, expect, it } from 'vitest';
+
+import { rendererFor } from '../src/components/wire/renderers';
+
+describe('wire renderers', () => {
+  it('renders both states of deferred turn bookkeeping', () => {
+    const deferred = rendererFor('turn.defer');
+    const consumed = rendererFor('turn.defer.consume');
+
+    expect(deferred).toMatchObject({ tone: 'warning', label: 'defer' });
+    expect(consumed).toMatchObject({
+      tone: 'lifecycle',
+      label: 'defer·consume',
+    });
+    expect(
+      deferred?.headline({
+        type: 'turn.defer',
+        id: 'deferred-1',
+        input: [{ type: 'text', text: 'pending' }],
+        origin: { kind: 'cron' },
+      }).main,
+    ).toBeDefined();
+    expect(
+      consumed?.headline({
+        type: 'turn.defer.consume',
+        id: 'deferred-1',
+      }).main,
+    ).toBeDefined();
+  });
+});