tricu

io-v2-driver.ts (4505B)

---

 import type { IOHost } from './io.js';
 import type { Command, Session } from './io-v2-session.js';
 import type { ValueView } from './codecs.js';
 import type { Event } from './io-v2-events.js';
 export interface V2Host {
   readonly stepping?: boolean;
   // These transport callbacks must not throw or synchronously reenter a turn.
   open(run: string, command: (command: Command) => void): void;
   event(event: Event): void;
   close(run: string, settled: boolean): void;
 }
 export type Report = Awaited<ReturnType<Session<unknown>['turn']>>;
 export interface Driver {
   run: string;
   // Emergency runtime loss, distinct from structured stop/acknowledgement.
   signal?: AbortSignal;
   turn(command: Command): Promise<Report>;
   view(): Promise<ValueView>;
   discard?(): void;
 }
 // Unlike repeated Promise.race against a lifetime-long promise, listeners are
 // removed on EVERY turn/step settlement, not retained until runtime disposal.
 function interrupt<T>(promise: Promise<T>, signal: AbortSignal | undefined, aborted: () => T): Promise<T> {
   if (!signal) return promise;
   let onAbort = () => {};
   return new Promise<T>((resolve, reject) => {
     onAbort = () => { try { resolve(aborted()); } catch (error) { reject(error); } };
     if (signal.aborted) onAbort(); else signal.addEventListener('abort', onAbort, { once: true });
     void promise.then(resolve, reject);
   }).finally(() => signal.removeEventListener('abort', onAbort));
 }
 // The queue stays alive while a read or step prompt is waiting. Stop is sticky,
 // independent of the resume queue, and never implemented by worker destruction.
 export async function drive(driver: Driver, host: IOHost): Promise<ValueView> {
   const transport = host.v2;
   if (!transport) throw new Error('IO version 2 requires a structured terminal host');
   let settled = false;
   const alive = <T>(promise: Promise<T>): Promise<T> => interrupt(promise, driver.signal, () => { throw driver.signal!.reason; });
   let active = true, stopping = false, stopSent = false, wake = () => {}, step = 0;
   const replies: Command[] = [];
   let boundaryFault = false, fault: unknown;
   const stopped = new AbortController();
   const enqueue = (command: Command) => {
     if (!active || !command || command.run !== driver.run) return;
     if (command.kind === 'stop') { stopping = true; stopped.abort(); }
     else if (command.kind === 'reply') replies.push(command);
     wake();
   };
   const close = () => {
     if (!active) return;
     active = false; stopped.abort(); wake(); wake = () => {}; replies.length = 0;
     // A failed transport/runtime cannot be reused with admitted work abandoned.
     if (!settled) driver.discard?.();
     transport.close(driver.run, settled);
   };
   try {
     transport.open(driver.run, enqueue);
     for (;;) {
       if (!stopping && transport.stepping === true) {
         try { await alive(interrupt(host.before({ step: ++step, name: 'v2 scheduler turn', detail: 'one scheduler advance; all tasks paused', depth: 0 }), stopped.signal, () => {})); }
         catch (error) { if (driver.signal?.aborted) throw error; boundaryFault = true; fault = error; enqueue({ run: driver.run, kind: 'stop' }); }
       }
       const command: Command = stopping && !stopSent ? (stopSent = true, { run: driver.run, kind: 'stop' }) :
         replies.shift() ?? { run: driver.run, kind: 'poll' };
       const report = await alive(driver.turn(command));
       for (const event of report.events) transport.event(event);
       if (report.done) {
         if (!Object.values(report.counts).every(group => Object.values(group).every(value => value === 0))) throw new Error('IO v2 settlement retains work');
         settled = true; close();
         if (boundaryFault) throw fault;
         if (report.outcome?.fault) throw new Error('IO v2 interpreter fault');
         if (report.outcome?.kind !== 'completed') throw new Error(`IO v2 ${report.outcome?.kind ?? 'failed'}`);
         return await alive(driver.view()); // Full codec, only after physical settlement.
       }
       // Yield even during pure scheduler progress so the outer worker can receive
       // stop/replies. This is not a hard deadline for native tree reduction.
       if (report.waiting && !replies.length && (!stopping || stopSent)) await alive(new Promise<void>(resolve => { wake = resolve; }));
       else await alive(new Promise<void>(resolve => { const timer = setTimeout(resolve, 0); wake = () => { clearTimeout(timer); resolve(); }; }));
       wake = () => {};
     }
   } finally { close(); }
 }