diff --git a/api-report/apps/api/src/modules/trace/contract.d.ts b/api-report/apps/api/src/modules/trace/contract.d.ts index ba4d8a8..c3970d1 100644 --- a/api-report/apps/api/src/modules/trace/contract.d.ts +++ b/api-report/apps/api/src/modules/trace/contract.d.ts @@ -76,6 +76,8 @@ export type TraceService = { workflowId?: string; nodeId?: string; output?: Record; + /** 非放行终态的原因 —— 有它就把这一段标成错,时间轴上一眼看出哪儿坏了 */ + error?: string; }): Promise; endWorkflow(input: { workflowId: string; diff --git a/apps/api/src/app/wiring/engines.ts b/apps/api/src/app/wiring/engines.ts index 890b1be..6e7eaca 100644 --- a/apps/api/src/app/wiring/engines.ts +++ b/apps/api/src/app/wiring/engines.ts @@ -991,8 +991,24 @@ export function wireEngines(deps: EngineWiringDeps): EngineWiring { const tracedProcess: ProcessService = { ...process, async onTaskTerminal(taskId, terminal, ports) { - // 节点 span 先收(这一步办完了),再让流程往下走 —— 时间轴上不倒序 - await deps.trace.endNode({ taskId, output: { terminal } }); + /* + * 节点 span 先收(这一步办完了),再让流程往下走 —— 时间轴上不倒序。 + * **非放行终态标成错**:blocked / exhausted / 失败都要在时间轴上看得出来, + * 收成一条无害的「已结束」等于把事故藏起来(实测:落地失败的那几单, + * 观测面上与走完的单长得一模一样)。原因从任务上现读。 + */ + const failed = terminal !== 'released'; + const reason = failed + ? await task + .get(taskId as TaskId) + .then((t) => JSON.stringify(t.reason ?? terminal)) + .catch(() => terminal) + : undefined; + await deps.trace.endNode({ + taskId, + output: { terminal }, + ...(reason === undefined ? {} : { error: `${terminal}:${reason}` }), + }); const state = await process.onTaskTerminal(taskId, terminal, ports); return state; }, diff --git a/apps/api/src/modules/task/__tests__/landing-failure.test.ts b/apps/api/src/modules/task/__tests__/landing-failure.test.ts new file mode 100644 index 0000000..92d3c81 --- /dev/null +++ b/apps/api/src/modules/task/__tests__/landing-failure.test.ts @@ -0,0 +1,138 @@ +/** + * 放行落地**失败**的收场 —— 实测抓到的一条真事故(2026-07-29)。 + * + * 症状:点了通过,任务却以 `exhausted` 收场,界面上什么都看不到。 + * 机理:`land()` 抛错(草稿与正式版并发冲突)→ 异常穿到推进器 → 推进器写一行 + * stderr 就 break;而这一拍**已经把 runtime 落回 running、已经扣了一步预算** —— + * 两秒后又来一次,一直空转到 steps 触顶(实测 3 步的活跑了 40 步)。 + * 真正的原因只在服务端 stderr 里刷屏,任务记忆、界面、trace 上一个字都没有。 + * + * 定的规矩:**落地失败是终态,不是可重试的中间态** —— 就地落 `blocked`, + * 原因写进任务记忆(人读得到),推进器再也不会来推它。 + */ +import { describe, expect, it } from 'vitest'; +import type { TaskId } from '@contracts'; +import type { Task } from '../contract.ts'; +import { createTaskService } from '../internal/usecase/task-service.ts'; +import type { TaskServiceDeps } from '../internal/usecase/task-service.ts'; + +const TASK = 't-1' as TaskId; + +type Harness = { + service: ReturnType; + readonly memories: Array<{ kind: string; title: string }>; + readonly landCalls: number[]; + task: Task; +}; + +/** 谱已走完、放行通过 —— 只在「落地这一下」上做文章 */ +function harness(land: () => Promise | Promise<{ converged: boolean; emittedPorts: string[]; ledgerRowsMerged: number; facts: unknown }>): Harness { + const memories: Harness['memories'] = []; + const landCalls: number[] = []; + const h: Harness = { service: undefined as never, memories, landCalls, task: undefined as never }; + h.task = { + id: TASK, + activityId: 'a-1', + paradigm: 'flow', + ownerId: 's-1' as Task['ownerId'], + title: '占位活动', + runtime: 'running', + envelope: { activityId: 'a-1', packageVersion: '0.7.0' }, + data: {}, + caps: { steps: 40, tokens: 200000, humanCalls: 2, activeMinutes: 60 }, + used: { steps: 3, tokens: 0, activeMs: 100, humanCalls: 1 }, + startedAt: '2026-07-29T00:00:00.000Z', + }; + + const deps: TaskServiceDeps = { + store: { + get: async () => h.task, + update: async (_id, patch) => { + h.task = { ...h.task, ...patch }; + return h.task; + }, + addConsumption: async () => h.task.used, + lastProgressAt: async () => new Date('2026-07-29T00:00:01.000Z'), + markProgress: async () => undefined, + listByOwner: async () => [], + insert: async () => undefined, + }, + activities: { + load: async () => ({ + paradigm: 'flow' as const, + caps: h.task.caps, + envelope: {}, + inputSchema: {}, + businessModel: {}, + stallAfter: 'PT30M', + }), + }, + schemas: { validate: () => ({ ok: true, errors: [] }) }, + engines: { + // 谱走完 + 放行通过 —— 接下来就该落地 + flow: { load: async () => undefined, advance: async () => ({ status: 'finished', released: true, selectedPorts: {} }), onEvent: async () => ({ status: 'running' }) }, + loop: { load: async () => undefined, advance: async () => ({ status: 'running' }), onEvent: async () => ({ status: 'running' }) }, + }, + memory: { + append: async (input) => { + memories.push({ kind: input.kind, title: input.title }); + }, + hasHistory: async () => true, + seal: async () => undefined, + distill: async () => undefined, + }, + landing: { + land: async () => { + landCalls.push(landCalls.length + 1); + return land(); + }, + } as TaskServiceDeps['landing'], + listeners: { hasArmedListener: async () => false }, + persistence: { hasDrafts: async () => false, hasRuntimeState: async () => true }, + events: { publishTerminal: async () => undefined, publishControl: async () => undefined }, + notifier: { notify: async () => undefined }, + interaction: { setPhase: async () => undefined }, + clock: { now: () => new Date('2026-07-29T00:00:02.000Z') }, + ids: { next: () => 'x' }, + }; + + h.service = createTaskService(deps); + return h; +} + +describe('放行落地失败 —— 是终态,不是可重试的中间态', () => { + it('land 抛错:落 blocked 终态,原因进任务记忆(不再空转)', async () => { + const h = harness(async () => { + throw new Error('草稿与正式版并发冲突:2 处字段已被他单改动'); + }); + + const out = await h.service.tick(TASK); + expect(out.task.terminal, '落地失败即终态').toBe('blocked'); + expect(String(out.task.reason)).toContain('放行落地失败'); + expect(String(out.task.reason)).toContain('并发冲突'); + + // 原因要人读得到 —— 从前只写 stderr,界面上一个字都没有 + const said = h.memories.find((m) => m.title.includes('放行落地失败')); + expect(said, '任务记忆里要有一条说清为什么').toBeDefined(); + expect(said!.title).toContain('并发冲突'); + }); + + it('终态之后不再被推 —— 空转到预算耗尽的那条路断了', async () => { + const h = harness(async () => { + throw new Error('落地炸了'); + }); + + await h.service.tick(TASK); + expect(h.landCalls).toHaveLength(1); + // 推进器只推 runtime==='running' 的任务;落了终态就再也不会来 + expect(h.task['runtime']).toBe('terminal'); + expect(h.task['terminal']).toBe('blocked'); + }); + + it('落地成功照旧 released —— 修的是失败那条路,别误伤正常路', async () => { + const h = harness(async () => ({ converged: true, emittedPorts: ['intake'], ledgerRowsMerged: 2, facts: {} })); + const out = await h.service.tick(TASK); + expect(out.task.terminal).toBe('released'); + expect(h.memories.some((m) => m.title.includes('落地收敛'))).toBe(true); + }); +}); diff --git a/apps/api/src/modules/task/internal/usecase/task-service.ts b/apps/api/src/modules/task/internal/usecase/task-service.ts index 41bb0aa..747ca14 100644 --- a/apps/api/src/modules/task/internal/usecase/task-service.ts +++ b/apps/api/src/modules/task/internal/usecase/task-service.ts @@ -169,12 +169,32 @@ export function createTaskService(deps: TaskServiceDeps): TaskService { await deps.interaction.setPhase(taskId, 'awaiting-event'); return { task: waiting, verdict, progressed: false }; } - const landed = await deps.landing.land({ - taskId, - ownerId: task.ownerId, - humanCallCap: task.caps.humanCalls, - selectedPorts: outcome.selectedPorts ?? {}, - }); + /* + * 落地会**抛**(草稿与正式版冲突、动作意向执行失败…)。抛出去就是灾难: + * 谱已经走完(status=finished),runtime 却还停在 running —— 推进器每两秒 + * 再来一次,每次白烧一步预算,最后以 `exhausted` 收场,而真正的原因只在 + * stderr 里刷屏(实测:一条试跑单空转 37 拍,界面上什么都看不到)。 + * 落地失败是**终态**,不是可重试的中间态:就地落 blocked,把原因写进任务记忆。 + */ + let landed; + try { + landed = await deps.landing.land({ + taskId, + ownerId: task.ownerId, + humanCallCap: task.caps.humanCalls, + selectedPorts: outcome.selectedPorts ?? {}, + }); + } catch (error) { + const detail = error instanceof Error ? error.message : String(error); + await deps.memory.append({ + taskId, + kind: 'observation', + title: `放行落地失败:${detail}`, + detail: { reason: detail }, + }); + const blocked = await this.finish(taskId, { kind: 'blocked', reason: `放行落地失败:${detail}` }); + return { task: blocked, verdict, progressed: true }; + } if (!landed.converged) { // 落地未收敛:输出口不得发射,任务受阻待人(落地失败卡已发) const blocked = await this.finish(taskId, { kind: 'blocked', reason: '落地未收敛,输出口未发射' }); diff --git a/apps/api/src/modules/trace/contract.ts b/apps/api/src/modules/trace/contract.ts index a3f6b59..4d8f4ed 100644 --- a/apps/api/src/modules/trace/contract.ts +++ b/apps/api/src/modules/trace/contract.ts @@ -69,7 +69,14 @@ export type TraceService = { input?: Record; }): SpanHandle | undefined; /** 收尾:节点终态 / 工作流终态 */ - endNode(input: { taskId?: string; workflowId?: string; nodeId?: string; output?: Record }): Promise; + endNode(input: { + taskId?: string; + workflowId?: string; + nodeId?: string; + output?: Record; + /** 非放行终态的原因 —— 有它就把这一段标成错,时间轴上一眼看出哪儿坏了 */ + error?: string; + }): Promise; endWorkflow(input: { workflowId: string; state: string }): Promise; /** 起一条 trace(工作流级)或挂在既有 trace 下 —— 低层口,便利层用它 */ diff --git a/apps/api/src/modules/trace/internal/usecase/trace-service.ts b/apps/api/src/modules/trace/internal/usecase/trace-service.ts index 45c2fad..85b512a 100644 --- a/apps/api/src/modules/trace/internal/usecase/trace-service.ts +++ b/apps/api/src/modules/trace/internal/usecase/trace-service.ts @@ -131,11 +131,16 @@ export function createTraceService(deps: TraceServiceDeps): TraceService { return parent.child({ name, kind, taskId, ...(input === undefined ? {} : { input }) }); }, - async endNode({ taskId, workflowId, nodeId, output }) { + async endNode({ taskId, workflowId, nodeId, output, error }) { const key = taskId ?? `${workflowId}:${nodeId}`; const span = nodeSpans.get(key); if (span === undefined) return; nodeSpans.delete(key); + // 坏了就标坏 —— 收成一条无害的「已结束」等于把事故藏起来 + if (error !== undefined) { + await span.fail(new Error(error)); + return; + } await span.end(output === undefined ? undefined : { output }); },