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,198 @@ 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<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;
+          }),
+      });
+      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();
+    }
+  });
+
+  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<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));
+      // 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();
+    }
+  });
+
+  it('stop() 清 silencedRuns:静默 run 执行中停机后再 runNow 不被不变量断言误杀(#1016 review)', async () => {
+    const h = makeHarness({
+      runnerImpl: () => new Promise<FireResult>(() => {}),
+    });
+    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();
+  });
+
+  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<void>((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();
+  });

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