fix(agent): 恢复桌面端执行轨迹
This commit is contained in:
+6
-1
@@ -62,7 +62,12 @@ app = FastAPI(
|
|||||||
|
|
||||||
app.add_middleware(
|
app.add_middleware(
|
||||||
CORSMiddleware,
|
CORSMiddleware,
|
||||||
allow_origins=["http://127.0.0.1:5173", "http://localhost:5173"],
|
allow_origins=[
|
||||||
|
"http://127.0.0.1:5173",
|
||||||
|
"http://localhost:5173",
|
||||||
|
"http://tauri.localhost",
|
||||||
|
"tauri://localhost",
|
||||||
|
],
|
||||||
allow_credentials=True,
|
allow_credentials=True,
|
||||||
allow_methods=["*"],
|
allow_methods=["*"],
|
||||||
allow_headers=["*"],
|
allow_headers=["*"],
|
||||||
|
|||||||
@@ -77,6 +77,18 @@ def test_health() -> None:
|
|||||||
assert response.model_dump() == {"status": "ok"}
|
assert response.model_dump() == {"status": "ok"}
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("origin", ["http://tauri.localhost", "tauri://localhost"])
|
||||||
|
def test_desktop_origins_can_read_streaming_api(origin: str) -> None:
|
||||||
|
from fastapi.testclient import TestClient
|
||||||
|
|
||||||
|
from app.main import app
|
||||||
|
|
||||||
|
response = TestClient(app).get("/health", headers={"Origin": origin})
|
||||||
|
|
||||||
|
assert response.status_code == 200
|
||||||
|
assert response.headers["access-control-allow-origin"] == origin
|
||||||
|
|
||||||
|
|
||||||
def test_mcp_create_and_trust_are_not_executed_on_event_loop(monkeypatch) -> None:
|
def test_mcp_create_and_trust_are_not_executed_on_event_loop(monkeypatch) -> None:
|
||||||
from app import routes
|
from app import routes
|
||||||
from app.contracts import McpServerCreateRequest, McpServerTrustRequest
|
from app.contracts import McpServerCreateRequest, McpServerTrustRequest
|
||||||
|
|||||||
@@ -50,8 +50,7 @@ export const useAgentStore = defineStore('agent', () => {
|
|||||||
)
|
)
|
||||||
|
|
||||||
const currentStep = computed(() => {
|
const currentStep = computed(() => {
|
||||||
const tc = events.value.filter((e) => e.event === 'ToolCall').length
|
return activeRun.value?.current_step ?? 0
|
||||||
return tc
|
|
||||||
})
|
})
|
||||||
|
|
||||||
async function loadTools() {
|
async function loadTools() {
|
||||||
@@ -79,6 +78,7 @@ export const useAgentStore = defineStore('agent', () => {
|
|||||||
async function loadRun(runId: string) {
|
async function loadRun(runId: string) {
|
||||||
const version = ++selectionVersion
|
const version = ++selectionVersion
|
||||||
stopStream()
|
stopStream()
|
||||||
|
error.value = null
|
||||||
activeRunId.value = runId
|
activeRunId.value = runId
|
||||||
resetEvents()
|
resetEvents()
|
||||||
permissionRequest.value = null
|
permissionRequest.value = null
|
||||||
@@ -91,9 +91,38 @@ export const useAgentStore = defineStore('agent', () => {
|
|||||||
events.value = []
|
events.value = []
|
||||||
toolCalls.value = []
|
toolCalls.value = []
|
||||||
permissionRequest.value = null
|
permissionRequest.value = null
|
||||||
|
try {
|
||||||
|
await loadPersistedTrace(runId, () => version === selectionVersion && activeRunId.value === runId)
|
||||||
|
error.value = null
|
||||||
|
} catch (cause) {
|
||||||
|
if (version !== selectionVersion || activeRunId.value !== runId) return
|
||||||
|
error.value = cause instanceof Error ? cause.message : String(cause)
|
||||||
|
}
|
||||||
|
if (version !== selectionVersion || activeRunId.value !== runId) return
|
||||||
|
if (terminal(activeRun.value?.status)) {
|
||||||
|
connectionState.value = 'idle'
|
||||||
|
return
|
||||||
|
}
|
||||||
subscribe(runId)
|
subscribe(runId)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async function loadPersistedTrace(runId: string, current: () => boolean) {
|
||||||
|
let cursor = lastSequence
|
||||||
|
do {
|
||||||
|
const trace = await agentService.getAgentTrace(runId, {
|
||||||
|
after_sequence: cursor,
|
||||||
|
limit: 500,
|
||||||
|
})
|
||||||
|
if (!current()) return
|
||||||
|
trace.items.forEach(processEvent)
|
||||||
|
const run = runs.value.find((item) => item.run_id === runId)
|
||||||
|
if (run) run.status = trace.status
|
||||||
|
if (!trace.has_more) return
|
||||||
|
if (trace.next_sequence <= cursor) throw new Error(t('运行轨迹分页游标未前进', 'Agent trace cursor did not advance'))
|
||||||
|
cursor = trace.next_sequence
|
||||||
|
} while (current())
|
||||||
|
}
|
||||||
|
|
||||||
function processEvent(event: AgentEvent) {
|
function processEvent(event: AgentEvent) {
|
||||||
// 服务端会先回放历史再发送实时事件,以 run_id + sequence 去重保证幂等。
|
// 服务端会先回放历史再发送实时事件,以 run_id + sequence 去重保证幂等。
|
||||||
if (seenSequences.has(event.sequence)) return
|
if (seenSequences.has(event.sequence)) return
|
||||||
@@ -107,7 +136,10 @@ export const useAgentStore = defineStore('agent', () => {
|
|||||||
const data = event.data
|
const data = event.data
|
||||||
const run = runs.value.find((item) => item.run_id === event.run_id)
|
const run = runs.value.find((item) => item.run_id === event.run_id)
|
||||||
if (event.event === 'RunStarted' && run) run.status = 'running'
|
if (event.event === 'RunStarted' && run) run.status = 'running'
|
||||||
if (event.event === 'ToolCall') {
|
if (event.event === 'ModelCallStarted' && run) {
|
||||||
|
const step = Number(data.step)
|
||||||
|
if (Number.isFinite(step)) run.current_step = Math.max(run.current_step, step)
|
||||||
|
} else if (event.event === 'ToolCall') {
|
||||||
toolCalls.value.push({
|
toolCalls.value.push({
|
||||||
tool_call_id: String(data.tool_call_id ?? ''),
|
tool_call_id: String(data.tool_call_id ?? ''),
|
||||||
name: String(data.name ?? 'unknown'),
|
name: String(data.name ?? 'unknown'),
|
||||||
@@ -168,11 +200,17 @@ export const useAgentStore = defineStore('agent', () => {
|
|||||||
retryTimer = setTimeout(async () => {
|
retryTimer = setTimeout(async () => {
|
||||||
retryTimer = null
|
retryTimer = null
|
||||||
try {
|
try {
|
||||||
|
await loadPersistedTrace(runId, current)
|
||||||
|
if (!current()) return
|
||||||
const run = await agentService.getAgentRun(runId)
|
const run = await agentService.getAgentRun(runId)
|
||||||
if (!current()) return
|
if (!current()) return
|
||||||
const index = runs.value.findIndex(item => item.run_id === runId)
|
const index = runs.value.findIndex(item => item.run_id === runId)
|
||||||
if (index >= 0) runs.value[index] = run
|
if (index >= 0) runs.value[index] = run
|
||||||
// 即使已结束仍续读一次缺失的尾部事件,保留完整 Trace。
|
if (terminal(run.status)) {
|
||||||
|
isRunning.value = false
|
||||||
|
connectionState.value = 'idle'
|
||||||
|
return
|
||||||
|
}
|
||||||
subscribe(runId)
|
subscribe(runId)
|
||||||
} catch (cause) {
|
} catch (cause) {
|
||||||
if (current()) interrupted(cause instanceof Error ? cause : new Error(String(cause)))
|
if (current()) interrupted(cause instanceof Error ? cause : new Error(String(cause)))
|
||||||
|
|||||||
@@ -2,10 +2,18 @@
|
|||||||
import { afterEach, beforeEach, expect, it, vi } from 'vitest'
|
import { afterEach, beforeEach, expect, it, vi } from 'vitest'
|
||||||
import { createPinia, setActivePinia, disposePinia } from 'pinia'
|
import { createPinia, setActivePinia, disposePinia } from 'pinia'
|
||||||
import { useAgentStore } from './agent'
|
import { useAgentStore } from './agent'
|
||||||
const mock = vi.hoisted(() => ({ stream: vi.fn(), get: vi.fn(), cancel: vi.fn() }))
|
const mock = vi.hoisted(() => ({ stream: vi.fn(), get: vi.fn(), trace: vi.fn(), cancel: vi.fn() }))
|
||||||
vi.mock('@/services/agentService', () => ({ streamAgentEvents: mock.stream, getAgentRun: mock.get, cancelAgentRun: mock.cancel }))
|
vi.mock('@/services/agentService', () => ({ streamAgentEvents: mock.stream, getAgentRun: mock.get, getAgentTrace: mock.trace, cancelAgentRun: mock.cancel }))
|
||||||
let pinia: ReturnType<typeof createPinia>
|
let pinia: ReturnType<typeof createPinia>
|
||||||
beforeEach(() => { vi.useFakeTimers(); vi.clearAllMocks(); pinia = createPinia(); setActivePinia(pinia); mock.stream.mockReturnValue({ cancel: vi.fn() }); mock.get.mockImplementation(async id => ({ run_id: id, status: 'running' })) })
|
beforeEach(() => {
|
||||||
|
vi.useFakeTimers()
|
||||||
|
vi.clearAllMocks()
|
||||||
|
pinia = createPinia()
|
||||||
|
setActivePinia(pinia)
|
||||||
|
mock.stream.mockReturnValue({ cancel: vi.fn() })
|
||||||
|
mock.get.mockImplementation(async id => ({ run_id: id, status: 'running', current_step: 0, max_steps: 6 }))
|
||||||
|
mock.trace.mockImplementation(async id => ({ run_id: id, status: 'running', items: [], next_sequence: -1, has_more: false }))
|
||||||
|
})
|
||||||
afterEach(() => { disposePinia(pinia); vi.useRealTimers() })
|
afterEach(() => { disposePinia(pinia); vi.useRealTimers() })
|
||||||
const handler = () => mock.stream.mock.calls.at(-1)![1]
|
const handler = () => mock.stream.mock.calls.at(-1)![1]
|
||||||
const event = (sequence: number, name = 'RunStarted') => ({ run_id: 'r', sequence, event: name, data: {}, timestamp: 'now' })
|
const event = (sequence: number, name = 'RunStarted') => ({ run_id: 'r', sequence, event: name, data: {}, timestamp: 'now' })
|
||||||
@@ -41,3 +49,41 @@ it('bounds retry attempts and supports explicit retry without losing events', as
|
|||||||
store.reconnect()
|
store.reconnect()
|
||||||
expect(mock.stream.mock.calls.at(-1)![2]).toBe(1)
|
expect(mock.stream.mock.calls.at(-1)![2]).toBe(1)
|
||||||
})
|
})
|
||||||
|
|
||||||
|
it('loads every persisted trace page for a completed run without opening SSE', async () => {
|
||||||
|
mock.get.mockResolvedValue({ run_id: 'r', status: 'completed', current_step: 2, max_steps: 6 })
|
||||||
|
mock.trace
|
||||||
|
.mockResolvedValueOnce({
|
||||||
|
run_id: 'r', status: 'completed',
|
||||||
|
items: [event(0), event(1, 'ModelCallStarted')], next_sequence: 1, has_more: true,
|
||||||
|
})
|
||||||
|
.mockResolvedValueOnce({
|
||||||
|
run_id: 'r', status: 'completed',
|
||||||
|
items: [event(2, 'ToolCall'), event(3, 'RunCompleted')], next_sequence: 3, has_more: false,
|
||||||
|
})
|
||||||
|
|
||||||
|
const store = useAgentStore()
|
||||||
|
await store.loadRun('r')
|
||||||
|
|
||||||
|
expect(mock.trace).toHaveBeenNthCalledWith(1, 'r', { after_sequence: -1, limit: 500 })
|
||||||
|
expect(mock.trace).toHaveBeenNthCalledWith(2, 'r', { after_sequence: 1, limit: 500 })
|
||||||
|
expect(store.events.map(item => item.sequence)).toEqual([0, 1, 2, 3])
|
||||||
|
expect(store.currentStep).toBe(2)
|
||||||
|
expect(mock.stream).not.toHaveBeenCalled()
|
||||||
|
})
|
||||||
|
|
||||||
|
it('fills missing events through REST before reconnecting a live run', async () => {
|
||||||
|
const store = useAgentStore()
|
||||||
|
await store.loadRun('r')
|
||||||
|
handler().onEvent(event(0))
|
||||||
|
handler().onError(new Error('desktop SSE unavailable'))
|
||||||
|
mock.trace.mockResolvedValueOnce({
|
||||||
|
run_id: 'r', status: 'running',
|
||||||
|
items: [event(1, 'ModelCallStarted'), event(2, 'ToolCall')], next_sequence: 2, has_more: false,
|
||||||
|
})
|
||||||
|
|
||||||
|
await vi.advanceTimersByTimeAsync(1000)
|
||||||
|
|
||||||
|
expect(store.events.map(item => item.sequence)).toEqual([0, 1, 2])
|
||||||
|
expect(mock.stream.mock.calls.at(-1)![2]).toBe(2)
|
||||||
|
})
|
||||||
|
|||||||
Reference in New Issue
Block a user