admin/Hermes-ui
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

revert: harden Hermes stream recovery around tool-call boundaries (#189) (#192)

Browse Source 
Reverts #189 due to reported bugs.

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
ekko
2026-04-24 22:18:32 +08:00
committed by GitHub
parent bff6f844e6
commit 70ed0e0dc2
6 changed files with 114 additions and 496 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/client/chat-api.test.ts
-59
View File
@@ -1,59 +0,0 @@
// @vitest-environment jsdom
import { beforeEach, describe, expect, it, vi } from 'vitest'
import { streamRunEvents, type RunEvent } from '@/api/hermes/chat'
class MockEventSource {
static instances: MockEventSource[] = []
url: string
onmessage: ((event: { data: string }) => void) | null = null
onerror: (() => void) | null = null
close = vi.fn()
constructor(url: string) {
this.url = url
MockEventSource.instances.push(this)
}
emit(data: string) {
this.onmessage?.({ data })
}
fail() {
this.onerror?.()
}
}
describe('streamRunEvents', () => {
beforeEach(() => {
window.localStorage.clear()
MockEventSource.instances = []
vi.stubGlobal('EventSource', MockEventSource)
})
it('maps non-JSON EventSource data to message.delta so raw text is rendered', () => {
const events: RunEvent[] = []
streamRunEvents('run-raw', event => events.push(event), vi.fn(), vi.fn())
MockEventSource.instances[0].emit('原因：raw fallback')
expect(events).toEqual([{ event: 'message.delta', delta: '原因：raw fallback' }])
})
it('parses colon-containing JSON deltas and closes on completion', () => {
const events: RunEvent[] = []
const onDone = vi.fn()
streamRunEvents('run-json', event => events.push(event), onDone, vi.fn())
const source = MockEventSource.instances[0]
source.emit(JSON.stringify({ event: 'message.delta', delta: '让我直接读文件：A: B' }))
source.emit(JSON.stringify({ event: 'run.completed' }))
expect(events).toEqual([
{ event: 'message.delta', delta: '让我直接读文件：A: B' },
{ event: 'run.completed' },
])
expect(source.close).toHaveBeenCalledTimes(1)
expect(onDone).toHaveBeenCalledTimes(1)
})
})
tests/client/chat-store.test.ts
+1 -206
View File
@@ -41,36 +41,13 @@ function makeSummary(id: string, title = 'Session') {
}
}
function makeDetail(id: string, messages: Array<Record<string, any>>, overrides: Record<string, any> = {}) {
function makeDetail(id: string, messages: Array<Record<string, any>>) {
return {
...makeSummary(id),
...overrides,
messages,
}
}
function makeHermesMessage(
id: number,
role: 'user' | 'assistant' | 'system' | 'tool',
content: string,
overrides: Record<string, any> = {},
) {
return {
id,
session_id: overrides.session_id || 'sess-1',
role,
content,
tool_call_id: null,
tool_calls: null,
tool_name: null,
timestamp: 1710000000 + id,
token_count: null,
finish_reason: null,
reasoning: null,
...overrides,
}
}
async function flushPromises() {
await Promise.resolve()
await Promise.resolve()
@@ -317,186 +294,4 @@ describe('Chat Store', () => {
expect(store.isRunActive).toBe(false)
expect(window.localStorage.getItem(inFlightKey('sess-1'))).toBeNull()
})
it('keeps colon deltas before and after a tool boundary', async () => {
mockChatApi.streamRunEvents.mockImplementation((
_runId: string,
onEvent: (event: any) => void,
) => {
onEvent({ event: 'message.delta', delta: '让我直接读文件：' })
onEvent({ event: 'tool.started', tool: 'read_file', preview: 'notes.md' })
onEvent({ event: 'tool.completed' })
onEvent({ event: 'message.delta', delta: '读取后结论: final' })
onEvent({ event: 'run.completed' })
return { abort: vi.fn() }
})
const store = useChatStore()
await flushPromises()
await store.sendMessage('check file')
await flushPromises()
const assistantText = store.messages
.filter(m => m.role === 'assistant')
.map(m => m.content)
.join('')
expect(assistantText).toBe('让我直接读文件：读取后结论: final')
expect(store.messages.some(m => m.role === 'tool' && m.toolName === 'read_file' && m.toolStatus === 'done')).toBe(true)
})
it('renders raw SSE fallback message events as assistant deltas', async () => {
mockChatApi.streamRunEvents.mockImplementation((
_runId: string,
onEvent: (event: any) => void,
) => {
onEvent({ event: 'message', delta: '原因：raw fallback' })
onEvent({ event: 'run.completed' })
return { abort: vi.fn() }
})
const store = useChatStore()
await flushPromises()
await store.sendMessage('raw stream')
await flushPromises()
expect(store.messages.some(m => m.role === 'assistant' && m.content === '原因：raw fallback')).toBe(true)
})
it('does not stop polling when server messages are stable but the session is still active', async () => {
vi.useFakeTimers()
let fetchSessionCalls = 0
mockSessionsApi.fetchSession.mockImplementation(async () => {
fetchSessionCalls += 1
if (fetchSessionCalls === 1) return null
return makeDetail('sess-1', [
makeHermesMessage(1, 'user', 'tool gap prompt'),
makeHermesMessage(2, 'assistant', '让我直接读文件：'),
], { ended_at: null })
})
mockChatApi.streamRunEvents.mockImplementation((
_runId: string,
onEvent: (event: any) => void,
_onDone: () => void,
onError: (err: Error) => void,
) => {
onEvent({ event: 'message.delta', delta: '让我直接读文件：' })
setTimeout(() => onError(new Error('SSE connection error')), 0)
return { abort: vi.fn() }
})
const store = useChatStore()
await flushPromises()
await store.sendMessage('tool gap prompt')
const sid = store.activeSessionId!
await vi.advanceTimersByTimeAsync(0)
await flushPromises()
await vi.advanceTimersByTimeAsync(9000)
await flushPromises()
expect(window.localStorage.getItem(inFlightKey(sid))).toBeTruthy()
expect(store.isRunActive).toBe(true)
})
it('reconciles the final session after run.completed to recover missed deltas', async () => {
let fetchSessionCalls = 0
mockSessionsApi.fetchSession.mockImplementation(async () => {
fetchSessionCalls += 1
if (fetchSessionCalls === 1) return null
return makeDetail('sess-1', [
makeHermesMessage(1, 'user', 'finish prompt'),
makeHermesMessage(2, 'assistant', '让我直接读文件：读取后结论：完整回答'),
], { ended_at: 1710000010 })
})
mockChatApi.streamRunEvents.mockImplementation((
_runId: string,
onEvent: (event: any) => void,
) => {
onEvent({ event: 'message.delta', delta: '让我直接读文件：' })
onEvent({ event: 'run.completed' })
return { abort: vi.fn() }
})
const store = useChatStore()
await flushPromises()
await store.sendMessage('finish prompt')
await flushPromises()
expect(store.messages.some(m => m.role === 'assistant' && m.content === '让我直接读文件：读取后结论：完整回答')).toBe(true)
})
it('does not replace longer local tool-boundary text with a stale shorter final fetch', async () => {
let fetchSessionCalls = 0
const stalePrefix = '让我直接读文件：较长的工具前说明'
mockSessionsApi.fetchSession.mockImplementation(async () => {
fetchSessionCalls += 1
if (fetchSessionCalls === 1) return null
return makeDetail('sess-1', [
makeHermesMessage(1, 'user', 'stale prompt'),
makeHermesMessage(2, 'assistant', stalePrefix),
], { ended_at: 1710000010 })
})
mockChatApi.streamRunEvents.mockImplementation((
_runId: string,
onEvent: (event: any) => void,
) => {
onEvent({ event: 'message.delta', delta: stalePrefix })
onEvent({ event: 'tool.started', tool: 'read_file', preview: 'notes.md' })
onEvent({ event: 'tool.completed' })
onEvent({ event: 'message.delta', delta: 'OK' })
onEvent({ event: 'run.completed' })
return { abort: vi.fn() }
})
const store = useChatStore()
await flushPromises()
await store.sendMessage('stale prompt')
await flushPromises()
const assistantText = store.messages
.filter(m => m.role === 'assistant')
.map(m => m.content)
.join('')
expect(assistantText).toBe(`${stalePrefix}OK`)
expect(store.messages.some(m => m.role === 'tool' && m.toolStatus === 'done')).toBe(true)
})
it('does not let delayed completion reconciliation clear a newer in-flight run', async () => {
let resolveReconcile: ((detail: any) => void) | null = null
const reconcilePromise = new Promise<any>(resolve => { resolveReconcile = resolve })
mockSessionsApi.fetchSession.mockImplementation(() => reconcilePromise)
mockChatApi.startRun
.mockResolvedValueOnce({ run_id: 'run-1', status: 'queued' })
.mockResolvedValueOnce({ run_id: 'run-2', status: 'queued' })
let firstRunEvent: ((event: any) => void) | null = null
mockChatApi.streamRunEvents.mockImplementation((
runId: string,
onEvent: (event: any) => void,
) => {
if (runId === 'run-1') firstRunEvent = onEvent
return { abort: vi.fn() }
})
const store = useChatStore()
await flushPromises()
await store.sendMessage('first')
firstRunEvent!({ event: 'run.completed' })
await flushPromises()
const sid = store.activeSessionId!
await store.sendMessage('second')
expect(JSON.parse(window.localStorage.getItem(inFlightKey(sid)) || '{}').runId).toBe('run-2')
resolveReconcile!(makeDetail(sid, [
makeHermesMessage(1, 'user', 'first', { session_id: sid }),
makeHermesMessage(2, 'assistant', 'first done', { session_id: sid }),
], { ended_at: 1710000010 }))
await flushPromises()
expect(JSON.parse(window.localStorage.getItem(inFlightKey(sid)) || '{}').runId).toBe('run-2')
})
})
tests/server/proxy-handler.test.ts
-85
View File
@@ -414,89 +414,4 @@ describe('SSE stream interception — run.completed', () => {
expect(mockUpdateUsage).toHaveBeenCalledWith('session-split', 200, 50)
})
it('forwards colon-containing SSE deltas around tool events unchanged and disables buffering', async () => {
const runId = 'run-colon-tool'
const sseData = [
`data: ${JSON.stringify({ event: 'message.delta', run_id: runId, delta: '让我直接读文件：A: B' })}\n\n`,
`data: ${JSON.stringify({ event: 'tool.started', run_id: runId, tool: 'read_file', preview: 'file:a.md' })}\n\n`,
`data: ${JSON.stringify({ event: 'tool.completed', run_id: runId })}\n\n`,
`data: ${JSON.stringify({ event: 'message.delta', run_id: runId, delta: '继续：done' })}\n\n`,
]
mockFetch.mockResolvedValue({
status: 200,
headers: new Headers({ 'content-type': 'text/event-stream' }),
body: createSSEBody(sseData),
})
const ctx = createMockCtx({
path: `/api/hermes/v1/runs/${runId}/events`,
search: '',
})
await proxy(ctx)
const forwarded = ctx.res.write.mock.calls
.map(([chunk]: [Uint8Array]) => new TextDecoder().decode(chunk))
.join('')
expect(forwarded).toBe(sseData.join(''))
expect(ctx.set).toHaveBeenCalledWith('Content-Type', 'text/event-stream')
expect(ctx.set).toHaveBeenCalledWith('Cache-Control', 'no-cache, no-transform')
expect(ctx.set).toHaveBeenCalledWith('X-Accel-Buffering', 'no')
})
it('intercepts run.completed with CRLF delimiters and data without a space', async () => {
const runId = 'run-crlf'
setRunSession(runId, 'session-crlf')
const completedJson = JSON.stringify({ event: 'run.completed', run_id: runId, usage: { input_tokens: 321, output_tokens: 45, total_tokens: 366 } })
const sseData = [`data:${completedJson}\r\n\r\n`]
mockFetch.mockResolvedValue({
status: 200,
headers: new Headers({ 'content-type': 'text/event-stream' }),
body: createSSEBody(sseData),
})
const ctx = createMockCtx({
path: `/api/hermes/v1/runs/${runId}/events`,
search: '',
})
await proxy(ctx)
expect(mockUpdateUsage).toHaveBeenCalledWith('session-crlf', 321, 45)
})
it('does not let usage accounting failures abort the SSE stream', async () => {
const runId = 'run-usage-fails'
setRunSession(runId, 'session-usage-fails')
mockUpdateUsage.mockImplementationOnce(() => {
throw new Error('usage db unavailable')
})
const sseData = [
`data: ${JSON.stringify({ event: 'message.delta', run_id: runId, delta: 'before：' })}\n\n`,
`data: ${JSON.stringify({ event: 'run.completed', run_id: runId, usage: { input_tokens: 1, output_tokens: 2, total_tokens: 3 } })}\n\n`,
]
mockFetch.mockResolvedValue({
status: 200,
headers: new Headers({ 'content-type': 'text/event-stream' }),
body: createSSEBody(sseData),
})
const ctx = createMockCtx({
path: `/api/hermes/v1/runs/${runId}/events`,
search: '',
})
await proxy(ctx)
const forwarded = ctx.res.write.mock.calls
.map(([chunk]: [Uint8Array]) => new TextDecoder().decode(chunk))
.join('')
expect(ctx.status).toBe(200)
expect(forwarded).toBe(sseData.join(''))
expect(ctx.res.end).toHaveBeenCalled()
})
})