From 8fb7ad8e3c445626587d934bf1bcfe671b904a82 Mon Sep 17 00:00:00 2001 From: ficowang Date: Thu, 30 Jul 2026 16:19:34 +0800 Subject: [PATCH 1/7] =?UTF-8?q?refactor(scheduler):=20attempt=20=E9=98=B6?= =?UTF-8?q?=E6=AE=B5=E6=9C=BA=E6=98=BE=E5=BC=8F=E5=8C=96=E2=80=94=E2=80=94?= =?UTF-8?q?=E7=BB=9F=E4=B8=80=E8=BD=AC=E7=A7=BB=E5=85=A5=E5=8F=A3=20+=20?= =?UTF-8?q?=E5=8D=95=E4=B8=80=E5=87=BA=E5=8F=A3=E6=B8=85=E5=8D=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 做 #1016:PR #944 review 里「某条出口分支漏做收口动作」同型缺陷出现四次, 根因是 attempt 生命周期靠多个隐式出口各自手工记得收口。三件事: - 转移表显式化(attemptLifecycle.ts,纯逻辑):由现网全部 7 处 phase 写点 穷举推导;所有写入统一走 transitionAttempt,非法转移抛错(「静默少做一 件事」不再可能静默),幂等重入(强制收口与迟到 settle 各置一次 finalizing)按 no-op 放行。 - 单一出口清单:finishInflightAttempt 删除 attempt 时矫正性清扫所有仍指向 它的登记(controller / per-schedule 索引 / session 双向映射 / 静默标记) 并响亮告警——残留即某条出口漏了收口,在日志与测试里直接可见。 abandonedRuns 刻意不碰(设计为跨生命周期由迟到 settle 消费)。 - 登记一致性不变量:begin(注册面唯一扩张点)断言全部按 runId 键控的登记 必须指向在账 attempt,违反抛错。 刻意不做的两件事(与 issue 建议的偏差,均有依据):终态落库/重排/补通知 仍留在各自路径——fireOne 重排而 runNow 不重排、defer 撤销语义等是逐条 review 钉下的有意差异,收进单函数会改变已评审语义;「slotsInUse 永不超过 maxConcurrentRuns」不作断言——runNow 有意绕过闸门挤压槽位(代码注释明示), 该不变量对手动触发不成立,改为断言登记一致性。 行为等价验证:既有 186 测全过未改一行;新增穷举矩阵 + 排队往返/排队中 interrupt 两条端到端链路(出口零残留告警)。 Signed-off-by: ficowang --- .../src/__tests__/attemptLifecycle.test.ts | 54 +++++++++ .../src/__tests__/scheduler.test.ts | 72 ++++++++++++ .../src/engine/attemptLifecycle.ts | 45 +++++++ .../maker-scheduler/src/engine/scheduler.ts | 110 ++++++++++++++++-- 4 files changed, 271 insertions(+), 10 deletions(-) create mode 100644 packages/maker-scheduler/src/__tests__/attemptLifecycle.test.ts create mode 100644 packages/maker-scheduler/src/engine/attemptLifecycle.ts diff --git a/packages/maker-scheduler/src/__tests__/attemptLifecycle.test.ts b/packages/maker-scheduler/src/__tests__/attemptLifecycle.test.ts new file mode 100644 index 00000000000..54fe37740f0 --- /dev/null +++ b/packages/maker-scheduler/src/__tests__/attemptLifecycle.test.ts @@ -0,0 +1,54 @@ +/** + * attemptLifecycle.test.ts — attempt 阶段机转移表(#1016)。 + * 穷举矩阵钉死:合法边恰为表列(含幂等重入),其余任意相对一律非法—— + * 转移表一旦被无意扩宽/收窄,这里立刻失败,而不是等运行期抛错或静默漏收口。 + */ + +import { describe, expect, it } from 'vitest'; + +import type { ScheduleRunPhase } from '../types.js'; +import { LEGAL_PHASE_TRANSITIONS, isLegalPhaseTransition } from '../engine/attemptLifecycle.js'; + +const ALL_PHASES: readonly ScheduleRunPhase[] = [ + 'loading', + 'claiming', + 'persisting', + 'running', + 'queued', + 'cancelling', + 'finalizing', +]; + +describe('isLegalPhaseTransition', () => { + it('穷举矩阵:合法边 = 表列 ∪ 幂等重入,其余全部非法', () => { + for (const from of ALL_PHASES) { + for (const to of ALL_PHASES) { + const expected = from === to || LEGAL_PHASE_TRANSITIONS[from].includes(to); + expect(isLegalPhaseTransition(from, to), `${from} -> ${to}`).toBe(expected); + } + } + }); + + it('关键语义边逐条钉死(防表被误改)', () => { + // 两个入口相只能进 persisting。 + expect(isLegalPhaseTransition('claiming', 'persisting')).toBe(true); + expect(isLegalPhaseTransition('loading', 'persisting')).toBe(true); + expect(isLegalPhaseTransition('claiming', 'running')).toBe(false); + expect(isLegalPhaseTransition('loading', 'running')).toBe(false); + // 排队往返与撤项。 + expect(isLegalPhaseTransition('running', 'queued')).toBe(true); + expect(isLegalPhaseTransition('queued', 'running')).toBe(true); + expect(isLegalPhaseTransition('queued', 'cancelling')).toBe(true); + expect(isLegalPhaseTransition('cancelling', 'queued')).toBe(false); + expect(isLegalPhaseTransition('cancelling', 'running')).toBe(false); + // 排队中被 interrupt:runner 直接抛错,不经过 endQueueWait。 + expect(isLegalPhaseTransition('queued', 'finalizing')).toBe(true); + // controller 注册后、running 置位前的守卫窗口。 + expect(isLegalPhaseTransition('persisting', 'finalizing')).toBe(true); + // finalizing 是吸收相:幂等重入放行,任何离开都非法。 + expect(isLegalPhaseTransition('finalizing', 'finalizing')).toBe(true); + for (const to of ALL_PHASES) { + if (to !== 'finalizing') expect(isLegalPhaseTransition('finalizing', to)).toBe(false); + } + }); +}); diff --git a/packages/maker-scheduler/src/__tests__/scheduler.test.ts b/packages/maker-scheduler/src/__tests__/scheduler.test.ts index 4119199a656..0daa0549eef 100644 --- a/packages/maker-scheduler/src/__tests__/scheduler.test.ts +++ b/packages/maker-scheduler/src/__tests__/scheduler.test.ts @@ -3854,3 +3854,75 @@ describe('Scheduler: 排队不占槽与卡死守卫', () => { } }); }); + +// ── #1016:attempt 生命周期状态机(转移统一入口 + 单一出口清单) ────────────── +describe('Scheduler: attempt 生命周期状态机(#1016)', () => { + function spyLogger(): { logger: Logger; warns: unknown[][] } { + const warns: unknown[][] = []; + const logger = { + info: vi.fn(), + warn: vi.fn((...args: unknown[]) => warns.push(args)), + error: vi.fn(), + debug: vi.fn(), + } as unknown as Logger; + return { logger, warns }; + } + + it('完整生命周期(含排队往返)合法收口:零非法转移、出口零残留告警', async () => { + const { logger, warns } = spyLogger(); + let ctxRef: FireContext | undefined; + let release: (() => void) | undefined; + const h = makeHarness({ + logger, + runnerImpl: (_s, ctx) => + new Promise((resolve) => { + ctxRef = ctx; + ctx.onQueueWaitStart?.(); + release = () => { + // 排队 → 回收槽位 → 正常完成:覆盖 running→queued→running→finalizing 全链。 + expect(ctx.endQueueWait?.(true)).toBe(true); + resolve({ sessionId: 'sess-full-lifecycle' }); + }; + }), + }); + const sch = await h.scheduler.create({ ...baseInput, manual: true }); + const p = h.scheduler.runNow(sch.id); + await vi.waitFor(() => expect(ctxRef).toBeDefined()); + expect(h.scheduler.getRuntimeSnapshot().inFlightRuns[0]?.phase).toBe('queued'); + release?.(); + await p; + const snap = h.scheduler.getRuntimeSnapshot(); + expect(snap.inFlight).toBe(0); + expect(snap.slotsInUse).toBe(0); + // 单一出口清单未发现任何残留登记(残留 = 某条路径漏了收口,响亮告警)。 + expect( + warns.some((args) => String(args[0]).includes('unreaped registrations')), + ).toBe(false); + await h.scheduler.stop(); + }); + + it('排队中 runner 直接抛错(不经过 endQueueWait)→ queued→finalizing 合法收口为 failed', async () => { + const { logger, warns } = spyLogger(); + let reject: ((err: Error) => void) | undefined; + const h = makeHarness({ + logger, + runnerImpl: (_s, ctx) => + new Promise((_resolve, rej) => { + ctx.onQueueWaitStart?.(); + reject = rej; + }), + }); + const sch = await h.scheduler.create({ ...baseInput, manual: true }); + const p = h.scheduler.runNow(sch.id); + await vi.waitFor(() => expect(reject).toBeDefined()); + reject?.(new Error('queued turn interrupted')); + await p; + const runs = await h.storage.listRuns(sch.id); + expect(runs[0]?.status).toBe('failed'); + expect(h.scheduler.getRuntimeSnapshot().inFlight).toBe(0); + expect( + warns.some((args) => String(args[0]).includes('unreaped registrations')), + ).toBe(false); + await h.scheduler.stop(); + }); +}); diff --git a/packages/maker-scheduler/src/engine/attemptLifecycle.ts b/packages/maker-scheduler/src/engine/attemptLifecycle.ts new file mode 100644 index 00000000000..fd76357419e --- /dev/null +++ b/packages/maker-scheduler/src/engine/attemptLifecycle.ts @@ -0,0 +1,45 @@ +/** + * attemptLifecycle —— InflightAttempt 阶段机的合法转移表(纯逻辑,#1016)。 + * + * 背景:PR #944 的 review 里「某条出口分支漏做收口动作」同型缺陷出现了四次—— + * attempt 的生命周期有多个隐式出口,每个出口都靠手工记得做全套收口。本表把 + * 阶段转移显式化:所有 phase 写入统一走 Scheduler.transitionAttempt,非法转移 + * **抛错**而不是静默容忍(多数漏项的表现正是「静默少做一件事」)。 + * + * 表由现网全部 7 处写点穷举推导(fireOneInner / runNowInner / updateInflightAttempt / + * buildOnQueueWaitStart / buildEndQueueWait ×2 / forceReleaseStalledRun): + * + * claiming ──→ persisting ──→ running ⇄ queued ──→ cancelling + * loading ───↗ │ │ │ + * └────→ finalizing ←────┘ + * + * - 'claiming'(自动 fire)与 'loading'(runNow)是两个入口相,只能进 'persisting'; + * 认领失败 / 行读不到等早退不经过任何转移,直接删除(删除 = 任意相合法出口, + * 出口清单由 finishInflightAttempt 的单一出口统一执行)。 + * - 'finalizing' 是吸收相:强制收口与迟到 settle 会各自尝试置一次,幂等重入 + * (from === to)按 no-op 放行,其余任何离开 'finalizing' 的转移都非法。 + * - 'queued' → 'finalizing' 合法:排队中的 turn 被 interrupt 时 runner 直接抛错, + * 不经过 endQueueWait。 + * - 'persisting' → 'finalizing' 合法:controller 在 registerInflight 后、 + * 'running' 置位前的窗口内就可能被卡死守卫强制收口。 + */ + +import type { ScheduleRunPhase } from '../types.js'; + +export const LEGAL_PHASE_TRANSITIONS: Readonly< + Record +> = Object.freeze({ + claiming: ['persisting'], + loading: ['persisting'], + persisting: ['running', 'finalizing'], + running: ['queued', 'finalizing'], + queued: ['running', 'cancelling', 'finalizing'], + cancelling: ['finalizing'], + finalizing: [], +}); + +/** 幂等重入(from === to)合法;其余按表判定。 */ +export function isLegalPhaseTransition(from: ScheduleRunPhase, to: ScheduleRunPhase): boolean { + if (from === to) return true; + return (LEGAL_PHASE_TRANSITIONS[from] ?? []).includes(to); +} diff --git a/packages/maker-scheduler/src/engine/scheduler.ts b/packages/maker-scheduler/src/engine/scheduler.ts index c9aee05f02b..39508cb3fd6 100644 --- a/packages/maker-scheduler/src/engine/scheduler.ts +++ b/packages/maker-scheduler/src/engine/scheduler.ts @@ -16,6 +16,7 @@ import type { PreRunHookRunResult, } from '../types.js'; import { SCRIPT_CAPABILITIES } from '../types.js'; +import { isLegalPhaseTransition } from './attemptLifecycle.js'; import type { ScheduleStorage } from '../interfaces/schedule-storage.js'; import type { ChildRunInput, ScheduleRunner } from '../interfaces/schedule-runner.js'; import type { Clock } from '../interfaces/clock.js'; @@ -1614,6 +1615,98 @@ export class Scheduler extends EventEmitter { } /** 在第一次 await 前同步登记一次槽位占用，并输出可配对的注册日志。 */ + /** + * 阶段转移的唯一写入口(#1016):合法性由 attemptLifecycle 的显式转移表判定, + * 非法转移**抛错**——「静默少做一件事」正是 #944 review 里同型出现四次的缺陷形态, + * 宁可响亮失败也不静默容忍。幂等重入(from === to)按 no-op 放行并返回 false + * (强制收口与迟到 settle 会各自把 attempt 置一次 'finalizing')。 + */ + private transitionAttempt( + attempt: InflightAttempt, + next: ScheduleRunPhase, + via: string, + ): boolean { + const from = attempt.phase; + if (from === next) return false; + if (!isLegalPhaseTransition(from, next)) { + throw new Error( + `scheduler: illegal attempt phase transition ${from} -> ${next} ` + + `(via ${via}, runId=${attempt.runId}, scheduleId=${attempt.scheduleId})`, + ); + } + attempt.phase = next; + if (next === 'finalizing' && attempt.finalizingSince === undefined) { + attempt.finalizingSince = this.clock.now(); + } + return true; + } + + /** + * 单一出口的「出口清单」矫正(#1016):attempt 删除时校验并清掉所有仍指向它的 + * 登记(controller / per-schedule 索引 / session 双向映射 / 静默标记)。这些登记 + * 本应由各路径自己收干净(unregisterInflight / 强制收口);此处发现残留说明某条 + * 出口路径漏了收口动作 —— 矫正之余响亮告警,让这类缺陷在日志/测试里直接可见, + * 而不是留成"槽位对不上 / 映射悬挂"的静默账。abandonedRuns 刻意不碰:它就是 + * 设计为跨 attempt 生命周期存活、由迟到 settle 消费的(见字段注释)。 + */ + private reapAttemptResiduals(runId: string, scheduleId: string): void { + const residuals: string[] = []; + if (this.inflightControllers.delete(runId)) residuals.push('controller'); + const set = this.inflightByschedule.get(scheduleId); + if (set?.delete(runId)) { + residuals.push('scheduleIndex'); + if (set.size === 0) this.inflightByschedule.delete(scheduleId); + } + const sessionId = this.runIdToSessionId.get(runId); + if (sessionId !== undefined) { + if (this.sessionIdToRunId.get(sessionId) === runId) this.sessionIdToRunId.delete(sessionId); + this.runIdToSessionId.delete(runId); + residuals.push('sessionMap'); + } + if (this.runIdToBoundSessionId.delete(runId)) residuals.push('boundSessionMap'); + if (this.silencedRuns.delete(runId)) residuals.push('silencedRuns'); + if (residuals.length > 0) { + this.logger?.warn?.( + 'scheduler: attempt exit found unreaped registrations (cleaned; a lifecycle path skipped its cleanup)', + { + schedulerInstanceId: this.schedulerInstanceId, + processId: this.processId, + runId, + scheduleId, + residuals, + }, + ); + } + } + + /** + * 登记一致性不变量(#1016):所有按 runId 键控的登记必须指向仍在账的 attempt。 + * 违反 = 某条出口漏了收口且 reap 也没兜住(理论不可达;可达即缺陷),抛错让 + * 单测与运行期都响亮失败。只在 begin(注册面唯一的扩张点)校验,O(登记数), + * 上限受并发闸门约束,代价可忽略。 + */ + private assertAttemptRegistryInvariants(): void { + for (const runId of this.inflightControllers.keys()) { + if (!this.inflightAttempts.has(runId)) { + throw new Error(`scheduler invariant violated: controller without attempt (runId=${runId})`); + } + } + for (const [scheduleId, runIds] of this.inflightByschedule) { + for (const runId of runIds) { + if (!this.inflightAttempts.has(runId)) { + throw new Error( + `scheduler invariant violated: schedule index entry without attempt (runId=${runId}, scheduleId=${scheduleId})`, + ); + } + } + } + for (const runId of this.runIdToSessionId.keys()) { + if (!this.inflightAttempts.has(runId)) { + throw new Error(`scheduler invariant violated: session map entry without attempt (runId=${runId})`); + } + } + } + private beginInflightAttempt( input: Omit, ): void { @@ -1621,6 +1714,7 @@ export class Scheduler extends EventEmitter { throw new Error(`duplicate scheduler run id: ${input.runId}`); } const before = this.inflightAttempts.size; + this.assertAttemptRegistryInvariants(); const startedAt = this.clock.now(); const attempt: InflightAttempt = { ...input, startedAt, lastProgressAt: startedAt }; this.inflightAttempts.set(input.runId, attempt); @@ -1644,10 +1738,7 @@ export class Scheduler extends EventEmitter { ): void { const current = this.inflightAttempts.get(runId); if (!current) return; - current.phase = phase; - if (phase === 'finalizing' && current.finalizingSince === undefined) { - current.finalizingSince = this.clock.now(); - } + this.transitionAttempt(current, phase, 'updateInflightAttempt'); if (schedule) { current.scheduleName = schedule.name; current.executionMode = schedule.executionMode ?? 'agent'; @@ -1672,6 +1763,7 @@ export class Scheduler extends EventEmitter { } const before = this.inflightAttempts.size; this.inflightAttempts.delete(runId); + this.reapAttemptResiduals(runId, attempt.scheduleId); const now = this.clock.now(); this.logger?.info?.('scheduler: in-flight run released', { schedulerInstanceId: this.schedulerInstanceId, @@ -2004,8 +2096,7 @@ export class Scheduler extends EventEmitter { return () => { const attempt = this.inflightAttempts.get(runId); if (!attempt) return; - if (attempt.phase === 'queued') return; - attempt.phase = 'queued'; + if (!this.transitionAttempt(attempt, 'queued', 'onQueueWaitStart')) return; attempt.lastProgressAt = this.clock.now(); this.logger?.info?.('scheduler: in-flight run entered pure queue wait (slot released)', { schedulerInstanceId: this.schedulerInstanceId, @@ -2041,7 +2132,7 @@ export class Scheduler extends EventEmitter { // 不会执行,所以不该占并发槽 —— 复位成 'running' 会让 slotsInUse 临时超过 // maxConcurrentRuns、UI 冒出 9/8,也与 endQueueWait 契约里"只复位记账"矛盾 // (review #944 第十五轮)。 - attempt.phase = 'cancelling'; + this.transitionAttempt(attempt, 'cancelling', 'endQueueWait'); attempt.lastProgressAt = this.clock.now(); this.emitRuntimeState(); return true; @@ -2057,7 +2148,7 @@ export class Scheduler extends EventEmitter { }); return false; } - attempt.phase = 'running'; + this.transitionAttempt(attempt, 'running', 'endQueueWait-reclaim'); attempt.lastProgressAt = this.clock.now(); this.logger?.info?.('scheduler: queued run reclaimed a slot', { schedulerInstanceId: this.schedulerInstanceId, @@ -2283,8 +2374,7 @@ export class Scheduler extends EventEmitter { // finishInflightAttempt 释放;真卡住就留在账上,由 logStorageStall 持续暴露。 this.abandonedRuns.add(runId); this.inflightControllers.delete(runId); - attempt.phase = 'finalizing'; - attempt.finalizingSince = now; + this.transitionAttempt(attempt, 'finalizing', 'force-release'); const set = this.inflightByschedule.get(scheduleId); if (set) { set.delete(runId); From b8007db068c5ac34f2d4e51f4a647595b2dddbe7 Mon Sep 17 00:00:00 2001 From: ficowang Date: Thu, 30 Jul 2026 16:44:50 +0800 Subject: [PATCH 2/7] =?UTF-8?q?fix(scheduler):=20=E5=BC=BA=E5=88=B6?= =?UTF-8?q?=E6=94=B6=E5=8F=A3=E5=90=8E=E7=9A=84=E8=BF=9F=E5=88=B0=E6=8E=92?= =?UTF-8?q?=E9=98=9F=E5=9B=9E=E8=B0=83=E6=8C=89=20no-op=20=E5=A4=84?= =?UTF-8?q?=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit review 反馈:强制收口把 attempt 置 finalizing 后,runner 的异步 continuation 仍可能调 onQueueWaitStart,原实现会抛非法转移错误——竞态 是预期而非状态机缺陷。与 endQueueWait 同款:attempt 已 finalizing 时 安静返回。新增用例:强制收口完成后迟到调用两个排队回调均不抛。 Signed-off-by: ficowang --- .../src/__tests__/scheduler.test.ts | 30 +++++++++++++++++++ .../maker-scheduler/src/engine/scheduler.ts | 5 +++- 2 files changed, 34 insertions(+), 1 deletion(-) diff --git a/packages/maker-scheduler/src/__tests__/scheduler.test.ts b/packages/maker-scheduler/src/__tests__/scheduler.test.ts index 0daa0549eef..0bf77e5e8df 100644 --- a/packages/maker-scheduler/src/__tests__/scheduler.test.ts +++ b/packages/maker-scheduler/src/__tests__/scheduler.test.ts @@ -3926,3 +3926,33 @@ describe('Scheduler: attempt 生命周期状态机(#1016)', () => { await h.scheduler.stop(); }); }); + + it('强制收口后 runner 迟到调用 onQueueWaitStart → no-op,不抛非法转移(#1016 review)', async () => { + vi.useFakeTimers(); + try { + let ctxRef: FireContext | undefined; + const h = makeHarness({ + runStallMs: 60_000, + runStallAbortGraceMs: 30_000, + runnerImpl: (_s, ctx) => + new Promise(() => { + ctxRef = ctx; + }), + }); + const sch = await h.scheduler.create({ ...baseInput, intervalMs: 3_600_000 }); + h.clock.advance(3_600_000); + void h.scheduler.tick(); + await vi.waitFor(() => expect(h.scheduler.getRuntimeSnapshot().slotsInUse).toBe(1)); + h.clock.advance(60_001); + await vi.advanceTimersByTimeAsync(RUN_HEARTBEAT_INTERVAL_MS); + h.clock.advance(30_001); + await vi.advanceTimersByTimeAsync(RUN_HEARTBEAT_INTERVAL_MS); + await vi.waitFor(() => expect(h.scheduler.getRuntimeSnapshot().inFlight).toBe(0)); + // 强制收口已完成,runner 的 continuation 迟到调排队回调:必须是安静的 no-op。 + expect(() => ctxRef?.onQueueWaitStart?.()).not.toThrow(); + expect(() => ctxRef?.endQueueWait?.(true)).not.toThrow(); + await h.scheduler.stop(); + } finally { + vi.useRealTimers(); + } + }); diff --git a/packages/maker-scheduler/src/engine/scheduler.ts b/packages/maker-scheduler/src/engine/scheduler.ts index 39508cb3fd6..986bf8b7cb0 100644 --- a/packages/maker-scheduler/src/engine/scheduler.ts +++ b/packages/maker-scheduler/src/engine/scheduler.ts @@ -2095,7 +2095,10 @@ export class Scheduler extends EventEmitter { private buildOnQueueWaitStart(runId: string): () => void { return () => { const attempt = this.inflightAttempts.get(runId); - if (!attempt) return; + // 迟到回调竞态是**预期**而非状态机缺陷:强制收口把 attempt 置 'finalizing' 后, + // runner 的异步 continuation 仍可能调进来 —— 与 endQueueWait 同款按 no-op 处理, + // 不让正常竞态伪装成非法转移错误(review 反馈)。 + if (!attempt || attempt.phase === 'finalizing') return; if (!this.transitionAttempt(attempt, 'queued', 'onQueueWaitStart')) return; attempt.lastProgressAt = this.clock.now(); this.logger?.info?.('scheduler: in-flight run entered pure queue wait (slot released)', { From 96c947252efa8caed525f1f46594352d33b81172 Mon Sep 17 00:00:00 2001 From: ficowang Date: Thu, 30 Jul 2026 17:02:48 +0800 Subject: [PATCH 3/7] =?UTF-8?q?fix(maker-scheduler):=20=E8=BF=9F=E5=88=B0?= =?UTF-8?q?=20onTurnActive/onSessionBound=20=E5=AE=88=E5=8D=AB=E4=B8=8E?= =?UTF-8?q?=E4=B8=8D=E5=8F=98=E9=87=8F=E8=A1=A5=E5=85=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit review 反馈第二轮: - codex P1: 强制收口删除 attempt 后, runner continuation 迟到上报 onTurnActive 会往 session 映射写悬挂登记, 下一次 beginInflightAttempt 的不变量断言将响亮抛错。补与 onQueueWaitStart 同款的迟到守卫 (attempt 不在账或已 finalizing → 安静 no-op)。 - onSessionBound 同类迟到写入 runIdToBoundSessionId 一并守卫, 否则 下述不变量扩展会被合法竞态触发。 - copilot: assertAttemptRegistryInvariants 补齐 runIdToBoundSessionId 与 silencedRuns 两类 runId 键控登记的覆盖。 - copilot: 孤立 JSDoc 挪回 beginInflightAttempt 头上。 新增迟到 onTurnActive/onSessionBound 用例: 强制收口后调用不留悬挂 登记, 后续 fire 的 begin 断言不抛。 Signed-off-by: ficowang --- .../src/__tests__/scheduler.test.ts | 38 +++++++++++++++++++ .../maker-scheduler/src/engine/scheduler.ts | 23 ++++++++++- 2 files changed, 60 insertions(+), 1 deletion(-) diff --git a/packages/maker-scheduler/src/__tests__/scheduler.test.ts b/packages/maker-scheduler/src/__tests__/scheduler.test.ts index 0bf77e5e8df..69b10e28098 100644 --- a/packages/maker-scheduler/src/__tests__/scheduler.test.ts +++ b/packages/maker-scheduler/src/__tests__/scheduler.test.ts @@ -3956,3 +3956,41 @@ describe('Scheduler: attempt 生命周期状态机(#1016)', () => { vi.useRealTimers(); } }); + + it('强制收口后迟到的 onTurnActive/onSessionBound 不留悬挂登记(#1016 review)', async () => { + vi.useFakeTimers(); + try { + let ctxRef: FireContext | undefined; + const h = makeHarness({ + runStallMs: 60_000, + runStallAbortGraceMs: 30_000, + runnerImpl: (_s, ctx) => + new Promise(() => { + ctxRef = ctx; + }), + }); + const sch = await h.scheduler.create({ ...baseInput, intervalMs: 3_600_000 }); + h.clock.advance(3_600_000); + void h.scheduler.tick(); + await vi.waitFor(() => expect(h.scheduler.getRuntimeSnapshot().slotsInUse).toBe(1)); + h.clock.advance(60_001); + await vi.advanceTimersByTimeAsync(RUN_HEARTBEAT_INTERVAL_MS); + h.clock.advance(30_001); + await vi.advanceTimersByTimeAsync(RUN_HEARTBEAT_INTERVAL_MS); + await vi.waitFor(() => expect(h.scheduler.getRuntimeSnapshot().inFlight).toBe(0)); + // attempt 已删并 reap:迟到的 turn-active / session-bound 上报必须整体 no-op, + // 不写 session 映射 / 绑定映射(悬挂登记会让下一次 begin 的不变量断言抛错)。 + expect(() => ctxRef?.onTurnActive?.('sess-late-turn')).not.toThrow(); + await ctxRef?.onSessionBound?.('sess-late-bind'); + expect(h.scheduler.resolveInflightRunForSession('sess-late-turn')).toBeUndefined(); + // 下一轮 fire 的 beginInflightAttempt 会跑 assertAttemptRegistryInvariants + // (含 bound-session / silenced 覆盖)——迟到写入若真落了账,这里会响亮抛错。 + h.clock.advance(3_600_000); + void h.scheduler.tick(); + await vi.waitFor(() => expect(h.scheduler.getRuntimeSnapshot().slotsInUse).toBe(1)); + expect(sch.id).toBeTruthy(); + await h.scheduler.stop(); + } finally { + vi.useRealTimers(); + } + }); diff --git a/packages/maker-scheduler/src/engine/scheduler.ts b/packages/maker-scheduler/src/engine/scheduler.ts index 986bf8b7cb0..dd04c0c1b96 100644 --- a/packages/maker-scheduler/src/engine/scheduler.ts +++ b/packages/maker-scheduler/src/engine/scheduler.ts @@ -1614,7 +1614,6 @@ export class Scheduler extends EventEmitter { } } - /** 在第一次 await 前同步登记一次槽位占用，并输出可配对的注册日志。 */ /** * 阶段转移的唯一写入口(#1016):合法性由 attemptLifecycle 的显式转移表判定, * 非法转移**抛错**——「静默少做一件事」正是 #944 review 里同型出现四次的缺陷形态, @@ -1705,8 +1704,21 @@ export class Scheduler extends EventEmitter { throw new Error(`scheduler invariant violated: session map entry without attempt (runId=${runId})`); } } + for (const runId of this.runIdToBoundSessionId.keys()) { + if (!this.inflightAttempts.has(runId)) { + throw new Error( + `scheduler invariant violated: bound-session map entry without attempt (runId=${runId})`, + ); + } + } + for (const runId of this.silencedRuns) { + if (!this.inflightAttempts.has(runId)) { + throw new Error(`scheduler invariant violated: silenced mark without attempt (runId=${runId})`); + } + } } + /** 在第一次 await 前同步登记一次槽位占用，并输出可配对的注册日志。 */ private beginInflightAttempt( input: Omit, ): void { @@ -2039,6 +2051,10 @@ export class Scheduler extends EventEmitter { private buildOnSessionBound(scheduleId: string, runId: string): (sessionId: string) => Promise { return async (sessionId: string) => { if (!sessionId) return; + // 与 onTurnActive 同款迟到守卫:run 已被强制收口时不再写绑定映射(悬挂 + // 登记会触发 begin 的不变量断言),也不再往已定案 failed 的 run 行补状态。 + const attempt = this.inflightAttempts.get(runId); + if (!attempt || attempt.phase === 'finalizing') return; try { this.runIdToBoundSessionId.set(runId, sessionId); await this.storage.updateRun(runId, { sessionId }); @@ -2079,6 +2095,11 @@ export class Scheduler extends EventEmitter { private buildOnTurnActive(runId: string): (sessionId: string) => void { return (sessionId: string) => { if (!sessionId) return; + const attempt = this.inflightAttempts.get(runId); + // 迟到回调竞态同 onQueueWaitStart:强制收口删除 attempt 后 runner 的 + // continuation 仍可能报 turn active,此时写映射会留下悬挂登记,被下一次 + // begin 的 assertAttemptRegistryInvariants 当成缺陷抛错(codex review P1)。 + if (!attempt || attempt.phase === 'finalizing') return; this.sessionIdToRunId.set(sessionId, runId); this.runIdToSessionId.set(runId, sessionId); }; From 9df95c3ebc9e894d4db9ecd421cb346ce93d2853 Mon Sep 17 00:00:00 2001 From: ficowang Date: Thu, 30 Jul 2026 17:17:39 +0800 Subject: [PATCH 4/7] =?UTF-8?q?test(maker-scheduler):=20=E5=8E=BB=E6=8E=89?= =?UTF-8?q?=E8=BF=9F=E5=88=B0=E6=8E=92=E9=98=9F=E5=9B=9E=E8=B0=83=E7=94=A8?= =?UTF-8?q?=E4=BE=8B=E4=B8=AD=E6=9C=AA=E4=BD=BF=E7=94=A8=E7=9A=84=20sch=20?= =?UTF-8?q?=E7=BB=91=E5=AE=9A?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CI 的 scheduler 架构守卫(lint)拦下 no-unused-vars:上一条迟到回调用例 只需要 create 的副作用,不需要返回值。 Signed-off-by: ficowang --- packages/maker-scheduler/src/__tests__/scheduler.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/maker-scheduler/src/__tests__/scheduler.test.ts b/packages/maker-scheduler/src/__tests__/scheduler.test.ts index 69b10e28098..79ddd8b0cc5 100644 --- a/packages/maker-scheduler/src/__tests__/scheduler.test.ts +++ b/packages/maker-scheduler/src/__tests__/scheduler.test.ts @@ -3939,7 +3939,7 @@ describe('Scheduler: attempt 生命周期状态机(#1016)', () => { ctxRef = ctx; }), }); - const sch = await h.scheduler.create({ ...baseInput, intervalMs: 3_600_000 }); + await h.scheduler.create({ ...baseInput, intervalMs: 3_600_000 }); h.clock.advance(3_600_000); void h.scheduler.tick(); await vi.waitFor(() => expect(h.scheduler.getRuntimeSnapshot().slotsInUse).toBe(1)); From 2982346522249ea5f33e697bc60519a02bc57419 Mon Sep 17 00:00:00 2001 From: ficowang Date: Thu, 30 Jul 2026 17:24:43 +0800 Subject: [PATCH 5/7] =?UTF-8?q?fix(maker-scheduler):=20stop()=20=E4=B8=80?= =?UTF-8?q?=E5=B9=B6=E6=B8=85=E7=A9=BA=20silencedRuns?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit codex review P1:stop() 清 inflightAttempts 与其余 runId 键控映射时漏了 silencedRuns。留着的标记会让同实例后续第一次 beginInflightAttempt 的 不变量断言把它当悬挂登记抛错。语义上与 silenceRun 文档一致 —— 静默 标记丢失的安全方向就是照常通知。 新增用例:静默 run 执行中 stop 后,标记被清、再 runNow 不被断言误杀。 Signed-off-by: ficowang --- .../src/__tests__/scheduler.test.ts | 20 +++++++++++++++++++ .../maker-scheduler/src/engine/scheduler.ts | 4 ++++ 2 files changed, 24 insertions(+) diff --git a/packages/maker-scheduler/src/__tests__/scheduler.test.ts b/packages/maker-scheduler/src/__tests__/scheduler.test.ts index 79ddd8b0cc5..318b09f3d3c 100644 --- a/packages/maker-scheduler/src/__tests__/scheduler.test.ts +++ b/packages/maker-scheduler/src/__tests__/scheduler.test.ts @@ -3994,3 +3994,23 @@ describe('Scheduler: attempt 生命周期状态机(#1016)', () => { vi.useRealTimers(); } }); + + it('stop() 清 silencedRuns:静默 run 执行中停机后再 runNow 不被不变量断言误杀(#1016 review)', async () => { + const h = makeHarness({ + runnerImpl: () => new Promise(() => {}), + }); + const sch = await h.scheduler.create({ ...baseInput, silentWhenIdle: true }); + const first = h.scheduler.runNow(sch.id); + first.catch(() => {}); + await vi.waitFor(() => expect(h.scheduler.getRuntimeSnapshot().inFlight).toBe(1)); + const [running] = await h.scheduler.listRuns(sch.id); + expect(h.scheduler.isRunSilenced(running.id)).toBe(true); + await h.scheduler.stop(); + // stop 清空 attempts 的同时必须一并清 silencedRuns:留着的话,同实例后续第一次 + // beginInflightAttempt 的不变量断言会把它当悬挂登记抛错(codex review P1)。 + expect(h.scheduler.isRunSilenced(running.id)).toBe(false); + const second = h.scheduler.runNow(sch.id); + second.catch(() => {}); + await vi.waitFor(() => expect(h.scheduler.getRuntimeSnapshot().inFlight).toBe(1)); + await h.scheduler.stop(); + }); diff --git a/packages/maker-scheduler/src/engine/scheduler.ts b/packages/maker-scheduler/src/engine/scheduler.ts index dd04c0c1b96..c916133a9f2 100644 --- a/packages/maker-scheduler/src/engine/scheduler.ts +++ b/packages/maker-scheduler/src/engine/scheduler.ts @@ -493,6 +493,10 @@ export class Scheduler extends EventEmitter { this.sessionIdToRunId.clear(); this.runIdToSessionId.clear(); this.runIdToBoundSessionId.clear(); + // silencedRuns 与上面同为 runId 键控登记,必须随 stop 一起清:留着会让重启后 + // 第一次 begin 的不变量断言把它当悬挂登记抛错(codex review P1)。语义上也与 + // silenceRun 文档一致 —— 标记丢失的安全方向就是照常通知。 + this.silencedRuns.clear(); this.activeSchedules.clear(); this.started = false; this.emitRuntimeState(); From b59b1444536a3ce0e1ee8d1a08fefe6dcef24b76 Mon Sep 17 00:00:00 2001 From: ficowang Date: Thu, 30 Jul 2026 17:40:38 +0800 Subject: [PATCH 6/7] =?UTF-8?q?fix(maker-scheduler):=20stop()=20=E7=AB=9E?= =?UTF-8?q?=E6=80=81=E4=B8=8B=E5=89=8D=E7=BD=AE=20await=20=E6=81=A2?= =?UTF-8?q?=E5=A4=8D=E7=9A=84=20continuation=20=E4=B8=8D=E5=86=8D=E7=99=BB?= =?UTF-8?q?=E8=AE=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit codex review P1:stop() 打在 fireOneInner(claimDueFire/insertRun)或 runNowInner(storage.get/insertRun/update)的前置 await 期间时,attempt 被清 但 continuation 还没有 controller、无从 abort;恢复后照常 registerInflight 会留下没有 attempt 的悬挂 controller/索引,此后同实例每次 begin 都被 不变量断言拦下。 两条路径在 registerInflight 前补 attempt 在账检查:fireOne 放弃本轮 (run 行交给下次 start() 僵尸清扫收敛,认领走崩溃恢复既有归一);runNow 按调用方显式动作的契约抛错,不静默吞掉。 新增用例:insertRun 卡住期间 stop(),恢复后 runNow 拒绝并抛错、无悬挂 登记,同实例再 runNow 正常完成。 Signed-off-by: ficowang --- .../src/__tests__/scheduler.test.ts | 35 +++++++++++++++++++ .../maker-scheduler/src/engine/scheduler.ts | 26 ++++++++++++++ 2 files changed, 61 insertions(+) diff --git a/packages/maker-scheduler/src/__tests__/scheduler.test.ts b/packages/maker-scheduler/src/__tests__/scheduler.test.ts index 318b09f3d3c..fdbc73f4a60 100644 --- a/packages/maker-scheduler/src/__tests__/scheduler.test.ts +++ b/packages/maker-scheduler/src/__tests__/scheduler.test.ts @@ -4014,3 +4014,38 @@ describe('Scheduler: attempt 生命周期状态机(#1016)', () => { await vi.waitFor(() => expect(h.scheduler.getRuntimeSnapshot().inFlight).toBe(1)); await h.scheduler.stop(); }); + + it('stop() 打在前置 await 期间:恢复的 continuation 不登记悬挂 controller(#1016 review)', async () => { + // stop() 清 attempt 时 continuation 还没有 controller,无从 abort;恢复后若照常 + // registerInflight,controller/索引就成了没有 attempt 的悬挂登记,此后同实例每次 + // begin 都被不变量断言拦下(codex review P1)。守卫应放弃本轮并(runNow 契约)抛错。 + const storage = new InMemoryStorage(); + let releaseInsert: (() => void) | null = null; + let gated = true; + const realInsertRun = storage.insertRun.bind(storage); + storage.insertRun = (run: ScheduleRun) => { + if (!gated) return realInsertRun(run); + return new Promise((resolve) => { + releaseInsert = () => resolve(realInsertRun(run)); + }); + }; + const h = makeHarness({ + storage, + runnerImpl: async () => ({ sessionId: 'sess-after-stop' }), + }); + const sch = await h.scheduler.create({ ...baseInput }); + const first = h.scheduler.runNow(sch.id); + const firstOutcome = first.then( + () => 'resolved', + (e) => String(e), + ); + await vi.waitFor(() => expect(releaseInsert).not.toBeNull()); + await h.scheduler.stop(); + gated = false; + releaseInsert!(); + expect(await firstOutcome).toMatch(/stopped while starting runNow/); + // 无悬挂登记:同实例再 runNow,begin 的不变量断言不抛,run 正常收尾。 + const second = await h.scheduler.runNow(sch.id); + expect(second.runId).toBeTruthy(); + await h.scheduler.stop(); + }); diff --git a/packages/maker-scheduler/src/engine/scheduler.ts b/packages/maker-scheduler/src/engine/scheduler.ts index c916133a9f2..a23d948a9a6 100644 --- a/packages/maker-scheduler/src/engine/scheduler.ts +++ b/packages/maker-scheduler/src/engine/scheduler.ts @@ -658,6 +658,20 @@ export class Scheduler extends EventEmitter { this.logger?.error?.('insertRun failed', err); return; } + // stop() 竞态守卫(codex review P1):前置 await(claimDueFire/insertRun)期间 + // stop() 会清掉 attempt,且此时还没有 controller 可 abort 本 continuation。恢复后 + // attempt 已不在账就不得再登记 controller/索引——悬挂登记会让停机后同实例的每次 + // begin 都被不变量断言拦下。放弃本轮:刚插入的 run 行与其他 stop 释放的 run 同样 + // 交给下次 start() 的僵尸清扫收敛成 interrupted,认领走崩溃恢复的既有归一路径。 + if (!this.inflightAttempts.has(runId)) { + this.logger?.info?.('scheduler: attempt released during pre-register await (stopped); dropping fire', { + schedulerInstanceId: this.schedulerInstanceId, + processId: this.processId, + runId, + scheduleId: schedule.id, + }); + return; + } const controller = new AbortController(); this.registerInflight(schedule.id, runId, controller); if (schedule.silentWhenIdle) { @@ -942,6 +956,18 @@ export class Scheduler extends EventEmitter { await this.storage.update(schedule.id, { lastFiredAt: firedAt }); const cached = this.activeSchedules.get(schedule.id); if (cached) this.activeSchedules.set(schedule.id, { ...cached, lastFiredAt: firedAt }); + // stop() 竞态守卫,与 fireOneInner 同款(codex review P1):storage.get/insertRun/ + // update 期间 stop() 清掉 attempt 后不得再登记 controller/索引。runNow 契约上 + // 以抛错收场(调用方显式动作,静默吞掉会让"没跑"看起来像"跑了")。 + if (!this.inflightAttempts.has(runId)) { + this.logger?.info?.('scheduler: attempt released during pre-register await (stopped); dropping runNow', { + schedulerInstanceId: this.schedulerInstanceId, + processId: this.processId, + runId, + scheduleId: schedule.id, + }); + throw new Error(`scheduler stopped while starting runNow (scheduleId=${schedule.id})`); + } const controller = new AbortController(); this.registerInflight(schedule.id, runId, controller); if (schedule.silentWhenIdle) { From e57b26cd17d12a2b75bd16e7cf290abfe93e30bc Mon Sep 17 00:00:00 2001 From: ficowang Date: Fri, 31 Jul 2026 22:29:55 +0800 Subject: [PATCH 7/7] =?UTF-8?q?test(maker-scheduler):=20=E4=BF=AE=20stop?= =?UTF-8?q?=20=E7=AB=9E=E6=80=81=E7=94=A8=E4=BE=8B=E7=9A=84=20mock=20?= =?UTF-8?q?=E8=BF=94=E5=9B=9E=E7=B1=BB=E5=9E=8B=E4=B8=8E=20describe=20?= =?UTF-8?q?=E4=BD=9C=E7=94=A8=E5=9F=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit auto-review P1:该用例把 storage.insertRun mock 成 Promise,真实 签名是 Promise,tsc --noEmit(包内 build 脚本)报 TS2322/ TS2345——vitest 经 esbuild 跳过类型检查所以测试仍绿。改为 Promise,resolve 直接透传 realInsertRun 结果。 顺修 P2:describe 在原有用例后提前闭合,后续 4 个追加用例掉到模块顶层 ——闭合挪到文件末尾,用例回到 describe 作用域内。 Signed-off-by: ficowang --- packages/maker-scheduler/src/__tests__/scheduler.test.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/packages/maker-scheduler/src/__tests__/scheduler.test.ts b/packages/maker-scheduler/src/__tests__/scheduler.test.ts index fdbc73f4a60..1b67744ff01 100644 --- a/packages/maker-scheduler/src/__tests__/scheduler.test.ts +++ b/packages/maker-scheduler/src/__tests__/scheduler.test.ts @@ -3925,7 +3925,6 @@ describe('Scheduler: attempt 生命周期状态机(#1016)', () => { ).toBe(false); await h.scheduler.stop(); }); -}); it('强制收口后 runner 迟到调用 onQueueWaitStart → no-op,不抛非法转移(#1016 review)', async () => { vi.useFakeTimers(); @@ -4025,7 +4024,7 @@ describe('Scheduler: attempt 生命周期状态机(#1016)', () => { const realInsertRun = storage.insertRun.bind(storage); storage.insertRun = (run: ScheduleRun) => { if (!gated) return realInsertRun(run); - return new Promise((resolve) => { + return new Promise((resolve) => { releaseInsert = () => resolve(realInsertRun(run)); }); }; @@ -4049,3 +4048,4 @@ describe('Scheduler: attempt 生命周期状态机(#1016)', () => { expect(second.runId).toBeTruthy(); await h.scheduler.stop(); }); +});