rocko
4bdc7e0e86
Unify halt and fire cancellation by verified run identity with durable first-owner receipts and shared deadlines. Integrate the instance control with the current host, library and tester-socket paths. Focused nine-file evidence is 62 passing tests; broad suites, mutants and review remain pending.
62 lines
2.7 KiB
JavaScript
62 lines
2.7 KiB
JavaScript
import { AsyncLocalStorage } from 'node:async_hooks';
|
|
import { randomUUID } from 'node:crypto';
|
|
import { assertProcessing } from './state.mjs';
|
|
import { registerRuntime, updateRuntime, unregisterRuntime, processIdentity, createRuntimeSignaller, runtimeGroupState, runtime } from './runtimes.mjs';
|
|
const contexts = new AsyncLocalStorage();
|
|
export const currentProcessing = () => contexts.getStore();
|
|
// Trusted compensation/temporary-file cleanup only; never an agent work entry point.
|
|
export const withoutProcessing = fn => contexts.run(undefined, fn);
|
|
export async function withProcessing(dataRoot, fn) {
|
|
const parent = currentProcessing();
|
|
const generation = assertProcessing(dataRoot, parent?.dataRoot === dataRoot ? parent.generation : null).generation;
|
|
const pending = registerRuntime(dataRoot, { kind: 'effect', business: 'instance', run: randomUUID() });
|
|
const controller = new AbortController();
|
|
const children = [];
|
|
const guard = () => assertProcessing(dataRoot, generation);
|
|
const poll = setInterval(() => {
|
|
try { guard(); } catch (e) { controller.abort(e); }
|
|
}, 25);
|
|
poll.unref();
|
|
const scope = {
|
|
dataRoot, generation, signal: controller.signal, guard,
|
|
track(child) {
|
|
const identity = processIdentity(child.pid);
|
|
if (!identity) return;
|
|
children.push(new Promise(resolve => child.once('close', resolve)));
|
|
let record;
|
|
try { record = registerRuntime(dataRoot, { kind: 'effect', business: 'instance', run: randomUUID() }); }
|
|
catch (e) {
|
|
createRuntimeSignaller({ ...identity, kind: 'effect', group: true })('SIGKILL');
|
|
throw e;
|
|
}
|
|
const running = updateRuntime(dataRoot, record.id, { ...identity, group: true, phase: 'running' });
|
|
const signal = createRuntimeSignaller(running, members => updateRuntime(dataRoot, record.id, { members }));
|
|
let kill = null;
|
|
const cancel = () => {
|
|
signal('SIGTERM');
|
|
kill = setTimeout(() => signal('SIGKILL'), 5000);
|
|
kill.unref();
|
|
};
|
|
controller.signal.addEventListener('abort', cancel, { once: true });
|
|
child.once('close', () => {
|
|
controller.signal.removeEventListener('abort', cancel); clearTimeout(kill);
|
|
try {
|
|
if (runtimeGroupState(runtime(dataRoot, record.id)) === false) unregisterRuntime(dataRoot, record.id);
|
|
} catch { /* keep unverifiable receipts for reconciliation */ }
|
|
});
|
|
try { guard(); } catch (e) { controller.abort(e); }
|
|
if (controller.signal.aborted && kill === null) cancel();
|
|
},
|
|
};
|
|
try {
|
|
guard();
|
|
const result = await contexts.run(scope, () => fn(scope));
|
|
guard();
|
|
return result;
|
|
} finally {
|
|
clearInterval(poll);
|
|
await Promise.all(children);
|
|
unregisterRuntime(dataRoot, pending.id);
|
|
}
|
|
}
|