tricu

io-v2-tasks.js (13181B)

---

 export const hostKey = (token) => `${token.kind}:${token.key}`;
 const empty = () => ({ items: [], omitted: 0 });
 const primary = (outcome) => ({ kind: outcome.kind, value: outcome.value });
 function append(left, right) {
     if (!Number.isSafeInteger(right.omitted) || right.omitted < 0)
         throw new Error('invalid diagnostics');
     const available = 16 - left.items.length;
     return { items: [...left.items, ...right.items.slice(0, available)],
         omitted: Math.min(Number.MAX_SAFE_INTEGER, left.omitted + right.omitted + Math.max(0, right.items.length - available)) };
 }
 const capture = (outcome) => {
     if (!['completed', 'failed', 'cancelled'].includes(outcome.kind))
         throw new Error('invalid outcome');
     return { ...primary(outcome), cleanup: append(empty(), outcome.cleanup ?? empty()) };
 };
 export class Tasks {
     identity;
     ownerStopped;
     cycle;
     identityLimit;
     resources;
     groups = new Map();
     tasks = new Map();
     hosts = new Map();
     nextScope = 1;
     nextHost = 1;
     constructor(identity, ownerStopped, cycle, identityLimit = Number.MAX_SAFE_INTEGER, resources) {
         this.identity = identity;
         this.ownerStopped = ownerStopped;
         this.cycle = cycle;
         this.identityLimit = identityLimit;
         this.resources = resources;
         if (typeof identity !== 'bigint' || identity <= 0n || identity > 0xffffffffffffffffn ||
             !Number.isSafeInteger(identityLimit) || identityLimit < 1)
             throw new Error('invalid run identity or limit');
         this.groups.set(0, this.group(null, 0, 'root', false));
     }
     group(parent, task, kind, propagate) {
         return { parent, task, kind, propagate, phase: 'open', reason: null, body: null,
             cleanup: empty(), lexical: null, wait: null, hostCalls: new Set(), releaseStarted: false };
     }
     get(scope) {
         const group = this.groups.get(scope);
         if (!group)
             throw new Error('closed-scope');
         return group;
     }
     // Read-only copies: embeddings cannot mutate lifecycle state or diagnostic arrays.
     state(scope) {
         const g = this.get(scope);
         return { parent: g.parent, task: g.task, kind: g.kind, phase: g.phase,
             reason: g.reason && primary(g.reason), lexical: g.lexical, wait: g.wait, hostCalls: g.hostCalls.size };
     }
     get counts() { return { scopes: this.groups.size, tasks: this.tasks.size, hosts: this.hosts.size }; }
     scopeIds() { return [...this.groups.keys()]; }
     within(current, owner) {
         for (let scope = current; scope !== null;) {
             const g = this.groups.get(scope);
             if (!g)
                 return false;
             if (scope === owner)
                 return true;
             scope = g.parent;
         }
         return false;
     }
     checkpoint(scope) {
         const g = this.get(scope);
         if (g.reason)
             throw new Error('stopped');
         if (g.phase !== 'open' || g.lexical !== null || g.wait !== null)
             throw new Error('invalid-transition');
     }
     scopeId() {
         if (this.nextScope > this.identityLimit)
             throw new Error('scope-identity-exhausted');
         return this.nextScope++;
     }
     nested(parent) {
         this.checkpoint(parent);
         const id = this.scopeId(), child = this.group(parent, this.get(parent).task, 'lexical', true);
         this.groups.set(id, child);
         this.get(parent).lexical = id;
         return id;
     }
     spawn(owner, supervised, keep) {
         this.checkpoint(owner);
         if (typeof supervised !== 'boolean' || typeof keep !== 'boolean')
             throw new Error('invalid-argument');
         const id = this.scopeId(), child = this.group(owner, id, 'spawned', !supervised);
         this.groups.set(id, child);
         this.tasks.set(id, { owner, supervised, ticket: keep, waiter: null, outcome: null });
         return { run: this.identity, key: id };
     }
     recordStop(scope, reason) {
         const g = this.groups.get(scope);
         if (!g || g.reason)
             return false;
         g.reason = primary(reason);
         return true;
     }
     stop(scope, reason) {
         if (reason.kind !== 'failed' && reason.kind !== 'cancelled')
             throw new Error('invalid stop');
         if (!this.recordStop(scope, reason))
             return;
         // Fixed point propagates first failure upward and owner cancellation downward.
         for (let changed = true; changed;) {
             changed = false;
             for (const [id, g] of this.groups)
                 if (g.parent !== null) {
                     if (g.reason?.kind === 'failed' && g.propagate)
                         changed = this.recordStop(g.parent, g.reason) || changed;
                     if (this.get(g.parent).reason)
                         changed = this.recordStop(id, { kind: 'cancelled', value: this.ownerStopped }) || changed;
                 }
         }
     }
     checked(current, token) {
         if (token.run !== this.identity)
             throw new Error('stale-task');
         const task = this.tasks.get(token.key);
         if (!task)
             throw new Error('stale-task');
         if (!this.within(current, task.owner))
             throw new Error('out-of-scope');
         return task;
     }
     // Edges are task identities, not lexical scope IDs. Normal scope exit adds
     // implicit edges to unfinished children; explicit join cycle checks include them.
     reaches(start, wanted) {
         const pending = [start], seen = new Set();
         while (pending.length) {
             const task = pending.pop();
             if (task === wanted)
                 return true;
             if (seen.has(task))
                 continue;
             seen.add(task);
             for (const [id, g] of this.groups)
                 if (g.task === task) {
                     if (g.wait !== null)
                         pending.push(g.wait);
                     if (g.phase === 'joining')
                         for (const [child, t] of this.tasks)
                             if (t.owner === id && t.outcome === null)
                                 pending.push(child);
                 }
         }
         return false;
     }
     claimJoin(current, token) {
         this.checkpoint(current);
         const task = this.checked(current, token);
         if (!task.ticket)
             throw new Error('stale-task');
         if (this.within(current, token.key) || this.reaches(token.key, this.get(current).task))
             throw new Error('join-cycle');
         task.ticket = false;
         task.waiter = current;
         this.get(current).wait = token.key;
     }
     pollJoin(current) {
         const g = this.get(current);
         if (g.wait === null)
             throw new Error('invalid-transition');
         const task = this.tasks.get(g.wait);
         if (!task.outcome)
             return null;
         const outcome = capture(task.outcome);
         this.tasks.delete(g.wait);
         g.wait = null;
         return outcome;
     }
     abandonJoin(current) {
         const g = this.groups.get(current);
         if (!g || g.wait === null)
             return;
         const task = this.tasks.get(g.wait);
         task.waiter = null;
         if (task.outcome)
             this.tasks.delete(g.wait);
         g.wait = null;
     }
     discard(current, token) {
         this.checkpoint(current);
         const task = this.checked(current, token);
         if (!task.ticket)
             throw new Error('stale-task');
         task.ticket = false;
         if (task.outcome)
             this.tasks.delete(token.key);
     }
     cancel(current, token, reason) {
         this.checkpoint(current);
         const task = this.checked(current, token);
         if (task.outcome)
             throw new Error('stale-task');
         this.stop(token.key, { kind: 'cancelled', value: reason });
     }
     bodyDone(scope, outcome) {
         const g = this.get(scope), captured = capture(outcome);
         if (g.phase !== 'open' || g.lexical !== null)
             throw new Error('invalid-transition');
         this.abandonJoin(scope);
         if (captured.kind !== 'completed')
             this.stop(scope, captured);
         g.body = captured;
         for (const [id, t] of this.tasks)
             if (t.owner === scope && !t.outcome && this.reaches(id, g.task)) {
                 this.stop(scope, { kind: 'failed', value: this.cycle });
                 break;
             }
         g.phase = 'joining';
     }
     beginHost(scope) {
         this.checkpoint(scope);
         return this.registerHost(scope, 'body');
     }
     beginReleaseHost(scope) {
         const group = this.get(scope);
         if (group.phase !== 'releasing' || group.releaseStarted)
             throw new Error('invalid-transition');
         // One cleanup batch per scope uses its already-reserved identity, in a
         // separate namespace. Body-call identity exhaustion cannot prevent cleanup.
         const token = this.registerHost(scope, 'release');
         group.releaseStarted = true;
         return token;
     }
     registerHost(scope, kind) {
         if (kind === 'body' && this.nextHost > this.identityLimit)
             throw new Error('host-identity-exhausted');
         const token = Object.freeze({ run: this.identity, key: kind === 'body' ? this.nextHost++ : scope, scope, kind });
         const key = hostKey(token);
         this.hosts.set(key, { token, cancelling: false });
         this.get(scope).hostCalls.add(key);
         return token;
     }
     // Delivery is separate from acknowledgement. Issuing AbortController.abort()
     // cannot settle a task whose host callback/lease is still outstanding.
     takeCancellations() {
         const pending = [];
         for (const call of this.hosts.values())
             if (call.token.kind === 'body' && !call.cancelling && this.get(call.token.scope).reason) {
                 call.cancelling = true;
                 pending.push(call.token);
             }
         return pending;
     }
     endHost(token) {
         const key = hostKey(token), call = this.hosts.get(key);
         if (token.run !== this.identity || !call || call.token.scope !== token.scope || call.token.kind !== token.kind)
             return 'stale';
         const g = this.get(token.scope);
         this.hosts.delete(key);
         g.hostCalls.delete(key);
         return token.kind === 'release' || !g.reason && !g.body ? 'reply' : 'ignored';
     }
     beginRelease(scope) {
         const g = this.get(scope);
         if (g.phase !== 'joining')
             throw new Error('invalid-transition');
         if (g.hostCalls.size || [...this.tasks.values()].some(t => t.owner === scope && !t.outcome) ||
             this.resources && !this.resources.readyForRelease(scope))
             return false;
         g.phase = 'releasing';
         return true;
     }
     // Call only after the embedding has physically released this scope's resources.
     settle(scope, ownCleanup = empty()) {
         const g = this.get(scope);
         if (g.phase !== 'releasing' || !g.body || g.hostCalls.size)
             throw new Error('invalid-transition');
         let final = primary(g.reason ?? g.body);
         let errors = append(append(g.body.cleanup ?? empty(), g.cleanup), ownCleanup);
         if (final.kind === 'completed' && errors.items.length) {
             final = { kind: 'failed', value: errors.items[0] };
             errors = { items: errors.items.slice(1), omitted: errors.omitted };
         }
         this.resources?.finishScope(scope);
         final = { ...final, cleanup: errors };
         if (final.kind === 'failed')
             this.stop(scope, final);
         for (const [id, task] of this.tasks)
             if (task.owner === scope)
                 this.tasks.delete(id);
         if (g.parent !== null) {
             const parent = this.get(g.parent);
             if (g.kind === 'lexical') {
                 parent.cleanup = append(parent.cleanup, errors);
                 parent.lexical = null;
             }
             else {
                 const task = this.tasks.get(scope);
                 if (!task.supervised)
                     parent.cleanup = append(parent.cleanup, errors);
                 task.outcome = final;
                 if (!task.ticket && task.waiter === null)
                     this.tasks.delete(scope);
             }
         }
         this.groups.delete(scope);
         return g.kind === 'lexical' ? primary(final) : capture(final);
     }
     // Evaluator roots, in addition to embedding-owned bodies, continuations and
     // pending host arguments. No task/result history remains after consumption.
     *payloads() {
         yield this.ownerStopped;
         yield this.cycle;
         for (const g of this.groups.values()) {
             if (g.reason)
                 yield g.reason.value;
             if (g.body) {
                 yield g.body.value;
                 yield* g.body.cleanup?.items ?? [];
             }
             yield* g.cleanup.items;
         }
         for (const task of this.tasks.values())
             if (task.outcome) {
                 yield task.outcome.value;
                 yield* task.outcome.cleanup?.items ?? [];
             }
     }
 }