From 17bf0f4a29d3ab9f3adf39ad9f4848c8cbe87456 Mon Sep 17 00:00:00 2001 From: Raghav Chari Date: Thu, 2 Jul 2026 20:59:00 -0400 Subject: [PATCH 1/2] =?UTF-8?q?feat(1.1):=20Scheduler=20=E2=80=94=20serial?= =?UTF-8?q?=20run=20queue=20built=20to=20the=20ratified=20Executor=20contr?= =?UTF-8?q?act=20(#56)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `enqueue(spec, {concurrent?}) → ScheduledRun{queueId, handle: Promise, cancel}` plus a multi-consumer lifecycle stream (queued/started/finished/cancelled/error) for RunsManager/StatusBar (1.2). Lives in @amicode/amico-run (node-only, no vscode) beside the LocalExecutor it drives. Built TO the Track C contract (ratified 2026-07-02) so Δ8's RemoteExecutor drops in with zero reshape: - S12: enqueue resolves to the executor's RunHandle UNTOUCHED (identity passthrough, pinned by test) — downstream never sees an executor type. - (b) abort() is a request, not a kill: the pump advances ONLY when `finished` resolves; a post-abort() run still holds the queue (pinned by test). - (c) per-executor warming budget: the Scheduler owns NO timers — structurally pinned (test greps the source for setTimeout/setInterval). - (d) `finished` never rejects per contract; a rogue rejection is survived (error event) rather than wedging every queued run. Semantics: strictly serial; cancel() dequeues only pre-start (a live run is stopped via RunHandle.abort(), never the queue); a submit() ConfigError rejects that entry's handle, emits `error`, and the queue advances; `concurrent: true` is the NAMED Phase-4 seam — rejected loudly (ConfigError) instead of silently serializing. Listener errors are isolated from the pump. TDD: 12 tests (RED first) — serial ordering, S12 identity, opts passthrough, abort≠terminated, lifecycle sequence with positions, cancel pre/post start, ConfigError advance, the concurrent seam, multi-listener + dispose, throwing listener, and the no-timers structural pin. Repo suite green (schema 34, amico-run 59, extension 80); build + typecheck clean. Closes #56. Co-Authored-By: Claude Fable 5 --- packages/amico-run/src/index.ts | 1 + packages/amico-run/src/scheduler.ts | 156 +++++++++++++++++ packages/amico-run/test/scheduler.test.ts | 204 ++++++++++++++++++++++ 3 files changed, 361 insertions(+) create mode 100644 packages/amico-run/src/scheduler.ts create mode 100644 packages/amico-run/test/scheduler.test.ts diff --git a/packages/amico-run/src/index.ts b/packages/amico-run/src/index.ts index f8af7d5f..7fd1e8e4 100644 --- a/packages/amico-run/src/index.ts +++ b/packages/amico-run/src/index.ts @@ -4,3 +4,4 @@ export * from './run_dir.js' export * from './schemas.js' export * from './event_queue.js' export * from './local_executor.js' +export * from './scheduler.js' diff --git a/packages/amico-run/src/scheduler.ts b/packages/amico-run/src/scheduler.ts new file mode 100644 index 00000000..b892b1b4 --- /dev/null +++ b/packages/amico-run/src/scheduler.ts @@ -0,0 +1,156 @@ +import { ConfigError, type Executor, type RunHandle, type RunStatus, type SubmitOpts } from './types.js' + +// ============================================================================ +// Scheduler (Phase 1.1, #56) — a serial run queue built TO the ratified +// Executor contract (Track C spec, locked 2026-07-02), so the cloud +// RemoteExecutor (Δ8/#32) drops in with zero reshape: +// +// - S12: downstream (RunsManager / Inspector / Catalog) sees ONLY the +// executor's RunHandle — enqueue() resolves to it untouched; nothing here +// branches on executor type. +// - (b) abort() is a REQUEST, not a kill: the queue advances ONLY when a +// run's `finished` resolves (a FINISHED — or executor-inferred terminal — +// landed). Post-abort() the run is still live; the Scheduler never treats +// abort as terminal. +// - (c) per-executor warming budget: the Scheduler owns NO timers. However +// long a run takes to warm/finish (remote cold-start ≫ local seconds) is +// between the executor and its handle; the queue just awaits `finished`. +// - (d) terminal resolution is the executor's job (`finished` never rejects +// per the contract); the Scheduler defensively survives a rogue rejection +// rather than wedging the queue. +// +// Serial by default; `{concurrent: true}` is the NAMED Phase-4 seam (§4.2) — +// rejected loudly today so nothing silently serializes when callers expect a +// parallel lane later. +// ============================================================================ + +/** What to run when this entry reaches the head of the queue. */ +export interface SubmitSpec { + scriptPath: string + /** Passed to Executor.submit verbatim (lab pointer, runsRoot, julia opts…). */ + opts?: SubmitOpts +} + +export interface EnqueueOpts { + /** Phase-4 seam (opt-in parallel lane) — NOT implemented; throws ConfigError. */ + concurrent?: boolean +} + +/** Run lifecycle the RunsManager / StatusBar consume (1.2). `queueId` is the + * Scheduler's own id (assigned at enqueue, before any run exists); `runId` + * appears once the executor has admitted the run. */ +export type SchedulerEvent = + | { kind: 'queued'; queueId: string; position: number } + | { kind: 'started'; queueId: string; runId: string; runDir: string } + | { kind: 'finished'; queueId: string; runId: string; status: RunStatus; exitCode: number } + | { kind: 'cancelled'; queueId: string } + | { kind: 'error'; queueId: string; message: string } + +export interface ScheduledRun { + queueId: string + /** Resolves with the executor's RunHandle when this entry reaches the head + * of the queue and submit() succeeds. Rejects if the entry is cancelled + * before starting, or if submit() throws (e.g. ConfigError). */ + handle: Promise + /** Dequeue BEFORE start: true if the entry was still queued (it will never + * run), false once started — a live run is stopped via RunHandle.abort() + * (a request, per contract (b)), never via the queue. */ + cancel(): boolean +} + +interface Entry { + queueId: string + spec: SubmitSpec + resolve: (h: RunHandle) => void + reject: (e: Error) => void +} + +export class Scheduler { + private readonly queue: Entry[] = [] + private running = false + private nextId = 1 + private readonly listeners = new Set<(e: SchedulerEvent) => void>() + + constructor(private readonly executor: Executor) {} + + /** Subscribe to lifecycle events. Returns a dispose function. Multi-consumer + * (RunsManager + StatusBar); a throwing listener is isolated. */ + onEvent(listener: (e: SchedulerEvent) => void): () => void { + this.listeners.add(listener) + return () => { this.listeners.delete(listener) } + } + + /** Queued + running entries — 0 means an enqueue() would start immediately. */ + get depth(): number { + return this.queue.length + (this.running ? 1 : 0) + } + + enqueue(spec: SubmitSpec, opts: EnqueueOpts = {}): ScheduledRun { + if (opts.concurrent) { + throw new ConfigError('Scheduler: the parallel lane (concurrent: true) is deferred to Phase 4 — runs are serial') + } + const queueId = `q${this.nextId++}` + let resolve!: (h: RunHandle) => void + let reject!: (e: Error) => void + const handle = new Promise((res, rej) => { resolve = res; reject = rej }) + // The Scheduler itself observes failures (error event) — callers that only + // consume events must not trip an unhandled-rejection on the same promise. + handle.catch(() => {}) + const entry: Entry = { queueId, spec, resolve, reject } + this.queue.push(entry) + this.emit({ kind: 'queued', queueId, position: this.queue.length - 1 + (this.running ? 1 : 0) }) + void this.pump() + return { + queueId, + handle, + cancel: (): boolean => { + const i = this.queue.indexOf(entry) + if (i === -1) return false // already started (or done) — abort via the handle + this.queue.splice(i, 1) + this.emit({ kind: 'cancelled', queueId }) + entry.reject(new Error(`Scheduler: ${queueId} cancelled before start`)) + return true + }, + } + } + + // -------- internal -------- + + private emit(e: SchedulerEvent): void { + for (const l of this.listeners) { + try { l(e) } catch { /* a bad listener must not wedge the pump */ } + } + } + + /** The serial pump: one entry at a time; advances ONLY on `finished` + * resolution (contract (b) — never on abort(), which is just a request). */ + private async pump(): Promise { + if (this.running) return + const entry = this.queue.shift() + if (!entry) return + this.running = true + try { + let handle: RunHandle + try { + handle = await this.executor.submit(entry.spec.scriptPath, entry.spec.opts) + } catch (e) { + const err = e instanceof Error ? e : new Error(String(e)) + this.emit({ kind: 'error', queueId: entry.queueId, message: err.message }) + entry.reject(err) + return // finally advances the queue — a config failure must not wedge it + } + this.emit({ kind: 'started', queueId: entry.queueId, runId: handle.runId, runDir: handle.runDir }) + entry.resolve(handle) + try { + const fin = await handle.finished // contract: never rejects… + this.emit({ kind: 'finished', queueId: entry.queueId, runId: handle.runId, status: fin.status, exitCode: fin.exitCode }) + } catch (e) { + // …but a rogue executor breaking that must not deadlock every queued run. + this.emit({ kind: 'error', queueId: entry.queueId, message: `finished rejected: ${(e as Error).message}` }) + } + } finally { + this.running = false + void this.pump() // next entry, if any + } + } +} diff --git a/packages/amico-run/test/scheduler.test.ts b/packages/amico-run/test/scheduler.test.ts new file mode 100644 index 00000000..e00e5187 --- /dev/null +++ b/packages/amico-run/test/scheduler.test.ts @@ -0,0 +1,204 @@ +import { describe, it, expect } from 'vitest' +import { Scheduler, type SchedulerEvent } from '../src/scheduler.js' +import { ConfigError, type Executor, type Finished, type RunEvent, type RunHandle, type SubmitOpts } from '../src/types.js' +import { EventQueue } from '../src/event_queue.js' + +// 1.1 Scheduler (#56) — serial queue built TO the ratified Executor contract +// (Track C spec, locked 2026-07-02). The load-bearing behaviors under test: +// - serial: entry N+1 submits only after entry N's `finished` RESOLVES; +// - (b) abort() is a REQUEST, not a kill — post-abort() the run is still +// alive and the queue must NOT advance until `finished` lands; +// - (c) no warming timeout — the Scheduler owns no timers at all; +// - S12 — downstream sees only the executor's RunHandle, passed through. + +/** Controllable fake executor: each submit() returns a handle whose `finished` + * the TEST resolves. Records submit order/args. */ +class FakeExecutor implements Executor { + submits: Array<{ scriptPath: string; opts?: SubmitOpts }> = [] + handles: Array<{ handle: RunHandle; finish: (f: Finished) => void; aborted: boolean[] }> = [] + /** scripts whose submit() should throw ConfigError */ + failFor = new Set() + + async submit(scriptPath: string, opts?: SubmitOpts): Promise { + this.submits.push({ scriptPath, opts }) + if (this.failFor.has(scriptPath)) throw new ConfigError(`bad config: ${scriptPath}`) + const n = this.submits.length + let finish!: (f: Finished) => void + const finished = new Promise(r => { finish = r }) + const aborted: boolean[] = [] + const handle: RunHandle = { + runId: `run-${n}`, + runDir: `/runs/run-${n}`, + events: new EventQueue(), + finished, + // Contract (b): abort resolves only when finished does (request, not kill). + abort: async () => { aborted.push(true); await finished }, + } + this.handles.push({ handle, finish, aborted }) + return handle + } +} + +const tick = () => new Promise(r => setTimeout(r, 0)) + +function collect(s: Scheduler): SchedulerEvent[] { + const seen: SchedulerEvent[] = [] + s.onEvent(e => seen.push(e)) + return seen +} + +describe('Scheduler — serial queue (#56)', () => { + it('runs entries strictly serially: N+1 submits only after N `finished` resolves', async () => { + const ex = new FakeExecutor() + const s = new Scheduler(ex) + const a = s.enqueue({ scriptPath: 'a.jl' }) + const b = s.enqueue({ scriptPath: 'b.jl' }) + await tick() + expect(ex.submits.map(x => x.scriptPath)).toEqual(['a.jl']) // b NOT submitted yet + ex.handles[0].finish({ status: 'completed', exitCode: 0 }) + await tick() + expect(ex.submits.map(x => x.scriptPath)).toEqual(['a.jl', 'b.jl']) + const [ha, hb] = [await a.handle, await b.handle] + expect(ha.runId).toBe('run-1') + expect(hb.runId).toBe('run-2') + }) + + it('S12: the resolved handle IS the executor RunHandle (identity passthrough)', async () => { + const ex = new FakeExecutor() + const s = new Scheduler(ex) + const r = s.enqueue({ scriptPath: 'a.jl' }) + await tick() + expect(await r.handle).toBe(ex.handles[0].handle) + }) + + it('passes SubmitOpts through to executor.submit verbatim', async () => { + const ex = new FakeExecutor() + const s = new Scheduler(ex) + const opts: SubmitOpts = { lab: 'lab-7', runsRoot: '/tmp/rr', julia: { project: '/p' } } + s.enqueue({ scriptPath: 'a.jl', opts }) + await tick() + expect(ex.submits[0].opts).toBe(opts) + }) + + it('contract (b): abort() does NOT advance the queue — only `finished` does', async () => { + const ex = new FakeExecutor() + const s = new Scheduler(ex) + const a = s.enqueue({ scriptPath: 'a.jl' }) + s.enqueue({ scriptPath: 'b.jl' }) + await tick() + const ha = await a.handle + void ha.abort() // request termination… + await tick(); await tick() + expect(ex.submits).toHaveLength(1) // …but the run is still alive: b must NOT start + ex.handles[0].finish({ status: 'aborted', exitCode: 143 }) // FINISHED lands + await tick() + expect(ex.submits).toHaveLength(2) // now b starts + }) + + it('emits the lifecycle: queued → started → finished, with queue position', async () => { + const ex = new FakeExecutor() + const s = new Scheduler(ex) + const seen = collect(s) + s.enqueue({ scriptPath: 'a.jl' }) + s.enqueue({ scriptPath: 'b.jl' }) + await tick() + ex.handles[0].finish({ status: 'completed', exitCode: 0 }) + await tick() + ex.handles[1].finish({ status: 'failed', exitCode: 1 }) + await tick() + expect(seen).toEqual([ + { kind: 'queued', queueId: 'q1', position: 0 }, + { kind: 'queued', queueId: 'q2', position: 1 }, + { kind: 'started', queueId: 'q1', runId: 'run-1', runDir: '/runs/run-1' }, + { kind: 'finished', queueId: 'q1', runId: 'run-1', status: 'completed', exitCode: 0 }, + { kind: 'started', queueId: 'q2', runId: 'run-2', runDir: '/runs/run-2' }, + { kind: 'finished', queueId: 'q2', runId: 'run-2', status: 'failed', exitCode: 1 }, + ]) + }) + + it('cancel() while queued: never submitted, cancelled event, handle rejects', async () => { + const ex = new FakeExecutor() + const s = new Scheduler(ex) + const seen = collect(s) + s.enqueue({ scriptPath: 'a.jl' }) + const b = s.enqueue({ scriptPath: 'b.jl' }) + await tick() + expect(b.cancel()).toBe(true) + ex.handles[0].finish({ status: 'completed', exitCode: 0 }) + await tick() + expect(ex.submits.map(x => x.scriptPath)).toEqual(['a.jl']) // b never ran + expect(seen.some(e => e.kind === 'cancelled' && e.queueId === 'q2')).toBe(true) + await expect(b.handle).rejects.toThrow(/cancel/i) + }) + + it('cancel() after start returns false and the run is untouched (abort via the handle instead)', async () => { + const ex = new FakeExecutor() + const s = new Scheduler(ex) + const a = s.enqueue({ scriptPath: 'a.jl' }) + await tick() + await a.handle + expect(a.cancel()).toBe(false) + expect(ex.handles[0].aborted).toHaveLength(0) // cancel is NOT an abort + }) + + it('a submit() ConfigError rejects that handle, emits error, and the queue advances', async () => { + const ex = new FakeExecutor() + ex.failFor.add('bad.jl') + const s = new Scheduler(ex) + const seen = collect(s) + const bad = s.enqueue({ scriptPath: 'bad.jl' }) + const ok = s.enqueue({ scriptPath: 'ok.jl' }) + await tick() + await expect(bad.handle).rejects.toThrow(/bad config/) + expect(seen.some(e => e.kind === 'error' && e.queueId === 'q1')).toBe(true) + await tick() + expect(ex.submits.map(x => x.scriptPath)).toEqual(['bad.jl', 'ok.jl']) // queue not wedged + expect((await ok.handle).runId).toBe('run-2') // FakeExecutor counts the failed submit too + }) + + it('concurrent: true is a NAMED SEAM — rejected loudly (parallel lane is Phase 4)', () => { + const s = new Scheduler(new FakeExecutor()) + expect(() => s.enqueue({ scriptPath: 'a.jl' }, { concurrent: true })).toThrow(ConfigError) + expect(() => s.enqueue({ scriptPath: 'a.jl' }, { concurrent: true })).toThrow(/Phase 4/) + }) + + it('multiple listeners both receive events; a disposed listener stops receiving', async () => { + const ex = new FakeExecutor() + const s = new Scheduler(ex) + const a: SchedulerEvent[] = [] + const b: SchedulerEvent[] = [] + const disposeA = s.onEvent(e => a.push(e)) + s.onEvent(e => b.push(e)) + s.enqueue({ scriptPath: 'x.jl' }) + await tick() + expect(a.length).toBeGreaterThan(0) + expect(b.length).toBe(a.length) + disposeA() + ex.handles[0].finish({ status: 'completed', exitCode: 0 }) + await tick() + expect(b.length).toBeGreaterThan(a.length) // b kept receiving after a disposed + }) + + it('a throwing listener cannot wedge the pump or starve other listeners', async () => { + const ex = new FakeExecutor() + const s = new Scheduler(ex) + const good: SchedulerEvent[] = [] + s.onEvent(() => { throw new Error('bad listener') }) + s.onEvent(e => good.push(e)) + s.enqueue({ scriptPath: 'x.jl' }) + await tick() + ex.handles[0].finish({ status: 'completed', exitCode: 0 }) + await tick() + expect(good.some(e => e.kind === 'finished')).toBe(true) // pump survived + }) + + it('contract (c): the Scheduler owns no timers (no warming timeout to hard-code)', async () => { + // Structural pin: remote cold-start ≫ local seconds, so ANY scheduler-side + // timeout would violate the per-executor warming budget. Assert the source + // has no timer calls at all. + const { readFileSync } = await import('node:fs') + const { fileURLToPath } = await import('node:url') + const src = readFileSync(fileURLToPath(new URL('../src/scheduler.ts', import.meta.url)), 'utf8') + expect(src).not.toMatch(/setTimeout|setInterval/) + }) +}) From b9ed9feeb0f5ec7b0838f0120e7efbb17e79414d Mon Sep 17 00:00:00 2001 From: Raghav Chari Date: Thu, 2 Jul 2026 21:12:13 -0400 Subject: [PATCH 2/2] =?UTF-8?q?fix(1.1):=20scheduler=20review=20=E2=80=94?= =?UTF-8?q?=20microtask-deferred=20re-pump=20+=20enforce=20the=20defensive?= =?UTF-8?q?=20claims?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adversarial review (mutation-tested; 0 must-fix, 2 should-fix) — all folded in: - finally's re-pump is now queueMicrotask-deferred: a contract-violating executor whose submit() throws SYNCHRONOUSLY previously made the finally a direct recursion — a backlog of such failures accumulated behind a pending run blew the stack on drain (RangeError) and STRANDED the rest of the queue. Mutation-verified: reverting to the direct call fails the new test (RangeError + timeout); the deferral drains 8000 sync-throwers flat. - The rogue-`finished`-rejection branch is now enforced, not just advertised: new test pins error-event + queue-advance (mutation-verified: deleting the branch fails it). Rejection reason normalized (instanceof Error) to match the submit path. - Explicit unhandledRejection pin for the internal handle.catch suppression (an untouched ScheduledRun.handle never trips the process on cancel). - cancel() docstring now names all three false cases (started / already cancelled / mid-submit, where handle may still reject); no-timers grep widened to setImmediate|Date.now. 15 tests green; repo suite green; typecheck clean. Co-Authored-By: Claude Fable 5 --- packages/amico-run/src/scheduler.ts | 19 ++++-- packages/amico-run/test/scheduler.test.ts | 73 ++++++++++++++++++++++- 2 files changed, 85 insertions(+), 7 deletions(-) diff --git a/packages/amico-run/src/scheduler.ts b/packages/amico-run/src/scheduler.ts index b892b1b4..dd1153bf 100644 --- a/packages/amico-run/src/scheduler.ts +++ b/packages/amico-run/src/scheduler.ts @@ -52,9 +52,12 @@ export interface ScheduledRun { * of the queue and submit() succeeds. Rejects if the entry is cancelled * before starting, or if submit() throws (e.g. ConfigError). */ handle: Promise - /** Dequeue BEFORE start: true if the entry was still queued (it will never - * run), false once started — a live run is stopped via RunHandle.abort() - * (a request, per contract (b)), never via the queue. */ + /** Dequeue BEFORE start: true iff the entry was still queued (it will never + * run). False in every other case — already started, already cancelled, or + * mid-submit (shifted but `started` not yet emitted; `handle` may still + * REJECT if that submit fails). To stop a live run, `await handle` (in a + * try/catch) and call RunHandle.abort() — a request, per contract (b); + * never via the queue. */ cancel(): boolean } @@ -146,11 +149,17 @@ export class Scheduler { this.emit({ kind: 'finished', queueId: entry.queueId, runId: handle.runId, status: fin.status, exitCode: fin.exitCode }) } catch (e) { // …but a rogue executor breaking that must not deadlock every queued run. - this.emit({ kind: 'error', queueId: entry.queueId, message: `finished rejected: ${(e as Error).message}` }) + const msg = e instanceof Error ? e.message : String(e) + this.emit({ kind: 'error', queueId: entry.queueId, message: `finished rejected: ${msg}` }) } } finally { this.running = false - void this.pump() // next entry, if any + // Microtask deferral, NOT a direct call: a contract-violating executor + // whose submit() throws SYNCHRONOUSLY would otherwise make this finally + // direct recursion — a long backlog of such failures blows the stack and + // strands the rest of the queue. Deferring one microtask keeps the chain + // flat regardless of how the executor misbehaves. + queueMicrotask(() => void this.pump()) } } } diff --git a/packages/amico-run/test/scheduler.test.ts b/packages/amico-run/test/scheduler.test.ts index e00e5187..4bf78981 100644 --- a/packages/amico-run/test/scheduler.test.ts +++ b/packages/amico-run/test/scheduler.test.ts @@ -192,13 +192,82 @@ describe('Scheduler — serial queue (#56)', () => { expect(good.some(e => e.kind === 'finished')).toBe(true) // pump survived }) + it('contract (d): a rogue `finished` REJECTION is survived — error event, queue advances', async () => { + // `finished` never rejects per contract; a broken executor must still not + // wedge every queued run behind it. (Pins the defensive branch — a mutation + // deleting it must fail here.) + class RogueExecutor extends FakeExecutor { + async submit(scriptPath: string, opts?: SubmitOpts): Promise { + const h = await super.submit(scriptPath, opts) + if (scriptPath === 'rogue.jl') return { ...h, finished: Promise.reject(new Error('boom')) } + return h + } + } + const ex = new RogueExecutor() + const s = new Scheduler(ex) + const seen = collect(s) + s.enqueue({ scriptPath: 'rogue.jl' }) + const ok = s.enqueue({ scriptPath: 'ok.jl' }) + await tick(); await tick() + expect(seen.some(e => e.kind === 'error' && /finished rejected: boom/.test((e as { message: string }).message))).toBe(true) + expect(ex.submits.map(x => x.scriptPath)).toEqual(['rogue.jl', 'ok.jl']) // queue advanced + expect((await ok.handle).runId).toBe('run-2') + }) + + it('a SYNC-throwing submit (contract-violating executor) cannot blow the stack or strand the queue', async () => { + // The dangerous shape: a big backlog of sync-throwers ACCUMULATES behind one + // pending run, then drains in a single chain when it finishes. With a direct + // finally re-pump that chain is real recursion (RangeError → stranded queue); + // the microtask deferral keeps it flat. (Enqueuing sync-throwers onto an idle + // scheduler never recurses — each enqueue drains its own entry — so the + // backlog-behind-a-pending-run setup is load-bearing for this pin.) + class SyncThrower implements Executor { + good = new FakeExecutor() + submit(scriptPath: string, opts?: SubmitOpts): Promise { + if (!scriptPath.startsWith('bad-')) return this.good.submit(scriptPath, opts) + throw new ConfigError(`sync boom: ${scriptPath}`) // sync, no Promise + } + } + const ex = new SyncThrower() + const s = new Scheduler(ex) + s.enqueue({ scriptPath: 'first.jl' }) // holds the queue while the backlog builds + await tick() + const bad = Array.from({ length: 8000 }, (_, i) => s.enqueue({ scriptPath: `bad-${i}.jl` })) + const good = s.enqueue({ scriptPath: 'good.jl' }) + ex.good.handles[0].finish({ status: 'completed', exitCode: 0 }) // release → drain the 8000 in one go + const h = await good.handle // resolves only if the whole backlog drained + expect(h.runId).toBe('run-2') + expect(s.depth).toBe(1) // just the good run, still running + await expect(bad[0].handle).rejects.toThrow(/sync boom/) + await expect(bad[7999].handle).rejects.toThrow(/sync boom/) + }) + + it('an untouched ScheduledRun.handle never surfaces an unhandledRejection (cancel path)', async () => { + // Pins the internal handle.catch(() => {}) suppression explicitly — callers + // that only consume lifecycle events never touch `handle`, and a cancel's + // rejection must not trip the process. + const seen: unknown[] = [] + const trap = (r: unknown): void => { seen.push(r) } + process.on('unhandledRejection', trap) + try { + const s = new Scheduler(new FakeExecutor()) + s.enqueue({ scriptPath: 'a.jl' }) + const b = s.enqueue({ scriptPath: 'b.jl' }) + expect(b.cancel()).toBe(true) // rejects b.handle — nobody is listening + await tick(); await tick() + expect(seen).toEqual([]) + } finally { + process.off('unhandledRejection', trap) + } + }) + it('contract (c): the Scheduler owns no timers (no warming timeout to hard-code)', async () => { // Structural pin: remote cold-start ≫ local seconds, so ANY scheduler-side // timeout would violate the per-executor warming budget. Assert the source - // has no timer calls at all. + // has no timer calls at all (microtasks are fine — they encode no duration). const { readFileSync } = await import('node:fs') const { fileURLToPath } = await import('node:url') const src = readFileSync(fileURLToPath(new URL('../src/scheduler.ts', import.meta.url)), 'utf8') - expect(src).not.toMatch(/setTimeout|setInterval/) + expect(src).not.toMatch(/setTimeout|setInterval|setImmediate|Date\.now/) }) })