修复实时命令过期去重与持久回执恢复

docs/http-sse-realtime.md

@@ -23,7 +23,9 @@ Chat frames retain their upstream seq separately from the opaque SSE `id`. Send
 
 `state` is `confirmed`, `rejected`, `pending`, or `unknown`. Only an actual upstream RPC acknowledgement is confirmed; receiving HTTP 200 alone is not a submission acknowledgement. Identical keys reuse the original operation; different payloads under one key conflict. Network retry must retain the key and original encoded command, not rerun attachment preparation. The iOS Outbox ID is the prompt key.
 
-`realtime-receipts.sqlite3` records write-ahead fingerprints and small control acknowledgements, not prompts, attachments, transcript content or full event streams. Terminal records retain 24 hours. Pending operations become unknown on process restart. Unknown receipts are never automatically re-executed or evicted to permit retries; capacity exhaustion rejects new admissions. Full RPC results have a separate bounded in-memory cache; after eviction/restart only the small retained control receipt remains, so clients must refresh resource data rather than assume a history snapshot was persisted.
+`realtime-receipts.sqlite3` records write-ahead fingerprints and bounded recovery metadata, not prompts, attachment bytes, transcript content or full event streams. Allowed attachment commands retain their file references; allowed configuration commands retain their applied value and model confirmation metadata. Terminal results retain 24 hours, then compact into permanent owner/request/fingerprint tombstones that return `unknown`. Compaction never makes an admitted request executable again, including after a restart; changed payloads still conflict. Tombstones contain no command bodies or RPC results and grow with unique retired admissions. Pending operations become unknown on process restart. Unknown receipts are never automatically re-executed or evicted to permit retries; capacity exhaustion rejects new admissions. Full RPC results have a separate bounded in-memory cache, so clients must refresh resource data rather than assume a history snapshot was persisted.
+
+Clients must persist each stable command ID as attempted **before** its POST. An absent receipt is permission to submit only when a durable local record proves that command was never attempted. For an attempted or legacy uncertain prompt, reconcile the exact `client_message_id` against the canonical transcript; absence from a recent page is not proof of non-execution. Keep unresolved sends uncertain, and never submit an attachment label without a recovered file reference.
 
 ## Resource limits and HTTP-only clients
 

src/server/realtimeBroker.ts

@@ -161,7 +161,7 @@ export class RealtimeBroker {
     if (existing) return this.commands.get(key) ?? Promise.resolve(this.completed.get(key)?.receipt ?? existing)
     const work = this.perform(c, frame, requestId).then(response => {
       const state = response.error ? 'rejected' : 'confirmed'
-      this.receipts.finish(c.principal.key, requestId, state, response)
+      this.receipts.finish(c.principal.key, requestId, state, response, frame)
       return { requestId, state, response } as CommandReceipt
     }, (error: unknown) => {
       if (error instanceof HttpError) {

src/server/realtimeReceipts.ts

@@ -10,6 +10,8 @@ export interface CommandReceipt {
   response?: Record<string, unknown>
 }
 
+interface ReceiptCommand { method?: unknown; params?: Record<string, unknown> }
+
 /** Write-ahead admission, not an event/transcript store. Never stores command bodies. */
 export class RealtimeReceipts {
   private db: DatabaseSync
@@ -24,36 +26,63 @@ export class RealtimeReceipts {
         state TEXT NOT NULL, response TEXT, updated INTEGER NOT NULL,
         PRIMARY KEY(owner,id));
       CREATE INDEX IF NOT EXISTS receipt_retention ON receipts(state,updated);
+      CREATE TABLE IF NOT EXISTS receipt_tombstones (
+        owner TEXT NOT NULL, id TEXT NOT NULL, fingerprint TEXT NOT NULL,
+        PRIMARY KEY(owner,id));
       UPDATE receipts SET state='unknown' WHERE state='pending';`)
   }
   lookup(owner: string, id: string): CommandReceipt | undefined {
     const row = this.db.prepare('SELECT state,response FROM receipts WHERE owner=? AND id=?').get(owner, id)
-    if (!row) return undefined
+    if (!row) return this.db.prepare('SELECT 1 FROM receipt_tombstones WHERE owner=? AND id=?').get(owner, id)
+      ? { requestId: id, state: 'unknown' } : undefined
     return { requestId: id, state: row.state as CommandReceipt['state'],
       ...(row.response ? { response: JSON.parse(String(row.response)) as Record<string, unknown> } : {}) }
   }
   reserve(owner: string, id: string, frame: string): CommandReceipt | undefined {
     if (!/^[A-Za-z0-9:_-]{1,200}$/.test(id)) throw new HttpError(400, 'Invalid request ID', 'invalid_request_id')
     const fingerprint = createHash('sha256').update(frame).digest('hex')
-    const row = this.db.prepare('SELECT fingerprint FROM receipts WHERE owner=? AND id=?').get(owner, id)
+    const row = this.db.prepare(`SELECT fingerprint FROM receipts WHERE owner=? AND id=?
+      UNION ALL SELECT fingerprint FROM receipt_tombstones WHERE owner=? AND id=?`).get(owner, id, owner, id)
     if (row) {
       if (row.fingerprint !== fingerprint) throw new HttpError(409, 'Request ID payload conflict', 'idempotency_conflict')
       return this.lookup(owner, id)
     }
-    this.db.prepare("DELETE FROM receipts WHERE updated<? AND state IN ('confirmed','rejected')").run(Date.now() - 86_400_000)
-    if (Number(this.db.prepare('SELECT count(*) AS n FROM receipts').get()!.n) >= 100_000) {
-      throw new HttpError(429, 'Command receipt capacity reached', 'receipt_capacity')
+    // Retire results, never admission identities. Old clients may retry years later.
+    // A compact fingerprint tombstone prevents those retries from executing again.
+    this.db.exec('BEGIN IMMEDIATE')
+    try {
+      const cutoff = Date.now() - 86_400_000
+      this.db.prepare(`INSERT OR IGNORE INTO receipt_tombstones
+        SELECT owner,id,fingerprint FROM receipts WHERE updated<? AND state IN ('confirmed','rejected')`).run(cutoff)
+      this.db.prepare("DELETE FROM receipts WHERE updated<? AND state IN ('confirmed','rejected')").run(cutoff)
+      if (Number(this.db.prepare('SELECT count(*) AS n FROM receipts').get()!.n) >= 100_000) {
+        throw new HttpError(429, 'Command receipt capacity reached', 'receipt_capacity')
+      }
+      this.db.prepare('INSERT INTO receipts VALUES(?,?,?,?,NULL,?)').run(owner, id, fingerprint, 'pending', Date.now())
+      this.db.exec('COMMIT')
+    } catch (error) {
+      this.db.exec('ROLLBACK')
+      throw error
     }
-    this.db.prepare('INSERT INTO receipts VALUES(?,?,?,?,NULL,?)').run(owner, id, fingerprint, 'pending', Date.now())
     return undefined
   }
-  finish(owner: string, id: string, state: CommandReceipt['state'], response?: Record<string, unknown>): void {
-    // Persist only control-plane acknowledgements, never prompt, attachment or history content.
+  finish(owner: string, id: string, state: CommandReceipt['state'], response?: Record<string, unknown>, command?: ReceiptCommand): void {
+    // Persist bounded recovery metadata, never command bodies, attachment bytes or history.
     const result = response?.result as Record<string, unknown> | undefined
     const safe: Record<string, unknown> = {}
     for (const key of ['status', 'session_id', 'stored_session_id', 'session_key', 'queued', 'running', 'ok']) {
       if (result && ['string', 'number', 'boolean'].includes(typeof result[key])) safe[key] = result[key]
     }
+    const strings = command?.method === 'config.set' && ['model', 'reasoning', 'fast'].includes(String(command.params?.key))
+      ? ['value', 'confirm_message', 'warning']
+      : ['file.attach', 'image.attach_bytes', 'pdf.attach'].includes(String(command?.method))
+        ? ['path', 'image_path', 'ref_path', 'ref_text'] : []
+    for (const key of strings) {
+      if (typeof result?.[key] === 'string' && result[key].length <= 4096) safe[key] = result[key]
+    }
+    if (command?.method === 'config.set') for (const key of ['deferred', 'confirm_required']) {
+      if (typeof result?.[key] === 'boolean') safe[key] = result[key]
+    }
     const saved = response?.error ? { error: { message: 'Upstream rejected command' } }
       : response ? { result: safe } : undefined
     this.db.prepare('UPDATE receipts SET state=?,response=?,updated=? WHERE owner=? AND id=?')

tests/server/realtimeReceipts.test.ts

@@ -0,0 +1,92 @@
+// @vitest-environment node
+import { afterEach, beforeEach, expect, it, vi } from 'vitest'
+import { mkdtempSync, readFileSync, rmSync } from 'node:fs'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { RealtimeReceipts } from '../../src/server/realtimeReceipts'
+import { RealtimeBroker, type RealtimePrincipal } from '../../src/server/realtimeBroker'
+
+let home: string, receipts: RealtimeReceipts
+beforeEach(() => {
+  home = mkdtempSync(join(tmpdir(), 'yaoyao-receipt-recovery-'))
+  receipts = new RealtimeReceipts(home)
+})
+afterEach(() => { receipts.close(); vi.restoreAllMocks(); rmSync(home, { recursive: true, force: true }) })
+
+it('retains owner-scoped deduplication after result expiry and database reopen', () => {
+  const now = vi.spyOn(Date, 'now').mockReturnValue(100_000)
+  receipts.reserve('alice', 'prompt-1', 'private prompt')
+  receipts.finish('alice', 'prompt-1', 'confirmed', { result: { status: 'submitted' } })
+  now.mockReturnValue(100_000 + 86_400_001)
+  receipts.reserve('alice', 'new-command', '{}')
+  receipts.close(); receipts = new RealtimeReceipts(home)
+  expect(receipts.lookup('alice', 'prompt-1')).toEqual({ requestId: 'prompt-1', state: 'unknown' })
+  expect(receipts.reserve('alice', 'prompt-1', 'private prompt')).toMatchObject({ state: 'unknown' })
+  expect(() => receipts.reserve('alice', 'prompt-1', 'changed payload')).toThrow('payload conflict')
+  expect(receipts.lookup('bob', 'prompt-1')).toBeUndefined()
+  expect(receipts.reserve('bob', 'prompt-1', 'private prompt')).toBeUndefined()
+  for (const file of ['realtime-receipts.sqlite3', 'realtime-receipts.sqlite3-wal']) {
+    expect(readFileSync(join(home, file)).includes(Buffer.from('private prompt'))).toBe(false)
+  }
+})
+
+it('does not call the executor a second time after expired results are compacted', async () => {
+  receipts.close()
+  const now = vi.spyOn(Date, 'now').mockReturnValue(100_000)
+  const broker = new RealtimeBroker(home)
+  const perform = vi.fn(async () => ({ result: { status: 'submitted' } }))
+  Object.assign(broker, { perform })
+  const principal: RealtimePrincipal = { key: 'alice', upstreamKey: 'mock', paired: false,
+    valid: () => true, url: async () => { throw new Error('No network allowed') } }
+  try {
+    const channel = { kind: 'chat', principal } as Parameters<RealtimeBroker['command']>[0]
+    const frame = { method: 'prompt.submit', params: { session_id: 'runtime', text: 'do work' } }
+    await broker.command(channel, 'prompt-1', frame)
+    now.mockReturnValue(100_000 + 86_400_001)
+    await broker.command(channel, 'new-prompt', frame)
+    // Simulate losing the process-local full response, as on a restart.
+    Object.assign(broker, { completed: new Map() })
+    expect(await broker.command(channel, 'prompt-1', frame)).toMatchObject({ state: 'unknown' })
+    expect(perform).toHaveBeenCalledTimes(2)
+  } finally {
+    broker.close()
+    await Promise.resolve()
+    receipts = new RealtimeReceipts(home)
+  }
+})
+
+it.each(['file.attach', 'image.attach_bytes', 'pdf.attach'])('restores %s references without persisting attachment contents', method => {
+  const id = method.replaceAll('.', '_')
+  receipts.reserve('alice', id, '{}')
+  receipts.finish('alice', id, 'confirmed', { result: {
+    path: '/files/upload', image_path: '/images/upload', ref_path: 'upload', ref_text: '@file:upload',
+    content: 'private bytes', content_base64: 'private bytes', text: 'private bytes',
+  } }, { method })
+  receipts.close(); receipts = new RealtimeReceipts(home)
+  expect(receipts.lookup('alice', id)?.response?.result).toEqual({
+    path: '/files/upload', image_path: '/images/upload', ref_path: 'upload', ref_text: '@file:upload',
+  })
+})
+
+it('restores model confirmation metadata only for allowed configuration keys', () => {
+  receipts.reserve('alice', 'model', '{}')
+  receipts.finish('alice', 'model', 'confirmed', { result: {
+    value: 'model --provider provider', deferred: false, confirm_required: true,
+    confirm_message: 'Please confirm', warning: 'Cost warning', api_key: 'private key',
+  } }, { method: 'config.set', params: { key: 'model' } })
+  receipts.close(); receipts = new RealtimeReceipts(home)
+  expect(receipts.lookup('alice', 'model')?.response?.result).toEqual({
+    value: 'model --provider provider', deferred: false, confirm_required: true,
+    confirm_message: 'Please confirm', warning: 'Cost warning',
+  })
+  receipts.reserve('alice', 'unrelated', '{}')
+  receipts.finish('alice', 'unrelated', 'confirmed', { result: { value: 'private key', path: '/private' } },
+    { method: 'config.set', params: { key: 'api_key' } })
+  expect(receipts.lookup('alice', 'unrelated')?.response?.result).toEqual({})
+})
+
+it('does not persist unbounded recovery strings', () => {
+  receipts.reserve('alice', 'large', '{}')
+  receipts.finish('alice', 'large', 'confirmed', { result: { path: 'x'.repeat(4097) } }, { method: 'file.attach' })
+  expect(receipts.lookup('alice', 'large')?.response?.result).toEqual({})
+})