From 7d4d916675200d890e52f19849856d30863ca35c Mon Sep 17 00:00:00 2001 From: YAN YIXIN Date: Thu, 5 Nov 2026 21:15:42 +0800 Subject: [PATCH] [bugfix] assistant: keep Agent runs alive across project navigation Switching projects aborted the chat request, and closing the SSE stream cancelled the Agent task, which then persisted as "stopped". Navigation and an explicit stop had the same effect. The run now belongs to the project rather than to the HTTP request: - Events of an accepted turn go to a per-run log instead of a bounded queue; closing a stream only detaches that viewer and never cancels the task. Only POST /turns/{id}/stop stops it. - GET /assistant reports activeTurnId while a turn is running, and GET /turns/{id}/events attaches a new stream that replays the turn from its start event (404 once it has ended). - The assistant hook keeps a still-running streaming reply instead of marking it interrupted, re-attaches on return, and falls back to the saved conversation if the turn finished in between. Leaving a project releases the view's turn state so the next project starts idle. Generation jobs submitted by the Agent were never tied to the stream and are unaffected. Fixes #81 Co-Authored-By: Claude Opus 5.5 --- apps/api/src/opendirector_api/agent.py | 81 +++++++++++++++------- src/backend/chat.ts | 41 ++++++++++- src/canvas/useAssistantChat.ts | 95 ++++++++++++++++++++------ tests/assistantResume.test.ts | 63 +++++++++++++++++ tests/test_assistant_conversations.py | 62 +++++++++++++++-- 5 files changed, 289 insertions(+), 53 deletions(-) create mode 100644 tests/assistantResume.test.ts diff --git a/apps/api/src/opendirector_api/agent.py b/apps/api/src/opendirector_api/agent.py index 03aad070..698983a4 100644 --- a/apps/api/src/opendirector_api/agent.py +++ b/apps/api/src/opendirector_api/agent.py @@ -149,6 +149,35 @@ class ConversationCreate(BaseModel): id: UUID +class RunEvents: + """Events of one accepted turn, kept until the turn ends. + + The run belongs to the project, not to the HTTP request that started it: + leaving the project closes the stream, and returning attaches a new one that + replays from the beginning. Only an explicit stop cancels the run. + """ + + def __init__(self): + self.items: list[dict] = [] + self.changed = asyncio.Event() + + def put(self, event: dict) -> None: + self.items.append(event) + self.changed.set() + self.changed = asyncio.Event() + + async def follow(self): + index = 0 + while True: + while index < len(self.items): + event = self.items[index] + index += 1 + yield 'data: ' + json.dumps(event, ensure_ascii=False) + '\n\n' + if event['type'] == 'end': + return + await self.changed.wait() + + def agent_routes(projects, generation, configs, provider=None) -> APIRouter: router = APIRouter(prefix='/api/creation/projects/{project_id}/assistant') conversations = Conversations(projects) @@ -174,12 +203,18 @@ def trace(project_id: UUID, turn_id: UUID, raise ProjectError('轨迹文件不存在。', 404) return FileResponse(file, media_type='text/plain', headers={'Cache-Control': 'no-store'}) - active: dict[UUID, tuple[UUID, asyncio.Task]] = {} + active: dict[UUID, tuple[UUID, asyncio.Task, RunEvents]] = {} admission: dict[UUID, asyncio.Lock] = {} @router.get('') async def conversation(project_id: UUID): - return await asyncio.to_thread(conversations.update, project_id) + value = await asyncio.to_thread(conversations.update, project_id) + running = active.get(project_id) + # A streaming message without an active turn was interrupted (for + # example by a restart); with one, the client can attach to it. + if running and not running[1].done(): + value = {**value, 'activeTurnId': str(running[0])} + return value @router.get('/conversations') async def list_conversations(project_id: UUID): @@ -244,6 +279,14 @@ async def stop(project_id: UUID, turn_id: UUID, x_opendirector_session: UUID = H await running[1] return {'ok': True} + @router.get('/turns/{turn_id}/events') + async def turn_events(project_id: UUID, turn_id: UUID): + await asyncio.to_thread(conversations.path, project_id) + running = active.get(project_id) + if not running or running[0] != turn_id: + raise ProjectError('这轮回复已结束，请重新读取对话。', 404) + return StreamingResponse(running[2].follow(), media_type='text/event-stream') + @router.post('/turns') async def turn(project_id: UUID, value: Turn, x_opendirector_session: UUID = Header(), accept_language: str | None = Header(default=None, alias='Accept-Language')): @@ -273,7 +316,7 @@ async def turn(project_id: UUID, value: Turn, x_opendirector_session: UUID = Hea model_names = {row['id']: row['modelId'] for row in configs.list()} preferences = {kind: {'modelConfigId': model_id, 'modelName': model_names.get(model_id)} for kind, model_id in model_selections.items() if model_id} - queue: asyncio.Queue = asyncio.Queue(maxsize=128) + events = RunEvents() async def produce() -> None: text, status, error, accepted = '', 'complete', None, False activities = [] @@ -288,7 +331,7 @@ async def produce() -> None: 'model': model, 'status': 'complete'} saved = await asyncio.to_thread(conversations.update, project_id, session, begin=user, conversation_id=value.conversationId) accepted = True - await queue.put({'type': 'start', 'conversation': saved}) + events.put({'type': 'start', 'conversation': saved}) context = [] for message in saved['messages'][:-1]: content = message['text'] or '(empty reply)' @@ -326,7 +369,7 @@ async def produce() -> None: else: activities[activities.index(previous)] = activity event = {'type': 'tool', **activity} - await queue.put(event) + events.put(event) except asyncio.CancelledError: status = 'stopped' raise @@ -355,33 +398,19 @@ async def produce() -> None: **({'plan': plan} if plan is not None else {})}, conversation_id=value.conversationId) else: saved = None - if asyncio.current_task().cancelling(): - while not queue.empty(): - queue.get_nowait() - await queue.put({'type': 'end', 'conversation': saved, 'error': error}) + events.put({'type': 'end', 'conversation': saved, 'error': error}) if project_id in active: raise ProjectError('当前工程已有回复正在执行。', 409) task = asyncio.create_task(produce()) - active[project_id] = (value.id, task) + active[project_id] = (value.id, task, events) def release(completed: asyncio.Task) -> None: - if active.get(project_id) == (value.id, completed): + running = active.get(project_id) + if running and running[0] == value.id and running[1] is completed: active.pop(project_id) task.add_done_callback(release) - async def events(): - try: - while True: - event = await queue.get() - yield 'data: ' + json.dumps(event, ensure_ascii=False) + '\n\n' - if event['type'] == 'end': - break - finally: - # A disconnected SSE request must not cancel persistence during cleanup. - with CancelScope(shield=True): - if not task.done() and not task.cancelling(): - task.cancel() - with suppress(asyncio.CancelledError): - await task - return StreamingResponse(events(), media_type='text/event-stream') + # Closing this stream (switching projects, reloading) only detaches + # the viewer; the run continues and can be attached again. + return StreamingResponse(events.follow(), media_type='text/event-stream') return router diff --git a/src/backend/chat.ts b/src/backend/chat.ts index 372d8c2d..5f7de42d 100644 --- a/src/backend/chat.ts +++ b/src/backend/chat.ts @@ -154,7 +154,15 @@ export async function chatRuntimes(id: string, onUpdate?: (items: RuntimeInfo[]) })); return rows; } -export const conversation = (id: string) => request(path(id)); +/** The saved conversation, plus the turn still running for it, if any. */ +export async function conversation( + id: string, +): Promise<{ conversation: Conversation; activeTurnId: string | null }> { + const { activeTurnId, ...saved } = await request< + Conversation & { activeTurnId?: string } + >(path(id)); + return { conversation: saved, activeTurnId: activeTurnId ?? null }; +} export const saveChatDraft = (id: string, draft: ChatDraft) => request(path(id) + "/draft", json("PUT", draft)); export const chatSkills = (id: string, signal?: AbortSignal) => @@ -173,6 +181,37 @@ export async function streamChat( headers: { ...headers, "Content-Type": "application/json" }, signal, }); + await readChatEvents(response, onEvent, t); +} + +/** Thrown when the turn already ended; read the conversation instead. */ +export class ChatTurnEnded extends Error {} + +/** + * Attach to a turn that is still running, for example after returning to the + * project. The server replays the turn from its start event; closing this + * stream only detaches the viewer and never stops the Agent. + */ +export async function attachChat( + id: string, + turn: string, + signal: AbortSignal, + onEvent: (event: ChatEvent) => void, + t: Translate = defaultTranslate, +) { + const response = await fetch("/api" + path(id) + `/turns/${turn}/events`, { + headers, + signal, + }); + if (response.status === 404) throw new ChatTurnEnded(); + await readChatEvents(response, onEvent, t); +} + +async function readChatEvents( + response: Response, + onEvent: (event: ChatEvent) => void, + t: Translate, +) { if (!response.ok) throw await responseError(response); if (!response.body) throw new Error(t("assistant.stream.unavailable")); const reader = response.body.getReader(); diff --git a/src/canvas/useAssistantChat.ts b/src/canvas/useAssistantChat.ts index a2837380..1d674fc8 100644 --- a/src/canvas/useAssistantChat.ts +++ b/src/canvas/useAssistantChat.ts @@ -2,6 +2,9 @@ import { useCallback, useEffect, useLayoutEffect, useRef, useState } from "react import { rememberAgentSelection, restoreAgentSelection } from "./agentPreferences"; import { errorMessage } from "../backend/api"; import { + attachChat, + ChatTurnEnded, + type ChatEvent, conversation, conversations, activateConversation, @@ -51,10 +54,15 @@ const empty = (): Conversation => ({ modelSelections: {}, }); -function restoreConversation(projectId: string, saved: Conversation): Conversation { +/** A streaming reply is interrupted unless its turn is still running on the server. */ +function restoreConversation( + projectId: string, + saved: Conversation, + running = false, +): Conversation { return { ...restoreAgentSelection(projectId, saved), - messages: saved.messages.map(message => message.status === "streaming" + messages: saved.messages.map(message => message.status === "streaming" && !running ? { ...message, status: "interrupted" } : message), }; } @@ -92,6 +100,7 @@ export function useAssistantChat( const submitting = useRef(false); const uploading = useRef(false); const turnId = useRef(null); + const resume = useRef<(turn: string) => Promise>(async () => {}); const current = useRef(value); current.current = value; // Retain metadata for editor undo after a reference is removed from the draft. @@ -171,10 +180,12 @@ export function useAssistantChat( }); void refreshConversations(); void conversation(projectId) - .then((saved) => { + .then(({ conversation: saved, activeTurnId }) => { if (cancelled) return; - setValue(restoreConversation(projectId, saved)); + setValue(restoreConversation(projectId, saved, Boolean(activeTurnId))); setLoaded(true); + // Navigation only detached the previous view; pick the run back up. + if (activeTurnId) void resume.current(activeTurnId); }) .catch((reason) => { if (!cancelled) @@ -185,7 +196,16 @@ export function useAssistantChat( return () => { cancelled = true; alive.current = false; + // Closing the stream detaches this view only; the server keeps the + // turn running. Release the turn now so the next project starts idle. controller.current?.abort(); + controller.current = null; + submitting.current = false; + turnId.current = null; + setSending(false); + setStarted(false); + setStopping(false); + setThinkingNow(false); uploadController.current?.abort(); listController.current?.abort(); }; @@ -398,18 +418,8 @@ export function useAssistantChat( !current.current.draft.trim() ) return; - submitting.current = true; - setSending(true); - setStarted(false); - setError(""); - setThinkingNow(false); - const abort = new AbortController(); - controller.current = abort; - const id = crypto.randomUUID(); - turnId.current = id; const input = current.current; - const assistantId = crypto.randomUUID(); - try { + await follow(crypto.randomUUID(), async (id, signal, onEvent) => { await flushChatDraft(projectId); await streamChat( projectId, @@ -426,6 +436,47 @@ export function useAssistantChat( modelSelections: input.modelSelections, skills: input.skills ?? [], }, + signal, + onEvent, + t, + ); + }); + }; + /** Re-attach to a turn that kept running while this project was not shown. */ + resume.current = async (id: string) => { + if (switching.current || submitting.current) return; + await follow(id, async (turn, signal, onEvent) => { + try { + await attachChat(projectId, turn, signal, onEvent, t); + } catch (reason) { + if (!(reason instanceof ChatTurnEnded)) throw reason; + // It finished in between: the saved conversation has the final reply. + const { conversation: saved } = await conversation(projectId); + if (alive.current && !signal.aborted) + setValue(restoreConversation(projectId, saved)); + } + }); + }; + const follow = async ( + id: string, + run: ( + id: string, + signal: AbortSignal, + onEvent: (event: ChatEvent) => void, + ) => Promise, + ) => { + submitting.current = true; + setSending(true); + setStarted(false); + setError(""); + setThinkingNow(false); + const abort = new AbortController(); + controller.current = abort; + turnId.current = id; + const assistantId = crypto.randomUUID(); + try { + await run( + id, abort.signal, (event) => { if (!alive.current) return; @@ -479,16 +530,20 @@ export function useAssistantChat( if (event.error) setError(displayError(event.error)); } }, - t, ); } catch (reason) { if (alive.current && !abort.signal.aborted) setError(displayError(errorMessage(reason))); } finally { - submitting.current = false; - turnId.current = null; - controller.current = null; - if (alive.current) { + // A newer view may already own these refs after a project switch. + if (controller.current === abort) { + submitting.current = false; + turnId.current = null; + controller.current = null; + } + if (alive.current && !abort.signal.aborted) { + // Aborting only detaches this view (project switch, unmount); the + // turn keeps running on the server and is resumed on return. setSending(false); setStarted(false); setStopping(false); diff --git a/tests/assistantResume.test.ts b/tests/assistantResume.test.ts new file mode 100644 index 00000000..b5279434 --- /dev/null +++ b/tests/assistantResume.test.ts @@ -0,0 +1,63 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { + attachChat, + ChatTurnEnded, + conversation, + type ChatEvent, +} from "../src/backend/chat.ts"; + +const sse = (events: unknown[]) => + new Response( + events.map((event) => `data: ${JSON.stringify(event)}\n\n`).join(""), + { headers: { "Content-Type": "text/event-stream" } }, + ); + +test("returning to a project replays the running turn from its start", async (t) => { + const saved = { messages: [], draft: "", assets: [], thinking: "auto", modelSelections: {} }; + const fetch = t.mock.method(globalThis, "fetch", async (url: string) => { + assert.ok(url.endsWith("/assistant/turns/turn-1/events")); + return sse([ + { type: "start", conversation: saved }, + { type: "delta", text: "继续" }, + { type: "end", conversation: saved, error: null }, + ]); + }); + const events: ChatEvent[] = []; + await attachChat("project", "turn-1", new AbortController().signal, (event) => + events.push(event), + ); + assert.deepEqual( + events.map((event) => event.type), + ["start", "delta", "end"], + ); + assert.equal(fetch.mock.calls[0]!.arguments[1]?.method, undefined); +}); + +test("a turn that already ended is reported so the saved conversation is read", async (t) => { + t.mock.method(globalThis, "fetch", async () => + Response.json({ detail: "ended", code: "not_found" }, { status: 404 }), + ); + await assert.rejects( + attachChat("project", "turn-1", new AbortController().signal, () => {}), + ChatTurnEnded, + ); +}); + +test("the running turn id stays out of the conversation that drafts are saved from", async (t) => { + t.mock.method(globalThis, "fetch", async () => + Response.json({ + conversationId: "c", + messages: [{ id: "m", role: "assistant", text: "", status: "streaming" }], + draft: "", + assets: [], + thinking: "auto", + modelSelections: {}, + activeTurnId: "turn-1", + }), + ); + const result = await conversation("project"); + assert.equal(result.activeTurnId, "turn-1"); + assert.equal("activeTurnId" in result.conversation, false); + assert.equal(result.conversation.messages[0]!.status, "streaming"); +}); diff --git a/tests/test_assistant_conversations.py b/tests/test_assistant_conversations.py index ca75fb57..111913c3 100644 --- a/tests/test_assistant_conversations.py +++ b/tests/test_assistant_conversations.py @@ -205,15 +205,69 @@ async def finished(self, run): assert frames(reply)[-1]['conversation']['messages'][-1]['text'] == '新回复' -def test_sse_disconnect_saves_partial_reply_and_releases_project(assistant, monkeypatch): +def test_sse_disconnect_keeps_the_run_and_a_new_stream_resumes_it(assistant, monkeypatch): + """#81: leaving the project closes the stream but must not stop the Agent.""" + client, path, _ = assistant + release, attached = threading.Event(), threading.Event() + async def stream(self, run): + yield {'type': 'delta', 'text': '断线前的回复'} + await asyncio.to_thread(release.wait, 10) + yield {'type': 'delta', 'text': '，返回后完成'} + monkeypatch.setattr(agent.AgentRuntimes, 'stream', stream) + follow, calls = agent.RunEvents.follow, [] + async def counted(self): + calls.append(self) + if len(calls) > 1: + attached.set() + async for item in follow(self): + yield item + monkeypatch.setattr(agent.RunEvents, 'follow', counted) + turn_id = str(uuid4()) + disconnect_turn(client, path, turn_id) + + current = client.get(path).json() + assert current['activeTurnId'] == turn_id + assert current['messages'][-1]['status'] == 'streaming' + assert client.post(path + '/conversations', json={'id': str(uuid4())}).status_code == 409 + with ThreadPoolExecutor() as pool: + pending = pool.submit(client.get, path + f'/turns/{turn_id}/events') + try: + assert attached.wait(5) + finally: + release.set() + events = frames(pending.result(timeout=10)) + assert events[0]['type'] == 'start' + assert [item['text'] for item in events if item['type'] == 'delta'] == ['断线前的回复', '，返回后完成'] + saved = events[-1]['conversation']['messages'][-1] + assert events[-1]['type'] == 'end' and events[-1]['error'] is None + assert saved['status'] == 'complete' and saved['text'] == '断线前的回复，返回后完成' + assert 'activeTurnId' not in client.get(path).json() + assert client.get(path + f'/turns/{turn_id}/events').status_code == 404 + assert client.post(path + '/conversations', json={'id': str(uuid4())}).status_code == 200 + + +def test_explicit_stop_after_disconnect_still_stops_the_run(assistant, monkeypatch): client, path, _ = assistant async def stream(self, run): yield {'type': 'delta', 'text': '断线前的回复'} await asyncio.Event().wait() monkeypatch.setattr(agent.AgentRuntimes, 'stream', stream) + turn_id = str(uuid4()) + disconnect_turn(client, path, turn_id) + assert client.get(path).json()['activeTurnId'] == turn_id + assert client.post(path + f'/turns/{turn_id}/stop').status_code == 200 + current = client.get(path).json() + assert 'activeTurnId' not in current + assert current['messages'][-1]['status'] == 'stopped' + assert current['messages'][-1]['text'] == '断线前的回复' + assert client.post(path + '/conversations', json={'id': str(uuid4())}).status_code == 200 + + +def disconnect_turn(client, path, turn_id): + """Start a turn over raw ASGI and close the stream after its first delta.""" async def disconnect(): disconnected = asyncio.Event() - body = json.dumps({'id': str(uuid4()), 'runtime': 'default', 'text': '断线测试'}).encode() + body = json.dumps({'id': turn_id, 'runtime': 'default', 'text': '断线测试'}).encode() requested = False async def receive(): nonlocal requested @@ -235,7 +289,3 @@ async def send_event(message): await asyncio.wait_for(client.app(scope, receive, send_event), 5) assert disconnected.is_set() client.portal.call(disconnect) - saved = client.get(path).json()['messages'][-1] - assert saved['status'] == 'stopped' - assert saved['text'] == '断线前的回复' - assert client.post(path + '/conversations', json={'id': str(uuid4())}).status_code == 200