diff --git a/.ai-project/state.yaml b/.ai-project/state.yaml index a1708ad..f220323 100644 --- a/.ai-project/state.yaml +++ b/.ai-project/state.yaml @@ -38,10 +38,10 @@ 状态: "已完成" - 编号: "F-005" 名称: "停止、失败与重试" - 状态: "进行中" + 状态: "已完成" - 编号: "F-006" 名称: "附件、文件列表、读取与搜索" - 状态: "待开始" + 状态: "进行中" - 编号: "F-007" 名称: "文件创建与安全修改" 状态: "待开始" diff --git a/apps/web-server/src/composition/run-registry.ts b/apps/web-server/src/composition/run-registry.ts index 031a092..1daa1e2 100644 --- a/apps/web-server/src/composition/run-registry.ts +++ b/apps/web-server/src/composition/run-registry.ts @@ -38,6 +38,26 @@ export class RunRegistry { return this.paused.has(runId); } + async waitForTerminal(runId: string, timeoutMs = 2_000): Promise { + if (this.isFinished(runId)) return; + await new Promise((resolve) => { + const finish = () => { + clearTimeout(timer); + unsubscribe(); + resolve(); + }; + const unsubscribe = this.subscribe(runId, async (event) => { + if ( + event.type === "run.completed" || + event.type === "run.failed" || + event.type === "run.cancelled" + ) + finish(); + }); + const timer = setTimeout(finish, timeoutMs); + }); + } + subscribe(runId: string, listener: Listener): () => void { const values = this.listeners.get(runId) ?? new Set(); values.add(listener); diff --git a/apps/web-server/src/http/error-handler.ts b/apps/web-server/src/http/error-handler.ts index d1af787..a7b35e2 100644 --- a/apps/web-server/src/http/error-handler.ts +++ b/apps/web-server/src/http/error-handler.ts @@ -26,7 +26,8 @@ export function createErrorHandler(logger: Logger): ErrorHandler { error.code === "INTERACTION_NOT_FOUND" ? 404 : error.code === "PROJECT_RUN_ACTIVE" || - error.code === "INTERACTION_ALREADY_RESOLVED" + error.code === "INTERACTION_ALREADY_RESOLVED" || + error.code === "RUN_ALREADY_ACTIVE" ? 409 : 400; return context.json( diff --git a/apps/web-server/src/index.ts b/apps/web-server/src/index.ts index 5412e9d..851e231 100644 --- a/apps/web-server/src/index.ts +++ b/apps/web-server/src/index.ts @@ -56,6 +56,7 @@ const model: ModelPort = environment.deepSeekApiKey apiKey: environment.deepSeekApiKey, baseURL: environment.deepSeekBaseUrl, model: environment.deepSeekModel, + timeoutMs: environment.modelTimeoutMs, }) : { async *stream() { @@ -97,6 +98,10 @@ logger.info( const server = Bun.serve({ hostname: environment.host, port: environment.port, + idleTimeout: Math.min( + 255, + Math.ceil(environment.modelTimeoutMs / 1_000) + 15, + ), fetch: app.fetch, }); diff --git a/apps/web-server/src/routes/runs.test.ts b/apps/web-server/src/routes/runs.test.ts index a23a962..a6cb10e 100644 --- a/apps/web-server/src/routes/runs.test.ts +++ b/apps/web-server/src/routes/runs.test.ts @@ -60,6 +60,12 @@ class Runs implements RunRepository { async getById(id: string) { return this.value?.id === id ? this.value : null; } + async findActive() { + return this.value?.status === "running" || + this.value?.status === "waiting_user" + ? this.value + : null; + } async appendEvent(event: RunEvent) { this.events.push(event); } diff --git a/apps/web-server/src/routes/runs.ts b/apps/web-server/src/routes/runs.ts index 524700e..51bafe0 100644 --- a/apps/web-server/src/routes/runs.ts +++ b/apps/web-server/src/routes/runs.ts @@ -9,6 +9,9 @@ export function createRunRoutes( registry: RunRegistry, ): Hono { const routes = new Hono(); + routes.get("/runs/:runId", async (context) => + context.json(await service.getRun(context.req.param("runId"))), + ); routes.post("/runs", async (context) => { const input = startRunRequestSchema.parse(await context.req.json()); const controller = new AbortController(); @@ -24,13 +27,31 @@ export function createRunRoutes( ); }); routes.post("/runs/:runId/cancel", async (context) => { - const events = await service.cancelWaiting(context.req.param("runId")); + const events = await service.cancel(context.req.param("runId")); for (const event of events) await registry.publish(event); + if (events.length === 0) + await registry.waitForTerminal(context.req.param("runId")); + const run = await service.getRun(context.req.param("runId")); return context.json({ runId: context.req.param("runId"), - status: "cancelled" as const, + status: run.status, }); }); + routes.post("/runs/:runId/retry", async (context) => { + const started = await service.retry( + context.req.param("runId"), + new AbortController().signal, + ); + void consume(started.events, registry); + return context.json( + { + conversationId: started.run.conversationId, + runId: started.run.id, + status: "running" as const, + }, + 202, + ); + }); routes.get("/runs/:runId/events", (context) => streamSSE(context, async (stream) => { const runId = context.req.param("runId"); diff --git a/apps/web/src/api/interactions.ts b/apps/web/src/api/interactions.ts index 9ecb79b..87ff39c 100644 --- a/apps/web/src/api/interactions.ts +++ b/apps/web/src/api/interactions.ts @@ -30,9 +30,3 @@ export async function answerInteraction( }), ); } - -export async function cancelWaitingRun(runId: string): Promise { - await request(`/api/runs/${encodeURIComponent(runId)}/cancel`, { - method: "POST", - }); -} diff --git a/apps/web/src/api/runs.ts b/apps/web/src/api/runs.ts index c24cbb6..1626346 100644 --- a/apps/web/src/api/runs.ts +++ b/apps/web/src/api/runs.ts @@ -3,7 +3,10 @@ import { startedRunSchema, type RunEventResponse, type StartedRunResponse, + agentRunSchema, + type AgentRunResponse, } from "@great-agent/web-contracts"; +import { request } from "./request"; export type StartRunInput = | { kind: "ordinary"; message: string } @@ -23,6 +26,26 @@ export async function startRun( return startedRunSchema.parse(value); } +export async function getRun(runId: string): Promise { + return agentRunSchema.parse( + await request(`/api/runs/${encodeURIComponent(runId)}`), + ); +} + +export async function retryRun(runId: string): Promise { + return startedRunSchema.parse( + await request(`/api/runs/${encodeURIComponent(runId)}/retry`, { + method: "POST", + }), + ); +} + +export async function cancelRun(runId: string): Promise { + await request(`/api/runs/${encodeURIComponent(runId)}/cancel`, { + method: "POST", + }); +} + export async function streamRun( runId: string, onEvent: (event: RunEventResponse) => void, diff --git a/apps/web/src/app/app-controller-types.ts b/apps/web/src/app/app-controller-types.ts new file mode 100644 index 0000000..0cd5373 --- /dev/null +++ b/apps/web/src/app/app-controller-types.ts @@ -0,0 +1,52 @@ +import type { State } from "vanjs-core"; +import type { + AgentRunResponse, + ConversationResponse, + ConversationSummaryResponse, + InteractionAnswerRequest, + InteractionResponse, + ProjectResponse, +} from "@great-agent/web-contracts"; + +export type Selection = + | { kind: "none" } + | { kind: "ordinary-draft" } + | { kind: "project-create" } + | { kind: "project-draft"; projectId: string } + | { kind: "project-rename"; projectId: string } + | { kind: "project-delete"; projectId: string } + | { kind: "conversation"; id: string }; + +export type AppController = Readonly<{ + selection: State; + recent: State; + projects: State; + projectConversations: State>; + active: State; + interactions: State; + currentRun: State; + activeRunId: State; + loading: State; + sending: State; + assistantDraft: State; + error: State; + initialize(): Promise; + startNewTask(): void; + startProjectCreation(): void; + createProject(name: string, workspaceRoot: string): Promise; + openProject(id: string): Promise; + startProjectTask(id: string): void; + startProjectRename(id: string): void; + renameProject(id: string, name: string): Promise; + startProjectDelete(id: string): void; + deleteProject(id: string): Promise; + openConversation(id: string): Promise; + send(content: string): Promise; + answerInteraction( + id: string, + answer: InteractionAnswerRequest, + ): Promise; + cancelInteraction(runId: string): Promise; + stopGenerating(): Promise; + retryRun(runId: string): Promise; +}>; diff --git a/apps/web/src/app/app-controller.ts b/apps/web/src/app/app-controller.ts index 1e99ef9..a399561 100644 --- a/apps/web/src/app/app-controller.ts +++ b/apps/web/src/app/app-controller.ts @@ -1,10 +1,11 @@ -import van, { type State } from "vanjs-core"; +import van from "vanjs-core"; import type { ConversationResponse, ConversationSummaryResponse, - ProjectResponse, InteractionAnswerRequest, InteractionResponse, + AgentRunResponse, + ProjectResponse, } from "@great-agent/web-contracts"; import { getConversation, @@ -17,51 +18,19 @@ import { listProjects, renameProject, } from "../api/projects"; -import { startRun, streamRun } from "../api/runs"; +import { + cancelRun, + getRun, + retryRun as requestRetryRun, + startRun, + streamRun, +} from "../api/runs"; import { answerInteraction as submitInteractionAnswer, - cancelWaitingRun, listInteractions, } from "../api/interactions"; - -export type Selection = - | { kind: "none" } - | { kind: "ordinary-draft" } - | { kind: "project-create" } - | { kind: "project-draft"; projectId: string } - | { kind: "project-rename"; projectId: string } - | { kind: "project-delete"; projectId: string } - | { kind: "conversation"; id: string }; - -export type AppController = Readonly<{ - selection: State; - recent: State; - projects: State; - projectConversations: State>; - active: State; - interactions: State; - loading: State; - sending: State; - assistantDraft: State; - error: State; - initialize(): Promise; - startNewTask(): void; - startProjectCreation(): void; - createProject(name: string, workspaceRoot: string): Promise; - openProject(id: string): Promise; - startProjectTask(id: string): void; - startProjectRename(id: string): void; - renameProject(id: string, name: string): Promise; - startProjectDelete(id: string): void; - deleteProject(id: string): Promise; - openConversation(id: string): Promise; - send(content: string): Promise; - answerInteraction( - id: string, - answer: InteractionAnswerRequest, - ): Promise; - cancelInteraction(runId: string): Promise; -}>; +import type { AppController, Selection } from "./app-controller-types"; +export type { AppController, Selection } from "./app-controller-types"; export function createAppController(): AppController { const selection = van.state({ kind: "none" }); @@ -72,6 +41,8 @@ export function createAppController(): AppController { >({}); const active = van.state(null); const interactions = van.state([]); + const currentRun = van.state(null); + const activeRunId = van.state(null); const loading = van.state(false); const sending = van.state(false); const assistantDraft = van.state(""); @@ -96,6 +67,7 @@ export function createAppController(): AppController { selection.val = next; active.val = null; interactions.val = []; + currentRun.val = null; error.val = ""; } function startNewTask() { @@ -163,12 +135,14 @@ export function createAppController(): AppController { await perform(async () => { active.val = await getConversation(id); interactions.val = await listInteractions(id); + currentRun.val = await loadLatestRun(active.val); selection.val = { kind: "conversation", id }; }); } async function send(content: string) { - if (!content.trim() || sending.val) return; + if (!content.trim() || sending.val) return false; + let accepted = false; sending.val = true; assistantDraft.val = ""; error.val = ""; @@ -189,16 +163,23 @@ export function createAppController(): AppController { } : { kind: "ordinary" as const, message: content }; const started = await startRun(input); + accepted = true; + activeRunId.val = started.runId; + currentRun.val = await getRun(started.runId); active.val = await getConversation(started.conversationId); selection.val = { kind: "conversation", id: started.conversationId }; await refreshLists(active.val.projectId); - await streamRun(started.runId, collectDelta); + await streamRun(started.runId, collectEvent); await refreshActive(started.conversationId); } catch (cause) { - error.val = readMessage(cause); + const message = readMessage(cause); + await settleActiveRun(); + error.val = message; } finally { + activeRunId.val = null; sending.val = false; } + return accepted; } async function answerInteraction( @@ -215,16 +196,21 @@ export function createAppController(): AppController { item.id === id ? resolved.interaction : item, ); if (resolved.resumed) { + activeRunId.val = resolved.interaction.runId; + currentRun.val = await getRun(resolved.interaction.runId); await streamRun( resolved.interaction.runId, - collectDelta, + collectEvent, resolved.resumeFromSequence, ); } await refreshActive(resolved.interaction.conversationId); } catch (cause) { - error.val = readMessage(cause); + const message = readMessage(cause); + await settleActiveRun(); + error.val = message; } finally { + activeRunId.val = null; sending.val = false; } } @@ -234,16 +220,62 @@ export function createAppController(): AppController { sending.val = true; error.val = ""; try { - await cancelWaitingRun(runId); + await cancelRun(runId); if (active.val) interactions.val = await listInteractions(active.val.id); + currentRun.val = await getRun(runId); } catch (cause) { - error.val = readMessage(cause); + const message = readMessage(cause); + await settleActiveRun(); + error.val = message; } finally { sending.val = false; } } - function collectDelta(event: { + async function stopGenerating() { + const runId = activeRunId.val; + if (!runId) return; + try { + await cancelRun(runId); + } catch (cause) { + error.val = readMessage(cause); + } + } + + async function settleActiveRun() { + const runId = activeRunId.val; + if (!runId) return; + try { + await cancelRun(runId); + currentRun.val = await getRun(runId); + if (active.val) await refreshActive(active.val.id); + } catch { + // Preserve the original stream or model error shown to the user. + } + } + + async function retryFailedRun(runId: string) { + if (sending.val) return; + sending.val = true; + assistantDraft.val = ""; + error.val = ""; + try { + const started = await requestRetryRun(runId); + activeRunId.val = started.runId; + currentRun.val = await getRun(started.runId); + await streamRun(started.runId, collectEvent); + await refreshActive(started.conversationId); + } catch (cause) { + const message = readMessage(cause); + await settleActiveRun(); + error.val = message; + } finally { + activeRunId.val = null; + sending.val = false; + } + } + + function collectEvent(event: { type: string; payload: Record; }) { @@ -252,6 +284,19 @@ export function createAppController(): AppController { typeof event.payload.delta === "string" ) assistantDraft.val += event.payload.delta; + if (currentRun.val && event.type === "run.failed") + currentRun.val = { + ...currentRun.val, + status: "failed", + errorCode: + typeof event.payload.code === "string" + ? event.payload.code + : "MODEL_UNAVAILABLE", + }; + if (currentRun.val && event.type === "run.cancelled") + currentRun.val = { ...currentRun.val, status: "cancelled" }; + if (currentRun.val && event.type === "run.completed") + currentRun.val = { ...currentRun.val, status: "completed" }; } async function refreshActive(conversationId: string) { @@ -259,6 +304,7 @@ export function createAppController(): AppController { getConversation(conversationId), listInteractions(conversationId), ]); + currentRun.val = await loadLatestRun(active.val); assistantDraft.val = ""; } @@ -293,6 +339,8 @@ export function createAppController(): AppController { projectConversations, active, interactions, + currentRun, + activeRunId, loading, sending, assistantDraft, @@ -311,9 +359,23 @@ export function createAppController(): AppController { send, answerInteraction, cancelInteraction, + stopGenerating, + retryRun: retryFailedRun, }; } function readMessage(cause: unknown): string { return cause instanceof Error ? cause.message : "操作失败,请重试"; } + +async function loadLatestRun( + conversation: ConversationResponse | null, +): Promise { + const runId = conversation?.messages.at(-1)?.runId; + if (!runId) return null; + try { + return await getRun(runId); + } catch { + return null; + } +} diff --git a/apps/web/src/features/conversations/conversation-pane.ts b/apps/web/src/features/conversations/conversation-pane.ts index f0fa165..a2ea328 100644 --- a/apps/web/src/features/conversations/conversation-pane.ts +++ b/apps/web/src/features/conversations/conversation-pane.ts @@ -3,6 +3,7 @@ import type { AppController } from "../../app/app-controller"; import "./conversation-pane.css"; import { ProjectPanel } from "../projects/project-panel"; import { InteractionCard } from "../interactions/interaction-card"; +import { RunStatus } from "../runs/run-status"; const { article, button, div, h1, header, main, p, span, textarea } = van.tags; @@ -12,8 +13,7 @@ export function ConversationPane(controller: AppController): HTMLElement { async function submit() { const content = draft.val; if (!content.trim()) return; - await controller.send(content); - draft.val = ""; + if (await controller.send(content)) draft.val = ""; } return main( @@ -58,6 +58,7 @@ export function ConversationPane(controller: AppController): HTMLElement { ) : null, ), + () => RunStatus(controller), ) : div( { class: "onboarding" }, @@ -113,11 +114,15 @@ function Composer( button( { class: "send", - disabled: () => controller.sending.val || !draft.val.trim(), - onclick: () => void submit(), - "aria-label": "发送消息", + disabled: () => !controller.sending.val && !draft.val.trim(), + onclick: () => + controller.sending.val + ? void controller.stopGenerating() + : void submit(), + "aria-label": () => + controller.sending.val ? "停止生成" : "发送消息", }, - () => (controller.sending.val ? "…" : "↑"), + () => (controller.sending.val ? "■" : "↑"), ), ), ), diff --git a/apps/web/src/features/runs/run-status.css b/apps/web/src/features/runs/run-status.css new file mode 100644 index 0000000..5679890 --- /dev/null +++ b/apps/web/src/features/runs/run-status.css @@ -0,0 +1,51 @@ +.run-status { + display: grid; + grid-template-columns: 26px 1fr auto; + gap: 10px; + align-items: start; + margin: 22px 0 0 37px; + padding: 12px 13px; + border: 1px solid #67443d; + border-radius: 10px; + background: #302522; +} +.run-status.cancelled { + border-color: #494641; + background: #282725; +} +.run-status-mark { + display: grid; + width: 24px; + height: 24px; + place-items: center; + border-radius: 7px; + color: #e6a092; + background: #5a302a; + font-size: 10px; + font-weight: 700; +} +.run-status.cancelled .run-status-mark { + color: #aaa59d; + background: #403e3a; +} +.run-status p { + margin: 0; +} +.run-status-title { + color: #e0dcd5; + font-size: 11px; + font-weight: 600; +} +.run-status .run-status-detail { + margin-top: 4px; + color: #918c85; + font-size: 10px; + line-height: 1.4; +} +.run-status button { + padding: 6px 10px; + border-radius: 7px; + color: #ddd8d0; + background: #45423e; + font-size: 10px; +} diff --git a/apps/web/src/features/runs/run-status.ts b/apps/web/src/features/runs/run-status.ts new file mode 100644 index 0000000..3f82711 --- /dev/null +++ b/apps/web/src/features/runs/run-status.ts @@ -0,0 +1,42 @@ +import van from "vanjs-core"; +import type { AppController } from "../../app/app-controller"; +import "./run-status.css"; + +const { button, div, p, span } = van.tags; + +export function RunStatus(controller: AppController): HTMLElement | null { + const run = controller.currentRun.val; + if (!run || (run.status !== "failed" && run.status !== "cancelled")) + return null; + const failed = run.status === "failed"; + return div( + { class: `run-status ${run.status}` }, + span({ class: "run-status-mark" }, failed ? "!" : "■"), + div( + p( + { class: "run-status-title" }, + failed ? errorTitle(run.errorCode) : "已停止生成", + ), + p( + { class: "run-status-detail" }, + failed + ? "本次任务没有成功完成,可以使用同一条用户消息重试。" + : "已保留停止前成功生成的内容。", + ), + ), + button( + { + disabled: controller.sending, + onclick: () => void controller.retryRun(run.id), + }, + "重试", + ), + ); +} + +function errorTitle(code?: string): string { + if (code === "MODEL_TIMEOUT") return "模型响应超时"; + if (code === "MODEL_CONFIG_MISSING") return "模型尚未配置"; + if (code === "MODEL_RESPONSE_INVALID") return "模型没有返回有效内容"; + return "任务运行失败"; +} diff --git a/docs/FEATURES.md b/docs/FEATURES.md index f83615b..a43c38e 100644 --- a/docs/FEATURES.md +++ b/docs/FEATURES.md @@ -67,13 +67,18 @@ ### F-005——停止、失败与重试 -- 状态:进行中 +- 状态:已完成(Agent 于 2026-08-14 验证通过,等待最终统一人工审核) - 用户可见结果:停止当前生成并从失败状态重试。 - 主要验收:AC-005、AC-010、AC-011。 +- 实现结果:Agent Core 持有每个活动 Run 的取消控制器,停止操作真正中断 ModelPort;停止前已生成内容持久化,Run 和事件日志进入 `cancelled` 终态。失败或取消后创建带 `retryOfRunId` 的新 Run,复用原触发消息且不重复写入用户消息。Repository 与串行启动锁共同执行全局单活动 Run 检查,第二个任务返回 `RUN_ALREADY_ACTIVE`。 +- 失败处理:DeepSeek 请求增加与环境配置一致的模型超时并映射 `MODEL_TIMEOUT`;Bun 服务空闲超时与模型超时对齐;流式连接提前中断时前端结束加载并主动取消仍在运行的后台任务。 +- 页面结果:生成时发送按钮切换为“停止生成”;失败和取消显示明确状态及“重试”按钮;API 启动失败时保留输入草稿。 +- 自动化验证结果:2026-08-14 通过全仓类型检查、24 项测试和 84 个断言、生产构建、代码检查、格式检查、文件规模检查与架构依赖检查。 +- 人工操作结果:2026-08-14 使用隔离慢速 Fake Model 验证停止按钮、部分回复保留、取消终态和重试成功;并通过两个连续 HTTP 请求确认活动任务期间第二个请求稳定返回 `409 RUN_ALREADY_ACTIVE`;页面控制台无错误。 ### F-006——附件、文件列表、读取与搜索 -- 状态:待开始 +- 状态:进行中 - 用户可见结果:附加工作区文件并让 Agent 安全读取和搜索。 - 主要验收:AC-007 至 AC-009、AC-021、AC-022、AC-028、AC-030。 diff --git a/packages/agent-core/src/domain/agent-run.ts b/packages/agent-core/src/domain/agent-run.ts index 6ca255a..83c9361 100644 --- a/packages/agent-core/src/domain/agent-run.ts +++ b/packages/agent-core/src/domain/agent-run.ts @@ -13,6 +13,7 @@ export type AgentRun = Readonly<{ createdAt: string; updatedAt: string; errorCode?: string; + retryOfRunId?: string; }>; export type RunEvent = Readonly<{ diff --git a/packages/agent-core/src/ports/run-repository.ts b/packages/agent-core/src/ports/run-repository.ts index 8bf77e3..02484d0 100644 --- a/packages/agent-core/src/ports/run-repository.ts +++ b/packages/agent-core/src/ports/run-repository.ts @@ -4,6 +4,7 @@ export interface RunRepository { create(run: AgentRun): Promise; update(run: AgentRun): Promise; getById(id: string): Promise; + findActive(): Promise; appendEvent(event: RunEvent): Promise; listEvents(runId: string): Promise; hasActiveForConversations( diff --git a/packages/agent-core/src/use-cases/agent-run-service.test.ts b/packages/agent-core/src/use-cases/agent-run-service.test.ts index 6792c77..78f48ed 100644 --- a/packages/agent-core/src/use-cases/agent-run-service.test.ts +++ b/packages/agent-core/src/use-cases/agent-run-service.test.ts @@ -56,15 +56,25 @@ class MemoryConversationRepository implements ConversationRepository { class MemoryRunRepository implements RunRepository { value: AgentRun | null = null; + readonly values = new Map(); readonly events: RunEvent[] = []; async create(run: AgentRun) { this.value = run; + this.values.set(run.id, run); } async update(run: AgentRun) { this.value = run; + this.values.set(run.id, run); } async getById(id: string) { - return this.value?.id === id ? this.value : null; + return this.values.get(id) ?? null; + } + async findActive() { + return ( + [...this.values.values()].find( + (run) => run.status === "running" || run.status === "waiting_user", + ) ?? null + ); } async appendEvent(event: RunEvent) { this.events.push(event); @@ -170,4 +180,117 @@ describe("AgentRunService", () => { .messages[0]?.id, ).toBe(started.run.triggerMessageId); }); + + test("停止生成会中断模型并保存部分内容和取消终态", async () => { + const fixture = createFixture({ + async *stream(_request, signal) { + yield { type: "text.delta", delta: "已经生成的部分" }; + if (!signal.aborted) + await new Promise((resolve) => + signal.addEventListener("abort", () => resolve(), { once: true }), + ); + throw signal.reason; + }, + }); + const started = await fixture.service.start( + { kind: "ordinary", message: "开始长任务" }, + new AbortController().signal, + ); + const events: RunEvent[] = []; + const consuming = (async () => { + for await (const event of started.events) { + events.push(event); + if (event.type === "message.delta") + await fixture.service.cancel(started.run.id); + } + })(); + await consuming; + expect(events.at(-1)?.type).toBe("run.cancelled"); + expect(fixture.runs.value?.status).toBe("cancelled"); + expect( + ( + await fixture.conversationService.getConversation( + started.run.conversationId, + ) + ).messages.at(-1)?.content, + ).toBe("已经生成的部分"); + }); + + test("全局已有活动任务时拒绝创建第二个 Run", async () => { + const fixture = createFixture({ + async *stream() { + yield { type: "text.delta", delta: "完成" }; + }, + }); + const first = await fixture.service.start( + { kind: "ordinary", message: "第一个任务" }, + new AbortController().signal, + ); + await expect( + fixture.service.start( + { kind: "ordinary", message: "第二个任务" }, + new AbortController().signal, + ), + ).rejects.toMatchObject({ code: "RUN_ALREADY_ACTIVE" }); + expect(fixture.conversations.values.size).toBe(1); + await fixture.service.cancel(first.run.id); + for await (const _event of first.events) { + // Drain the cancelled execution so it reaches a terminal state. + } + }); + + test("失败后重试关联原 Run 且不重复用户消息", async () => { + let calls = 0; + const fixture = createFixture({ + async *stream() { + calls += 1; + if (calls === 1) throw new Error("temporary"); + yield { type: "text.delta", delta: "重试成功" }; + }, + }); + const first = await fixture.service.start( + { kind: "ordinary", message: "只保存一次" }, + new AbortController().signal, + ); + for await (const _event of first.events) { + // Complete the failed attempt. + } + const retried = await fixture.service.retry( + first.run.id, + new AbortController().signal, + ); + for await (const _event of retried.events) { + // Complete the retry. + } + const conversation = await fixture.conversationService.getConversation( + first.run.conversationId, + ); + expect(retried.run.retryOfRunId).toBe(first.run.id); + expect(retried.run.triggerMessageId).toBe(first.run.triggerMessageId); + expect( + conversation.messages.filter((message) => message.role === "user"), + ).toHaveLength(1); + expect(conversation.messages.at(-1)?.content).toBe("重试成功"); + }); }); + +function createFixture(model: ModelPort) { + let id = 0; + const conversations = new MemoryConversationRepository(); + const runs = new MemoryRunRepository(); + const conversationService = new ConversationService({ + conversations, + clock: { now: () => new Date("2026-08-14T00:00:00Z") }, + ids: { create: () => `fixture_${++id}` }, + }); + const service = new AgentRunService( + conversationService, + model, + runs, + { now: () => new Date("2026-08-14T00:00:00Z") }, + { create: () => `fixture_${++id}` }, + projects, + interactions, + ); + return { service, runs, conversations, conversationService }; +} diff --git a/packages/agent-core/src/use-cases/agent-run-service.ts b/packages/agent-core/src/use-cases/agent-run-service.ts index 2854781..c6adab9 100644 --- a/packages/agent-core/src/use-cases/agent-run-service.ts +++ b/packages/agent-core/src/use-cases/agent-run-service.ts @@ -2,54 +2,25 @@ import type { AgentRun, RunEvent } from "../domain/agent-run"; import type { Conversation } from "../domain/conversation"; import type { UserInteraction } from "../domain/user-interaction"; import { CoreError } from "../errors/core-error"; -import type { ModelPort, ModelRequest, ModelTool } from "../ports/model-port"; +import type { ModelPort, ModelRequest } from "../ports/model-port"; import type { RunRepository } from "../ports/run-repository"; import type { ClockPort, IdPort } from "../ports/system-ports"; import type { ConversationService } from "./conversation-service"; import type { InteractionService } from "./interaction-service"; import type { ProjectService } from "./project-service"; - -export type StartRunInput = - | Readonly<{ kind: "ordinary"; message: string }> - | Readonly<{ kind: "project"; projectId: string; message: string }> - | Readonly<{ kind: "existing"; conversationId: string; message: string }>; - -export type StartedRun = Readonly<{ - run: AgentRun; - events: AsyncIterable; -}>; - -const interactionTool: ModelTool = { - name: "request_user_interaction", - description: "需要用户选择、确认或补充意见时调用。调用后等待用户回答。", - parameters: { - type: "object", - required: ["kind", "question"], - properties: { - kind: { - type: "string", - enum: ["single_choice", "multiple_choice", "confirmation", "free_text"], - }, - question: { type: "string" }, - description: { type: "string" }, - required: { type: "boolean" }, - options: { - type: "array", - minItems: 2, - maxItems: 10, - items: { - type: "object", - required: ["id", "label"], - properties: { id: { type: "string" }, label: { type: "string" } }, - }, - }, - minSelections: { type: "integer", minimum: 0 }, - maxSelections: { type: "integer", minimum: 1 }, - }, - }, -}; +import { interactionTool } from "./interaction-tool"; +import { + lastSequence, + messagesThroughTrigger, + safeErrorMessage, +} from "./run-service-helpers"; +import type { StartedRun, StartRunInput } from "./run-types"; +export type { StartedRun, StartRunInput } from "./run-types"; export class AgentRunService { + private readonly controllers = new Map(); + private startQueue: Promise = Promise.resolve(); + constructor( private readonly conversations: ConversationService, private readonly model: ModelPort, @@ -61,22 +32,28 @@ export class AgentRunService { ) {} async start(input: StartRunInput, signal: AbortSignal): Promise { - const runId = this.ids.create(); - const conversation = await this.prepareConversation(input, runId); - const timestamp = this.clock.now().toISOString(); - const triggerMessage = conversation.messages.at(-1); - if (triggerMessage?.role !== "user") - throw new CoreError("RUN_TRIGGER_INVALID", "无法确定本次运行的用户消息"); - const run: AgentRun = { - id: runId, - conversationId: conversation.id, - triggerMessageId: triggerMessage.id, - status: "running", - createdAt: timestamp, - updatedAt: timestamp, - }; - await this.runs.create(run); - return { run, events: this.execute(run, signal, 0) }; + return this.withStartLock(async () => { + await this.ensureNoActiveRun(); + const runId = this.ids.create(); + const conversation = await this.prepareConversation(input, runId); + const timestamp = this.clock.now().toISOString(); + const triggerMessage = conversation.messages.at(-1); + if (triggerMessage?.role !== "user") + throw new CoreError( + "RUN_TRIGGER_INVALID", + "无法确定本次运行的用户消息", + ); + const run: AgentRun = { + id: runId, + conversationId: conversation.id, + triggerMessageId: triggerMessage.id, + status: "running", + createdAt: timestamp, + updatedAt: timestamp, + }; + await this.runs.create(run); + return this.started(run, signal, 0); + }); } async resume( @@ -101,17 +78,27 @@ export class AgentRunService { toolArguments: interaction.toolArguments, result: JSON.stringify(interaction.answer), }; - return { - run, - events: this.execute(run, signal, sequence, continuation, interaction), - }; + return this.started(run, signal, sequence, continuation, interaction); } - async cancelWaiting(runId: string): Promise { + async cancel(runId: string): Promise { const run = await this.requireRun(runId); - if (run.status === "cancelled") return []; - if (run.status !== "waiting_user") - throw new CoreError("RUN_NOT_WAITING", "该任务当前不在等待用户回答"); + if ( + run.status === "cancelled" || + run.status === "completed" || + run.status === "failed" + ) + return []; + if (run.status === "running") { + const controller = this.controllers.get(runId); + if (!controller) + throw new CoreError( + "RUN_NOT_CANCELLABLE", + "任务正在恢复中,请稍后重试", + ); + controller.abort(); + return []; + } const interaction = await this.interactions.cancelPending(runId); if (!interaction) throw new CoreError("INTERACTION_NOT_FOUND", "等待中的交互请求不存在"); @@ -136,6 +123,38 @@ export class AgentRunService { return [cancelled, finished]; } + async retry(runId: string, signal: AbortSignal): Promise { + return this.withStartLock(async () => { + await this.ensureNoActiveRun(); + const original = await this.requireRun(runId); + if (original.status !== "failed" && original.status !== "cancelled") + throw new CoreError( + "RUN_NOT_RETRYABLE", + "只有失败或已取消的任务可以重试", + ); + const conversation = await this.conversations.getConversation( + original.conversationId, + ); + const trigger = conversation.messages.find( + (message) => message.id === original.triggerMessageId, + ); + if (trigger?.role !== "user") + throw new CoreError("RUN_TRIGGER_INVALID", "原任务的用户消息不存在"); + const timestamp = this.clock.now().toISOString(); + const run: AgentRun = { + id: this.ids.create(), + conversationId: original.conversationId, + triggerMessageId: original.triggerMessageId, + retryOfRunId: original.id, + status: "running", + createdAt: timestamp, + updatedAt: timestamp, + }; + await this.runs.create(run); + return this.started(run, signal, 0); + }); + } + async lastSequence(runId: string): Promise { await this.requireRun(runId); return lastSequence(await this.runs.listEvents(runId)); @@ -145,6 +164,45 @@ export class AgentRunService { return this.requireRun(runId); } + private started( + run: AgentRun, + externalSignal: AbortSignal, + sequence: number, + continuation?: NonNullable, + interaction?: UserInteraction, + ): StartedRun { + const controller = new AbortController(); + this.controllers.set(run.id, controller); + const signal = AbortSignal.any([externalSignal, controller.signal]); + return { + run, + events: this.execute(run, signal, sequence, continuation, interaction), + }; + } + + private async ensureNoActiveRun(): Promise { + const active = await this.runs.findActive(); + if (active) + throw new CoreError( + "RUN_ALREADY_ACTIVE", + "已有任务正在运行,请先停止或完成当前任务", + ); + } + + private async withStartLock(operation: () => Promise): Promise { + const previous = this.startQueue; + let release = () => {}; + this.startQueue = new Promise((resolve) => { + release = resolve; + }); + await previous; + try { + return await operation(); + } finally { + release(); + } + } + private async prepareConversation( input: StartRunInput, runId: string, @@ -179,6 +237,9 @@ export class AgentRunService { resolvedInteraction?: UserInteraction, ): AsyncIterable { let sequence = initialSequence; + let messageId: string | undefined; + let content = ""; + let messagePersisted = false; const event = async ( type: RunEvent["type"], payload: Record, @@ -193,21 +254,25 @@ export class AgentRunService { interactionId: resolvedInteraction.id, answer: resolvedInteraction.answer, }); - const messageId = this.ids.create(); + messageId = this.ids.create(); yield await event("message.started", { messageId }); const conversation = await this.conversations.getConversation( run.conversationId, ); - let content = ""; + const modelMessages = run.retryOfRunId + ? messagesThroughTrigger(conversation, run.triggerMessageId) + : conversation.messages; + signal.throwIfAborted(); for await (const modelEvent of this.model.stream( { model: "default", - messages: conversation.messages, + messages: modelMessages, tools: [interactionTool], ...(continuation ? { continuation } : {}), }, signal, )) { + signal.throwIfAborted(); if (modelEvent.type === "text.delta") { content += modelEvent.delta; yield await event("message.delta", { @@ -228,6 +293,7 @@ export class AgentRunService { run.id, content, ); + messagePersisted = true; yield await event("message.completed", { messageId, content }); const interaction = await this.interactions.create({ runId: run.id, @@ -245,6 +311,7 @@ export class AgentRunService { return; } } + signal.throwIfAborted(); if (!content) throw new CoreError("MODEL_RESPONSE_INVALID", "模型没有返回有效内容"); await this.conversations.appendAssistantMessage( @@ -253,6 +320,7 @@ export class AgentRunService { run.id, content, ); + messagePersisted = true; yield await event("message.completed", { messageId, content }); await this.runs.update({ ...run, @@ -261,6 +329,28 @@ export class AgentRunService { }); yield await event("run.completed", {}); } catch (cause) { + if (signal.aborted) { + if (content && messageId && !messagePersisted) { + await this.conversations.appendAssistantMessage( + run.conversationId, + messageId, + run.id, + content, + ); + yield await event("message.completed", { + messageId, + content, + cancelled: true, + }); + } + await this.runs.update({ + ...run, + status: "cancelled", + updatedAt: this.clock.now().toISOString(), + }); + yield await event("run.cancelled", {}); + return; + } const code = cause instanceof CoreError ? cause.code : "MODEL_UNAVAILABLE"; await this.runs.update({ @@ -273,6 +363,8 @@ export class AgentRunService { code, message: safeErrorMessage(cause), }); + } finally { + this.controllers.delete(run.id); } } @@ -299,10 +391,3 @@ export class AgentRunService { return value; } } - -function lastSequence(events: readonly RunEvent[]): number { - return events.at(-1)?.sequence ?? 0; -} -function safeErrorMessage(cause: unknown): string { - return cause instanceof CoreError ? cause.message : "模型服务暂时不可用"; -} diff --git a/packages/agent-core/src/use-cases/interaction-run.test.ts b/packages/agent-core/src/use-cases/interaction-run.test.ts index ad31e85..3c48c88 100644 --- a/packages/agent-core/src/use-cases/interaction-run.test.ts +++ b/packages/agent-core/src/use-cases/interaction-run.test.ts @@ -53,6 +53,13 @@ class Runs implements RunRepository { async getById(id: string) { return this.values.get(id) ?? null; } + async findActive() { + return ( + [...this.values.values()].find( + (run) => run.status === "running" || run.status === "waiting_user", + ) ?? null + ); + } async appendEvent(event: RunEvent) { this.events.push(event); } @@ -135,7 +142,7 @@ describe("等待用户的 Run", () => { new AbortController().signal, ); await collect(started.events); - const events = await fixture.service.cancelWaiting(started.run.id); + const events = await fixture.service.cancel(started.run.id); expect(events.map((event) => event.type)).toEqual([ "interaction.cancelled", "run.cancelled", diff --git a/packages/agent-core/src/use-cases/interaction-tool.ts b/packages/agent-core/src/use-cases/interaction-tool.ts new file mode 100644 index 0000000..65a6f9d --- /dev/null +++ b/packages/agent-core/src/use-cases/interaction-tool.ts @@ -0,0 +1,31 @@ +import type { ModelTool } from "../ports/model-port"; + +export const interactionTool: ModelTool = { + name: "request_user_interaction", + description: "需要用户选择、确认或补充意见时调用。调用后等待用户回答。", + parameters: { + type: "object", + required: ["kind", "question"], + properties: { + kind: { + type: "string", + enum: ["single_choice", "multiple_choice", "confirmation", "free_text"], + }, + question: { type: "string" }, + description: { type: "string" }, + required: { type: "boolean" }, + options: { + type: "array", + minItems: 2, + maxItems: 10, + items: { + type: "object", + required: ["id", "label"], + properties: { id: { type: "string" }, label: { type: "string" } }, + }, + }, + minSelections: { type: "integer", minimum: 0 }, + maxSelections: { type: "integer", minimum: 1 }, + }, + }, +}; diff --git a/packages/agent-core/src/use-cases/project-service.test.ts b/packages/agent-core/src/use-cases/project-service.test.ts index b2ba352..8d5d104 100644 --- a/packages/agent-core/src/use-cases/project-service.test.ts +++ b/packages/agent-core/src/use-cases/project-service.test.ts @@ -62,6 +62,9 @@ class Runs implements RunRepository { async getById(_id: string) { return null; } + async findActive() { + return null; + } async appendEvent(_event: RunEvent) {} async listEvents(_id: string) { return []; diff --git a/packages/agent-core/src/use-cases/run-service-helpers.ts b/packages/agent-core/src/use-cases/run-service-helpers.ts new file mode 100644 index 0000000..3a84313 --- /dev/null +++ b/packages/agent-core/src/use-cases/run-service-helpers.ts @@ -0,0 +1,23 @@ +import type { RunEvent } from "../domain/agent-run"; +import type { Conversation } from "../domain/conversation"; +import { CoreError } from "../errors/core-error"; + +export function lastSequence(events: readonly RunEvent[]): number { + return events.at(-1)?.sequence ?? 0; +} + +export function safeErrorMessage(cause: unknown): string { + return cause instanceof CoreError ? cause.message : "模型服务暂时不可用"; +} + +export function messagesThroughTrigger( + conversation: Conversation, + triggerMessageId: string, +) { + const index = conversation.messages.findIndex( + (message) => message.id === triggerMessageId, + ); + return index < 0 + ? conversation.messages + : conversation.messages.slice(0, index + 1); +} diff --git a/packages/agent-core/src/use-cases/run-types.ts b/packages/agent-core/src/use-cases/run-types.ts new file mode 100644 index 0000000..8b25441 --- /dev/null +++ b/packages/agent-core/src/use-cases/run-types.ts @@ -0,0 +1,11 @@ +import type { AgentRun, RunEvent } from "../domain/agent-run"; + +export type StartRunInput = + | Readonly<{ kind: "ordinary"; message: string }> + | Readonly<{ kind: "project"; projectId: string; message: string }> + | Readonly<{ kind: "existing"; conversationId: string; message: string }>; + +export type StartedRun = Readonly<{ + run: AgentRun; + events: AsyncIterable; +}>; diff --git a/packages/local-data/src/repositories/file-run-repository.ts b/packages/local-data/src/repositories/file-run-repository.ts index 2eb5202..efea4df 100644 --- a/packages/local-data/src/repositories/file-run-repository.ts +++ b/packages/local-data/src/repositories/file-run-repository.ts @@ -33,6 +33,14 @@ export class FileRunRepository implements RunRepository { } } + async findActive(): Promise { + return ( + (await this.readAll()).find( + (run) => run.status === "running" || run.status === "waiting_user", + ) ?? null + ); + } + async appendEvent(event: RunEvent): Promise { await appendNdjson(this.eventsPath(event.runId), event); } diff --git a/packages/model-deepseek/src/adapter/deepseek-model-adapter.ts b/packages/model-deepseek/src/adapter/deepseek-model-adapter.ts index 04ea66d..a87ea2a 100644 --- a/packages/model-deepseek/src/adapter/deepseek-model-adapter.ts +++ b/packages/model-deepseek/src/adapter/deepseek-model-adapter.ts @@ -1,4 +1,5 @@ import OpenAI from "openai"; +import { CoreError } from "@great-agent/agent-core"; import type { Message, ModelEvent, @@ -10,6 +11,7 @@ export type DeepSeekModelOptions = Readonly<{ apiKey: string; baseURL: string; model: string; + timeoutMs: number; }>; export class DeepSeekModelAdapter implements ModelPort { @@ -26,56 +28,65 @@ export class DeepSeekModelAdapter implements ModelPort { request: ModelRequest, signal: AbortSignal, ): AsyncIterable { - const response = await this.client.chat.completions.create( - { - model: request.model === "default" ? this.options.model : request.model, - messages: toModelMessages(request), - ...(request.tools - ? { - tools: request.tools.map((tool) => ({ - type: "function" as const, - function: { - name: tool.name, - description: tool.description, - parameters: tool.parameters, - }, - })), - } - : {}), - stream: true, - }, - { signal }, - ); - const toolCalls = new Map< - number, - { id: string; name: string; arguments: string } - >(); - for await (const chunk of response) { - const choice = chunk.choices[0]?.delta; - const delta = choice?.content; - if (delta) yield { type: "text.delta", delta }; - for (const call of choice?.tool_calls ?? []) { - const current = toolCalls.get(call.index) ?? { - id: call.id ?? "", - name: call.function?.name ?? "", - arguments: "", - }; - toolCalls.set(call.index, { - id: call.id ?? current.id, - name: call.function?.name ?? current.name, - arguments: current.arguments + (call.function?.arguments ?? ""), - }); + const timeoutSignal = AbortSignal.timeout(this.options.timeoutMs); + const requestSignal = AbortSignal.any([signal, timeoutSignal]); + try { + const response = await this.client.chat.completions.create( + { + model: + request.model === "default" ? this.options.model : request.model, + messages: toModelMessages(request), + ...(request.tools + ? { + tools: request.tools.map((tool) => ({ + type: "function" as const, + function: { + name: tool.name, + description: tool.description, + parameters: tool.parameters, + }, + })), + } + : {}), + stream: true, + }, + { signal: requestSignal }, + ); + const toolCalls = new Map< + number, + { id: string; name: string; arguments: string } + >(); + for await (const chunk of response) { + const choice = chunk.choices[0]?.delta; + const delta = choice?.content; + if (delta) yield { type: "text.delta", delta }; + for (const call of choice?.tool_calls ?? []) { + const current = toolCalls.get(call.index) ?? { + id: call.id ?? "", + name: call.function?.name ?? "", + arguments: "", + }; + toolCalls.set(call.index, { + id: call.id ?? current.id, + name: call.function?.name ?? current.name, + arguments: current.arguments + (call.function?.arguments ?? ""), + }); + } } + for (const call of toolCalls.values()) { + yield { + type: "tool.requested", + toolCallId: call.id, + name: call.name, + arguments: call.arguments, + }; + } + yield { type: "response.completed" }; + } catch (cause) { + if (timeoutSignal.aborted && !signal.aborted) + throw new CoreError("MODEL_TIMEOUT", "模型响应超时,请重试"); + throw cause; } - for (const call of toolCalls.values()) { - yield { - type: "tool.requested", - toolCallId: call.id, - name: call.name, - arguments: call.arguments, - }; - } - yield { type: "response.completed" }; } } diff --git a/packages/web-contracts/src/index.ts b/packages/web-contracts/src/index.ts index 3855c6e..2e8d0e6 100644 --- a/packages/web-contracts/src/index.ts +++ b/packages/web-contracts/src/index.ts @@ -8,7 +8,12 @@ export { type RenameProjectRequest, } from "./requests/project"; export { runEventSchema, type RunEventResponse } from "./events/run-event"; -export { startedRunSchema, type StartedRunResponse } from "./responses/run"; +export { + agentRunSchema, + startedRunSchema, + type AgentRunResponse, + type StartedRunResponse, +} from "./responses/run"; export { conversationSchema, conversationSummarySchema, diff --git a/packages/web-contracts/src/responses/run.ts b/packages/web-contracts/src/responses/run.ts index 2c9176e..ef1492a 100644 --- a/packages/web-contracts/src/responses/run.ts +++ b/packages/web-contracts/src/responses/run.ts @@ -6,4 +6,22 @@ export const startedRunSchema = z.object({ status: z.literal("running"), }); +export const agentRunSchema = z.object({ + id: z.string(), + conversationId: z.string(), + triggerMessageId: z.string(), + status: z.enum([ + "running", + "waiting_user", + "completed", + "failed", + "cancelled", + ]), + createdAt: z.string(), + updatedAt: z.string(), + errorCode: z.string().optional(), + retryOfRunId: z.string().optional(), +}); + export type StartedRunResponse = z.infer; +export type AgentRunResponse = z.infer;