tricu

io-host-work.ts (4302B)

---

 import { Tasks, hostKey, type HostToken } from './io-tasks.js';
 export type Reply<T> = { readonly token: HostToken; readonly outcome:
   { readonly ok: true; readonly value: T } | { readonly ok: false; readonly error: unknown } };
 // Adapter-side asynchronous work. Starting one callback never waits for it, so
 // sibling actions/cancellation can proceed. The provider must settle its promise
 // AFTER cancellation cleanup; abort delivery alone is never an acknowledgement.
 export class HostWork<P, T> {
   private readonly active = new Map<string, { token: HostToken; controller: AbortController }>();
   private readonly replies = new Map<string, Reply<T>>();
   private notification: { promise: Promise<void>; resolve: () => void } | null = null;
   constructor(private readonly tasks: Tasks<P>) {}
   start(scope: number, operation: (signal: AbortSignal) => T | Promise<T>): HostToken {
     return this.register(this.tasks.beginHost(scope), operation);
   }
   // Cleanup is not cancelled by a new stop request. Its result (including a
   // failure) must still reach the driver to preserve primary/secondary outcomes.
   release(scope: number, operation: () => T | Promise<T>): HostToken {
     return this.register(this.tasks.beginReleaseHost(scope), operation);
   }
   private register(token: HostToken, operation: (signal: AbortSignal) => T | Promise<T>): HostToken {
     const controller = new AbortController();
     this.active.set(hostKey(token), { token, controller });
     const complete = (outcome: Reply<T>['outcome']) => {
       // Settle first; a cancelled reply is discarded without observing its value.
       const disposition = this.tasks.endHost(token);
       this.active.delete(hostKey(token));
       if (disposition === 'reply') this.replies.set(hostKey(token), { token, outcome });
       const notice = this.notification; this.notification = null; notice?.resolve();
     };
     try { Promise.resolve(operation(controller.signal)).then(
       value => complete({ ok: true, value }), error => complete({ ok: false, error })); }
     catch (error) { complete({ ok: false, error }); }
     return token;
   }
   private usable(token: HostToken): boolean {
     if (!this.tasks.within(token.scope, token.scope)) return false;
     const state = this.tasks.state(token.scope);
     return token.kind === 'release' ? state.phase === 'releasing' : state.phase === 'open' && state.reason === null;
   }
   // Resource close may interrupt one operation without cancelling its task.
   // The resulting provider reply still reaches that body as an ordinary error.
   // Never abort a cleanup batch or acknowledge work merely by sending this signal.
   cancelBody(token: HostToken): boolean {
     const work = this.active.get(hostKey(token));
     if (token.kind !== 'body' || !work || work.token.run !== token.run ||
         work.token.scope !== token.scope || work.controller.signal.aborted) return false;
     work.controller.abort(); return true;
   }
   cancelStopped(): void {
     for (const token of this.tasks.takeCancellations()) {
       const work = this.active.get(hostKey(token));
       if (!work || work.token.scope !== token.scope) throw new Error('host work ownership mismatch');
       work.controller.abort();
     }
     // Completion may have won the race with cancellation. Drop undelivered data
     // too; a queued reply cannot revive a stopped/settled body.
     for (const [key, reply] of this.replies) if (!this.usable(reply.token)) this.replies.delete(key);
   }
   take(): Reply<T> | null {
     for (const [key, reply] of this.replies) {
       this.replies.delete(key);
       if (this.usable(reply.token)) return reply;
     }
     return null;
   }
   wait(): Promise<void> {
     if (this.replies.size || !this.active.size) return Promise.resolve();
     if (!this.notification) {
       let resolve!: () => void;
       const promise = new Promise<void>(done => { resolve = done; });
       this.notification = { promise, resolve };
     }
     return this.notification.promise;
   }
   get counts() { return { pending: this.active.size, replies: this.replies.size }; }
   // Reply data roots are the embedding's responsibility until taken/discarded.
   *values(): Iterable<T> {
     for (const reply of this.replies.values()) if (this.usable(reply.token) && reply.outcome.ok) yield reply.outcome.value;
   }
 }