fix(chat): 同步普通聊天运行状态

src/client/api/chatTranscript.ts

@@ -5,6 +5,7 @@ import {
   CHAT_TRANSCRIPT_FEATURE,
   TranscriptGap,
   type TranscriptEvent,
+  type TranscriptControlEvent,
   type TranscriptMessage,
   type TranscriptSnapshot,
 } from '@shared/chatTranscript'
@@ -185,6 +186,33 @@ export class ChatTranscriptClient {
       await this.commit(next)
     })
   }
+  private async control(input: unknown) {
+    const value = input as Partial<TranscriptControlEvent>
+    const cursor = value.cursor
+    await this.serial(async () => {
+      const current = this.snapshot
+      if (
+        !current ||
+        value.epoch !== current.epoch ||
+        typeof cursor !== 'number' ||
+        !Number.isSafeInteger(cursor) ||
+        cursor < current.cursor ||
+        typeof value.running !== 'boolean' ||
+        typeof value.queued !== 'boolean'
+      ) throw new TranscriptGap()
+      const nextCursor = cursor
+      await this.commit({
+        ...current,
+        cursor: nextCursor,
+        running: value.running,
+        queued: value.queued,
+        pendingApproval: value.pendingApproval ?? null,
+        pendingClarification: value.pendingClarification ?? null,
+        liveStatus: value.liveStatus ?? null,
+        error: value.error ?? null,
+      })
+    })
+  }
   async run() {
     let needsSnapshot = !this.snapshot
     let restorePending = Boolean(this.snapshot)
@@ -225,6 +253,10 @@ export class ChatTranscriptClient {
           if (done) break
           for (const frame of parser.feed(decoder.decode(value, { stream: true }))) {
             if (frame.event === 'reset') throw new TranscriptGap()
+            if (frame.event === 'ready' || frame.event === 'state') {
+              await this.control(JSON.parse(frame.data))
+              continue
+            }
             if (frame.event === 'transcript') await this.receive(JSON.parse(frame.data))
           }
         }

src/server/chatTranscriptApi.ts

@@ -67,8 +67,13 @@ export function chatTranscriptRouter(cache:ChatCacheCoordinator,auth:LocalAuthSt
     res.writeHead(200,{'Content-Type':'text/event-stream; charset=utf-8','Cache-Control':'no-store','X-Accel-Buffering':'no'})
     res.flushHeaders();res.once('close',close);res.once('error',close)
     off=transcripts.subscribe(changed)
-    flush();write(`event: ready\ndata: ${JSON.stringify({epoch,cursor})}\n\n`)
-    heartbeat=setInterval(()=>{if(valid())write(': heartbeat\n\n');else close()},15_000);heartbeat.unref()
+    const control=()=>({epoch,cursor:transcripts.cursor(owner,profile,id),...transcripts.control(owner,profile,id)})
+    flush();write(`event: ready\ndata: ${JSON.stringify(control())}\n\n`)
+    heartbeat=setInterval(()=>{
+      if(!valid()){close();return}
+      if(transcripts.epochFor(owner,profile,id)!==epoch){write('event: reset\ndata: {}\n\n');close();return}
+      write(`event: state\ndata: ${JSON.stringify(control())}\n\n`)
+    },5_000);heartbeat.unref()
   })
   return router
 }

src/server/chatTranscriptStore.ts

@@ -3,6 +3,7 @@ import type { DatabaseSync } from 'node:sqlite'
 import {
   CHAT_TRANSCRIPT_FEATURE,
   type TranscriptEvent,
+  type TranscriptControlState,
   type TranscriptMessage,
   type TranscriptSnapshot,
 } from '../shared/chatTranscript.js'
@@ -32,7 +33,7 @@ interface TurnHead {
 }
 const emptyHead = (): TurnHead => ({ running: false, queued: false, queue: [], tools: {}, pending: {} })
 const text = (p: Data): string => String(p.output ?? p.text ?? p.content ?? '')
-const controlState = (head: TurnHead) => ({
+const controlState = (head: TurnHead): TranscriptControlState => ({
   running: head.running,
   queued: head.queued,
   pendingApproval: head.pendingApproval ?? null,
@@ -309,6 +310,9 @@ export class ChatTranscriptStore {
       .get(...scope)
     return row ? JSON.parse(String(row.data)) : emptyHead()
   }
+  control(...scope: Scope): TranscriptControlState {
+    return controlState(this.head(...scope))
+  }
   message(owner: string, profile: string, sessionId: string, id: string): TranscriptMessage | undefined {
     const row = this.db
       .prepare(
@@ -1051,7 +1055,7 @@ export class ChatTranscriptStore {
     before = Number.MAX_SAFE_INTEGER,
     limit = 150,
   ): TranscriptSnapshot {
-    const run = this.head(owner, profile, sessionId)
+    const run = this.control(owner, profile, sessionId)
     const coverage = this.db
       .prepare(
         'SELECT complete,message_total FROM chat_sessions WHERE owner=? AND profile=? AND session_id=?',

src/shared/chatTranscript.ts

@@ -57,6 +57,18 @@ export interface TranscriptSnapshot {
   liveStatus?: string | null
   error?: string | null
 }
+export interface TranscriptControlState {
+  running: boolean
+  queued: boolean
+  pendingApproval: Record<string, unknown> | null
+  pendingClarification: Record<string, unknown> | null
+  liveStatus: string | null
+  error: string | null
+}
+export interface TranscriptControlEvent extends TranscriptControlState {
+  epoch: string
+  cursor: number
+}
 export class TranscriptGap extends Error {
   constructor() {
     super('普通聊天消息版本不连续，需要重新同步')

tests/client/ordinaryTranscriptV2.test.ts

@@ -109,4 +109,42 @@ describe('ordinary v2 durable Web replica', () => {
       vi.unstubAllGlobals()
     }
   })
+  it('reconciles a stale running checkpoint from the authoritative SSE ready state', async () => {
+    const { client, changed } = setup()
+    const saved = { ...fixture.snapshot, running: true, queued: false, liveStatus: '正在思考' }
+    client.restore(saved)
+    const fetchMock = vi.fn((_url: unknown, options: RequestInit) => {
+      const stream = new ReadableStream<Uint8Array>({
+        start(controller) {
+          const frame = `event: ready\ndata: ${JSON.stringify({
+            epoch: 'fixture-epoch',
+            cursor: fixture.snapshot.cursor,
+            running: false,
+            queued: false,
+            pendingApproval: null,
+            pendingClarification: null,
+            liveStatus: null,
+            error: null,
+          })}\n\n`
+          controller.enqueue(new TextEncoder().encode(frame))
+          options.signal?.addEventListener('abort', () => controller.close(), { once: true })
+        },
+      })
+      return Promise.resolve(new Response(stream, { status: 200 }))
+    })
+    vi.stubGlobal('fetch', fetchMock)
+    const running = client.run()
+    try {
+      await vi.waitFor(() => expect(changed.mock.lastCall![0]).toMatchObject({
+        cursor: fixture.snapshot.cursor,
+        running: false,
+        queued: false,
+        liveStatus: null,
+      }))
+    } finally {
+      client.close()
+      await running
+      vi.unstubAllGlobals()
+    }
+  })
 })

tests/server/ordinaryTranscriptApi.test.ts

@@ -104,6 +104,19 @@ describe('ordinary v2 HTTP snapshot and durable SSE integration', () => {
     const reader = stream.body!.getReader(),
       parser = new SSEParser(),
       decoder = new TextDecoder()
+    const handshake = await reader.read()
+    expect(handshake.done).toBe(false)
+    const readyFrame = parser.feed(decoder.decode(handshake.value!, { stream: true }))
+      .find(frame => frame.event === 'ready')
+    expect(readyFrame).toBeTruthy()
+    expect(JSON.parse(readyFrame!.data)).toMatchObject({
+      epoch: base.epoch,
+      cursor: base.cursor,
+      running: false,
+      queued: false,
+      pendingApproval: null,
+      pendingClarification: null,
+    })
     let seq = 0
     for (const [type, payload] of [
       ['message.start', {}],