refactor(scheduler): attempt 阶段机显式化——统一转移入口 + 单一出口清单

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);
+    }
+  });
+});

packages/maker-scheduler/src/__tests__/scheduler.test.ts

@@ -3854,3 +3854,105 @@
     }
   });
 });
+
+// ── #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<FireResult>((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<FireResult>((_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();
+  });
+});
+
+  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<FireResult>(() => {
+            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();
+    }
+  });

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<ScheduleRunPhase, readonly ScheduleRunPhase[]>
+> = 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);
+}

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,13 +1615,106 @@ 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<SchedulerInflightRun, 'startedAt' | 'lastProgressAt'>,
   ): void {
     if (this.inflightAttempts.has(input.runId)) {
       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,
@@ -2003,9 +2095,11 @@ export class Scheduler extends EventEmitter {
   private buildOnQueueWaitStart(runId: string): () => void {
     return () => {
       const attempt = this.inflightAttempts.get(runId);
-      if (!attempt) return;
-      if (attempt.phase === 'queued') return;
-      attempt.phase = 'queued';
+      // 迟到回调竞态是**预期**而非状态机缺陷:强制收口把 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)', {
         schedulerInstanceId: this.schedulerInstanceId,
@@ -2041,7 +2135,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 +2151,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 +2377,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);