feat: implement PI core chat timeline

electron/coding-runtime/conversation-service.ts

@@ -1,4 +1,5 @@
 import type {
+  CodingConversationPatchBatchEvent,
   CodingConversationRuntime,
   CodingRuntimeCommand,
   CodingRuntimeDiagnostics,
@@ -49,6 +50,9 @@ export interface CodingConversationServiceOptions {
     projectId: string;
     sessionKey: string;
   }): Promise<void>;
+  deliveryBatchWindowMs?: number;
+  deliveryBatchMaxItems?: number;
+  deliveryBatchMaxBytes?: number;
 }
 
 export type CodingConversationStreamEvent =
@@ -59,7 +63,7 @@ export type CodingConversationStreamEvent =
     seq: number;
     snapshot: ConversationSnapshot;
   }
-  | ({ type: 'patch' } & ConversationPatchEnvelope);
+  | CodingConversationPatchBatchEvent;
 
 export interface CodingConversationEventStream {
   snapshots: ConversationSnapshot[];
@@ -160,6 +164,100 @@ class PatchQueue implements AsyncIterable<CodingConversationStreamEvent> {
   }
 }
 
+interface PendingPatchBatch {
+  conversationId: string;
+  workerGeneration: number;
+  items: CodingConversationPatchBatchEvent['items'];
+  byteLength: number;
+  timer: ReturnType<typeof setTimeout>;
+}
+
+class PatchBatchScheduler {
+  private readonly pending = new Map<string, PendingPatchBatch>();
+  private readonly windowMs: number;
+  private readonly maxItems: number;
+  private readonly maxBytes: number;
+  private closed = false;
+
+  constructor(
+    private readonly publish: (event: CodingConversationPatchBatchEvent) => void,
+    options: CodingConversationServiceOptions,
+  ) {
+    this.windowMs = Math.min(33, Math.max(16, options.deliveryBatchWindowMs ?? 24));
+    this.maxItems = Math.max(1, options.deliveryBatchMaxItems ?? 128);
+    this.maxBytes = Math.max(1, options.deliveryBatchMaxBytes ?? 256 * 1024);
+  }
+
+  push(envelope: ConversationPatchEnvelope): void {
+    if (this.closed) return;
+    const key = this.key(envelope.conversationId, envelope.workerGeneration);
+    const item = {
+      ...(envelope.runId ? { runId: envelope.runId } : {}),
+      seq: envelope.seq,
+      at: envelope.at,
+      patch: envelope.patch,
+    };
+    const itemBytes = Buffer.byteLength(JSON.stringify(item), 'utf8');
+    let batch = this.pending.get(key);
+    if (batch && batch.items.length > 0
+      && (batch.items.length >= this.maxItems || batch.byteLength + itemBytes > this.maxBytes)) {
+      this.flush(key);
+      batch = undefined;
+    }
+    if (!batch) {
+      batch = {
+        conversationId: envelope.conversationId,
+        workerGeneration: envelope.workerGeneration,
+        items: [],
+        byteLength: 0,
+        timer: setTimeout(() => this.flush(key), this.windowMs),
+      };
+      this.pending.set(key, batch);
+    }
+    batch.items.push(item);
+    batch.byteLength += itemBytes;
+    if (batch.items.length >= this.maxItems || batch.byteLength >= this.maxBytes) {
+      this.flush(key);
+    }
+  }
+
+  cancel(conversationId: string): void {
+    for (const [key, batch] of this.pending) {
+      if (batch.conversationId !== conversationId) continue;
+      clearTimeout(batch.timer);
+      this.pending.delete(key);
+    }
+  }
+
+  close(): void {
+    this.closed = true;
+    for (const batch of this.pending.values()) clearTimeout(batch.timer);
+    this.pending.clear();
+  }
+
+  private key(conversationId: string, workerGeneration: number): string {
+    return `${conversationId}\u0000${workerGeneration}`;
+  }
+
+  private flush(key: string): void {
+    const batch = this.pending.get(key);
+    if (!batch || this.closed) return;
+    this.pending.delete(key);
+    clearTimeout(batch.timer);
+    const first = batch.items[0];
+    const last = batch.items.at(-1);
+    if (!first || !last) return;
+    this.publish({
+      type: 'patch-batch',
+      conversationId: batch.conversationId,
+      workerGeneration: batch.workerGeneration,
+      fromSeq: first.seq,
+      toSeq: last.seq,
+      items: batch.items,
+    });
+  }
+}
+
 export class CodingConversationService {
   private readonly prepareFlights = new Map<string, Promise<PrepareConversationInput>>();
   private readonly acceptances = new Map<string, AcceptanceRecord>();
@@ -466,6 +564,7 @@ export class CodingConversationService {
       : requiredString(conversationId, 'Conversation id', 128);
     if (id) await this.projects.findActiveConversation(id);
     const queue = new PatchQueue();
+    const batches = new PatchBatchScheduler((event) => queue.push(event), this.options);
     const cursors = new Map<string, ConversationSnapshot['cursor']>();
     let initialize!: () => void;
     const initialized = new Promise<void>((resolve) => { initialize = resolve; });
@@ -479,6 +578,7 @@ export class CodingConversationService {
         if (!cursor
           || event.workerGeneration !== cursor.workerGeneration
           || event.seq !== cursor.seq + 1) {
+          batches.cancel(event.conversationId);
           try {
             const snapshot = await this.runtime.getSnapshot(event.conversationId);
             cursors.set(event.conversationId, snapshot.cursor);
@@ -498,7 +598,7 @@ export class CodingConversationService {
           workerGeneration: event.workerGeneration,
           seq: event.seq,
         });
-        queue.push({ type: 'patch', ...event });
+        batches.push(event);
       }).catch(() => undefined);
     });
     try {
@@ -516,12 +616,14 @@ export class CodingConversationService {
         events: queue,
         close: () => {
           unsubscribe();
+          batches.close();
           queue.close();
         },
       };
     } catch (error) {
       initialize();
       unsubscribe();
+      batches.close();
       queue.close();
       throw error;
     }

electron/coding-runtime/pi/session-registry.ts

@@ -60,6 +60,7 @@ export class PiSessionRegistry {
   private readonly createConversationStore: typeof createCodingConversationStore;
   private readonly records = new Map<string, RegistryRecord>();
   private readonly prepareFlights = new Map<string, Promise<RegistryRecord>>();
+  private readonly stores = new Map<string, ReturnType<typeof createCodingConversationStore>>();
 
   constructor(options: PiSessionRegistryOptions) {
     this.projectStore = options.projectStore;
@@ -126,7 +127,11 @@ export class PiSessionRegistry {
       candidate.id === input.agentId && candidate.enabled && !candidate.archivedAt
     ));
     if (!agent) throw new Error('Coding Agent does not exist');
-    const store = this.createConversationStore(project.path);
+    let store = this.stores.get(project.path);
+    if (!store) {
+      store = this.createConversationStore(project.path);
+      this.stores.set(project.path, store);
+    }
     const conversation = await store.get(input.conversationId);
     if (!conversation || conversation.agentId !== input.agentId) {
       throw new Error('Coding Conversation does not exist for the selected Agent');