补充 OpenViking 跨客户端同步冲突与部分追加回归用例

tests/server/openVikingSessionSync.test.ts

@@ -14,6 +14,7 @@ import type { WorkspaceAgent, WorkspaceConversation, WorkspaceMessage } from '..
 let home: string, store: WorkspaceStore, service: OpenVikingService, sync: OpenVikingSessionSync, agent: WorkspaceAgent, conversation: WorkspaceConversation
 const remote = new Map<string, any[]>(), clients: any[] = []
 let failAfterWrite = false, offline = false, autoCommit = false
+let partialWrite: number | undefined, silentPartial = false
 const notFound = () => new OpenVikingError('not found', { code: 'NOT_FOUND', statusCode: 404 })
 const config = { enabled: true, url: 'http://localhost:1933', accountId: 'test', adminKey: 'admin-secret' }
 beforeEach(async () => {
@@ -28,7 +29,7 @@ beforeEach(async () => {
       createSession: vi.fn(async ({ sessionId }: any) => { if (!remote.has(prefix + sessionId)) remote.set(prefix + sessionId, []); return { session_id: sessionId } }),
       getSession: vi.fn(async (id: string) => { if (offline) throw new Error('offline'); if (!remote.has(prefix + id)) throw notFound(); return { uri: id, message_count: remote.get(prefix + id)!.length, auto_commit_policy: autoCommit ? { message_count_threshold: 1 } : null } }),
       read: vi.fn(async (uri: string) => remote.get(prefix + uri.replace('/messages.jsonl', ''))!.map(row => JSON.stringify({ role: row.role, parts: [{ type: 'text', text: row.content }], created_at: row.createdAt })).join('\n')),
-      batchAddMessages: vi.fn(async (id: string, messages: any[]) => { remote.get(prefix + id)!.push(...messages); if (failAfterWrite) { failAfterWrite = false; throw new Error('response lost') } return {} }),
+      batchAddMessages: vi.fn(async (id: string, messages: any[]) => { remote.get(prefix + id)!.push(...messages.slice(0, partialWrite ?? messages.length)); if (partialWrite !== undefined && !silentPartial) { partialWrite = undefined; throw new Error('partial write') } partialWrite = undefined; if (failAfterWrite) { failAfterWrite = false; throw new Error('response lost') } return {} }),
       deleteSession: vi.fn(async (id: string) => { remote.delete(prefix + id) }),
       commitSession: vi.fn(),
     }
@@ -39,7 +40,7 @@ beforeEach(async () => {
   conversation = store.list<WorkspaceConversation>('owner', 'conversation')[0]!
   sync = new OpenVikingSessionSync(store, service)
 })
-afterEach(() => { sync.close(); store.close(); rmSync(home, { recursive: true, force: true }); remote.clear(); clients.length = 0; offline = false; failAfterWrite = false; autoCommit = false; vi.restoreAllMocks() })
+afterEach(() => { sync.close(); store.close(); rmSync(home, { recursive: true, force: true }); remote.clear(); clients.length = 0; offline = false; failAfterWrite = false; autoCommit = false; partialWrite = undefined; silentPartial = false; vi.restoreAllMocks() })
 
 function message(role: 'user' | 'assistant' | 'system', content: string, fields: Partial<WorkspaceMessage> = {}) {
   const value: WorkspaceMessage = { id: randomUUID(), conversationId: conversation.id, seq: 0, role, ...(role === 'assistant' ? { agentId: agent.id } : {}), content, reasoning: 'private thought', status: 'complete', attachments: [], tools: [{ secret: 'private tool' }], createdAt: Date.now(), ...fields }
@@ -185,3 +186,120 @@ it('mirrors Bot communication bodies with sender and receiver roles, without int
   await drain()
   expect([...remote.values()].flat().map(row => [row.role, row.content]).sort()).toEqual([['assistant', 'collaboration body'], ['user', 'collaboration body']])
 })
+
+
+it('catches messages from another client while an append is in flight', async () => {
+  message('user', 'client A')
+  await drain()
+  message('user', 'client B')
+  const client = clients.find(row => row.batchAddMessages.mock.calls.length)!
+  const append = client.batchAddMessages.getMockImplementation()!
+  client.batchAddMessages.mockImplementationOnce(async (id: string, batch: any[]) => {
+    const result = await append(id, batch)
+    message('user', 'client A while B appends')
+    return result
+  })
+  await drain()
+  expect(texts()).toEqual(['client A', 'client B', 'client A while B appends'])
+  expect(sync.status()).toMatchObject({ complete: 1, pending: 0, failed: 0 })
+})
+
+it('serializes overlapping ticks from different sync instances sharing the same database', async () => {
+  message('user', 'one copy')
+  const other = new OpenVikingSessionSync(store, service)
+  try {
+    await Promise.all([sync.tick(), other.tick()])
+    await drain()
+    expect(texts()).toEqual(['one copy'])
+    expect(sync.status().complete).toBe(1)
+  } finally { other.close() }
+})
+
+it('recovers a partial failed append after restart without repeating the accepted prefix', async () => {
+  message('user', 'first'); message('assistant', 'second'); message('user', 'third')
+  partialWrite = 1
+  await sync.tick()
+  expect(texts()).toEqual(['first'])
+  sync.close(); sync = new OpenVikingSessionSync(store, service)
+  await drain()
+  expect(texts()).toEqual(['first', 'second', 'third'])
+  expect(sync.status()).toMatchObject({ complete: 1, failed: 0 })
+})
+
+it('verifies writes even when a server acknowledges a partial batch as successful', async () => {
+  message('user', 'first'); message('assistant', 'second'); message('user', 'third')
+  partialWrite = 1; silentPartial = true
+  await drain()
+  expect(texts()).toEqual(['first', 'second', 'third'])
+  expect(sync.status()).toMatchObject({ complete: 1, failed: 0 })
+})
+
+it('detects a local revision received during the remote read before appending a suffix', async () => {
+  const original = message('user', 'original'); await drain()
+  message('assistant', 'must not append')
+  const client = clients.find(row => row.batchAddMessages.mock.calls.length)!
+  const read = client.read.getMockImplementation()!
+  client.read.mockImplementationOnce(async (uri: string) => {
+    const raw = await read(uri)
+    store.saveMessage('owner', { ...original, content: 'changed on another client' })
+    return raw
+  })
+  await drain()
+  expect(texts()).toEqual(['original'])
+  expect(sync.status()).toMatchObject({ failed: 1 })
+  expect(sync.status().lastError).toContain('已停止追加')
+})
+
+it('recovers a legacy failed job only after verifying the remote prefix on startup', async () => {
+  message('user', 'prefix'); await drain()
+  message('assistant', 'suffix')
+  store.db.prepare("UPDATE openviking_sync_jobs SET status='failed',attempts=3,error=?").run('OpenViking 会话记录与本地顺序不一致，已停止追加，请核对会话')
+  sync.close(); sync = new OpenVikingSessionSync(store, service)
+  await drain()
+  expect(texts()).toEqual(['prefix', 'suffix'])
+  expect(sync.status()).toMatchObject({ failed: 0, complete: 1 })
+})
+
+it.each(['reordered', 'changed', 'extra'])('keeps a real %s remote conflict closed across restart and explicit retry', async kind => {
+  message('user', 'first', { createdAt: 1000 }); message('assistant', 'second', { createdAt: 2000 })
+  await drain()
+  const rows = [...remote.values()][0]!
+  if (kind === 'reordered') rows.reverse()
+  if (kind === 'changed') rows[0] = { ...rows[0], content: 'foreign revision' }
+  if (kind === 'extra') rows.push({ role: 'user', content: 'foreign message', createdAt: new Date(3000).toISOString() })
+  const before = JSON.stringify(rows)
+  message('user', 'new suffix', { createdAt: 4000 })
+  await drain()
+  expect(sync.status().failed).toBe(1)
+  sync.close(); sync = new OpenVikingSessionSync(store, service); await drain()
+  sync.retry(); await drain()
+  expect(JSON.stringify(rows)).toBe(before)
+  expect(sync.status().failed).toBe(1)
+  expect(clients.every(client => client.deleteSession.mock.calls.length === 0)).toBe(true)
+})
+
+it('does not deduplicate distinct local messages with identical text and timestamps', async () => {
+  message('user', 'repeat', { createdAt: 1000 }); message('user', 'repeat', { createdAt: 1000 })
+  failAfterWrite = true; await drain()
+  expect(texts()).toEqual(['repeat', 'repeat'])
+  expect(sync.status().complete).toBe(1)
+})
+
+it('uses local sequence rather than client clocks and keeps it stable after reconnect', async () => {
+  const first = message('user', 'client A first', { createdAt: 2000 })
+  const second = message('assistant', 'client B second', { createdAt: 1000 })
+  await drain()
+  store.saveMessage('owner', { ...first, seq: 11 }); store.saveMessage('owner', { ...second, seq: 12 })
+  sync.close(); sync = new OpenVikingSessionSync(store, service); await drain()
+  expect(texts()).toEqual(['client A first', 'client B second'])
+  expect(sync.status().complete).toBe(1)
+})
+
+it('rejects remote history loss instead of appending an already acknowledged transcript again', async () => {
+  message('user', 'acknowledged'); await drain()
+  const rows = [...remote.values()][0]!
+  rows.length = 0
+  message('user', 'later'); await drain()
+  expect(rows).toEqual([])
+  expect(sync.status().failed).toBe(1)
+})