require('ts-node/register/transpile-only'); const assert = require('node:assert/strict'); const { afterEach, test } = require('node:test'); const { Sequelize } = require('sequelize'); const { defineSchemaMigrationModel, } = require('../../back/data/schemaMigration'); const { runSchemaMigration } = require('../../back/migrations/0002-run-schema'); const { runCancellationRequestMigration, } = require('../../back/migrations/0004-run-cancellation-request'); const { runAttemptDeadlineMigration, } = require('../../back/migrations/0006-run-attempt-deadline'); const { runMigrations } = require('../../back/migrations/runner'); const { LegacySequelizeRunRepository, } = require('../../back/runtime/adapters/legacy-sequelize/runRepository'); const { ManualPrimaryOwnershipError, ManualPrimaryRuntime, } = require('../../back/runtime/application/manualPrimaryRuntime'); const { installManualPrimaryExecutionRouter, selectManualPrimaryExecutionRouter, stopManualPrimaryAttempt, stopManualPrimaryCron, } = require('../../back/runtime/compatibility/manualPrimaryExecutionBridge'); const { parseLegacyLogOutputRef, } = require('../../back/runtime/compatibility/legacyLogOutputRef'); const { RuntimeRolloutPolicy, } = require('../../back/runtime/domain/runtimeRollout'); const databases = []; const restorers = []; const BASE_TIME = 1_750_000_000_000; let idSequence = 1_100; function nextId() { idSequence += 1; return '019f7120-0000-7000-8000-' + String(idSequence).padStart(12, '0'); } function deferred() { let resolve; let reject; const promise = new Promise((resolvePromise, rejectPromise) => { resolve = resolvePromise; reject = rejectPromise; }); return { promise, resolve, reject }; } async function createRepository() { const database = new Sequelize({ dialect: 'sqlite', storage: ':memory:', logging: false, }); const migrationModel = defineSchemaMigrationModel(database); await runMigrations({ database, migrationModel, migrations: [ runSchemaMigration, runCancellationRequestMigration, runAttemptDeadlineMigration, ], logger: { info() {} }, }); databases.push(database); return new LegacySequelizeRunRepository(database); } function rollout(mode) { return new RuntimeRolloutPolicy({ defaultMode: 'off', origins: { manual: mode }, allowLegacyFallbackBeforeStart: false, }); } class FakeExecutor { type = 'local_process'; starts = []; stops = []; completion = deferred(); capabilities() { return { timeout: true, processGroupTermination: true, workingDirectory: true, isolatedEnvironment: true, memoryLimit: 'none', cpuLimit: 'none', filesystemIsolation: 'none', networkIsolation: 'none', }; } async start(spec, context) { this.starts.push({ spec, context }); if (context.signal && context.signal.aborted) { throw new Error('aborted before fake spawn'); } return { id: nextId(), executorType: this.type, runId: spec.runId, attemptId: spec.attemptId, startedAtMs: BASE_TIME + 10, pid: 7301, completion: this.completion.promise, }; } async stop(handle, reason) { this.stops.push({ handle, reason }); return { status: 'termination_requested', termSignalSent: true, killSignalSent: false, }; } async inspect() { return { status: 'running' }; } } class FakeLogs { prepareCalls = 0; closeCalls = 0; committedAttempts = []; writes = []; async prepare(input) { this.prepareCalls += 1; return { logPath: 'task-' + input.cron.id + '/manual.log', output: { write: async (output) => { this.writes.push(output); }, }, completionCommitted: async (attemptId) => { this.committedAttempts.push(attemptId); }, close: async () => { this.closeCalls += 1; }, }; } } function input(cronId = 21) { return { cron: { id: cronId, name: 'manual primary', command: 'demo.js arg', schedule: '0 * * * *', extraSchedules: ['30 * * * *'], taskBefore: 'echo before', taskAfter: 'echo after', workDirectory: '/tmp', logName: 'task-' + cronId, }, acceptedAtMs: BASE_TIME, }; } afterEach(async () => { while (restorers.length > 0) restorers.pop()(); await Promise.all(databases.splice(0).map((database) => database.close())); }); test('off mode selects Legacy without preparing logs or spawning', async () => { const repository = await createRepository(); const executor = new FakeExecutor(); const logs = new FakeLogs(); const runtime = new ManualPrimaryRuntime( repository, executor, rollout('off'), logs, { orchestrator: { createId: nextId } }, ); assert.equal(runtime.ownsNewRuns(), false); await assert.rejects(runtime.start(input()), ManualPrimaryOwnershipError); assert.equal(logs.prepareCalls, 0); assert.equal(executor.starts.length, 0); }); test('Primary mode builds one durable Run and one legacy-compatible spec', async () => { const repository = await createRepository(); const executor = new FakeExecutor(); const logs = new FakeLogs(); const runtime = new ManualPrimaryRuntime( repository, executor, rollout('primary'), logs, { clock: { now: () => BASE_TIME + 20 }, orchestrator: { createId: nextId }, }, ); const active = await runtime.start(input()); assert.equal(executor.starts.length, 1); assert.equal(active.pid, 7301); assert.equal(active.logPath, 'task-21/manual.log'); const started = executor.starts[0]; assert.equal(started.spec.command.kind, 'shell'); assert.match( started.spec.command.command, /real_log_path='task-21\/manual\.log'/, ); assert.match(started.spec.command.command, /task demo\.js arg/); assert.equal(started.spec.environmentPolicy, 'inherit'); assert.equal(started.context.signal.aborted, false); const run = await repository.findRunById(active.runId); assert.equal(run.executionOwner, 'runtime'); assert.equal(run.executionOrigin, 'manual'); assert.equal(parseLegacyLogOutputRef(run.outputRef), 'task-21/manual.log'); executor.completion.resolve({ outcome: 'succeeded', startedAtMs: BASE_TIME + 10, finishedAtMs: BASE_TIME + 30, exitCode: 0, }); const completed = await active.completion; assert.equal(completed.outcome, 'succeeded'); assert.equal(completed.exitCode, 0); assert.deepEqual(logs.committedAttempts, [active.attemptId]); assert.equal(logs.closeCalls, 1); }); test('routes stop by cron and attempt through the owning Executor', async () => { const repository = await createRepository(); const executor = new FakeExecutor(); const runtime = new ManualPrimaryRuntime( repository, executor, rollout('primary'), new FakeLogs(), { orchestrator: { createId: nextId } }, ); const active = await runtime.start(input(22)); assert.deepEqual(await runtime.stopCron(999, BASE_TIME + 40), { matched: 0, failed: 0, }); assert.deepEqual( await runtime.stopAttempt(active.attemptId, BASE_TIME + 41), { matched: 1, failed: 0 }, ); assert.equal(executor.stops.length, 1); assert.equal(executor.stops[0].reason.kind, 'user'); executor.completion.resolve({ outcome: 'cancelled', startedAtMs: BASE_TIME + 10, finishedAtMs: BASE_TIME + 50, exitCode: 143, }); assert.equal((await active.completion).outcome, 'cancelled'); }); test('a stop racing log preparation aborts before Executor side effects', async () => { const repository = await createRepository(); const executor = new FakeExecutor(); const preparing = deferred(); const release = deferred(); const logs = { async prepare() { preparing.resolve(); await release.promise; return { logPath: 'task-24/racing.log', output: { async write() {} }, async close() {}, }; }, }; const runtime = new ManualPrimaryRuntime( repository, executor, rollout('primary'), logs, { orchestrator: { createId: nextId } }, ); const starting = runtime.start(input(24)); await preparing.promise; assert.deepEqual(await runtime.stopCron(24, BASE_TIME + 1), { matched: 1, failed: 0, }); release.resolve(); await assert.rejects(starting); assert.equal(executor.starts.length, 1); assert.equal(executor.starts[0].context.signal.aborted, true); }); test('bridge is default-off and keeps stop routing available for its owner', async () => { assert.equal(selectManualPrimaryExecutionRouter(), undefined); assert.deepEqual(await stopManualPrimaryCron(1, BASE_TIME), { matched: 0, failed: 0, }); const calls = []; const router = { ownsNewRuns: () => true, async start() { throw new Error('not used'); }, async stopCron(cronId, requestedAtMs) { calls.push(['cron', cronId, requestedAtMs]); return { matched: 1, failed: 0 }; }, async stopAttempt(attemptId, requestedAtMs) { calls.push(['attempt', attemptId, requestedAtMs]); return { matched: 1, failed: 0 }; }, }; restorers.push(installManualPrimaryExecutionRouter(router)); assert.equal(selectManualPrimaryExecutionRouter(), router); assert.deepEqual(await stopManualPrimaryCron(23, BASE_TIME + 1), { matched: 1, failed: 0, }); assert.deepEqual( await stopManualPrimaryAttempt('attempt-23', BASE_TIME + 2), { matched: 1, failed: 0, }, ); assert.deepEqual(calls, [ ['cron', 23, BASE_TIME + 1], ['attempt', 'attempt-23', BASE_TIME + 2], ]); }); test('rolling new triggers to off preserves the previous in-flight stop owner', async () => { const oldOwner = { ownsNewRuns: () => true, async start() { throw new Error('not used'); }, async stopCron(cronId) { return { matched: cronId === 25 ? 1 : 0, failed: 0, }; }, async stopAttempt() { return { matched: 0, failed: 0 }; }, }; const offRouter = { ownsNewRuns: () => false, async start() { throw new Error('not used'); }, async stopCron() { return { matched: 0, failed: 0 }; }, async stopAttempt() { return { matched: 0, failed: 0 }; }, }; restorers.push(installManualPrimaryExecutionRouter(oldOwner)); restorers.push(installManualPrimaryExecutionRouter(offRouter)); assert.equal(selectManualPrimaryExecutionRouter(), undefined); assert.deepEqual(await stopManualPrimaryCron(25, BASE_TIME + 3), { matched: 1, failed: 0, }); });