tricu

io-events.ts (4772B)

---

 // Cloneable transport only: no evaluator handles, task tokens or continuations.
 // A reply acknowledges physical provider completion, not merely abort delivery.
 export type Event = { run: string; id: number } & (
   { kind: 'read'; limit: number } | { kind: 'write'; stream: 'output' | 'error'; bytes: number[] } |
   { kind: 'cancel'; operation: 'read' | 'write' });
 export type Completion = { kind: 'read'; bytes: number[]; eof: boolean } |
   { kind: 'write'; count: number } | { kind: 'cancelled' } | { kind: 'error' };
 interface Pending {
   event: Event; delivered: boolean; cancelling: boolean;
   resolve(value: Completion): void; remove(): void;
 }
 function bytes(value: unknown, limit: number): value is number[] {
   return Array.isArray(value) && value.length <= limit && value.every(x => Number.isInteger(x) && x >= 0 && x <= 255);
 }
 export class Events {
   private next = 1;
   private readonly pending = new Map<number, Pending>();
   private queue: Event[] = [];
   constructor(readonly run: string, private readonly limit = Number.MAX_SAFE_INTEGER) {
     if (!/^[1-9][0-9]*$/.test(run) || run.length > 20 || BigInt(run) > 0xffffffffffffffffn ||
         !Number.isSafeInteger(limit) || limit < 1) throw new Error('invalid event identity');
   }
   get counts() { return { pending: this.pending.size, queued: this.queue.length }; }
   read(limit: number, signal: AbortSignal): Promise<Completion> {
     if (!Number.isInteger(limit) || limit < 1 || limit > 4096) throw new Error('invalid read bound');
     return this.start({ run: this.run, id: 0, kind: 'read', limit }, signal);
   }
   write(stream: 'output' | 'error', value: readonly number[], signal: AbortSignal): Promise<Completion> {
     if (!['output', 'error'].includes(stream) || !bytes(value, 4096) || !value.length) throw new Error('invalid write chunk');
     return this.start({ run: this.run, id: 0, kind: 'write', stream, bytes: [...value] }, signal);
   }
   private start(event: Event, signal: AbortSignal): Promise<Completion> {
     if (signal.aborted) return Promise.resolve({ kind: 'cancelled' });
     if (this.next > this.limit) throw new Error('event identity exhausted');
     event.id = this.next++;
     return new Promise(resolve => {
       const cancel = () => {
         const entry = this.pending.get(event.id); if (!entry || entry.cancelling) return;
         entry.cancelling = true;
         if (!entry.delivered) this.finish(event.id, { kind: 'cancelled' });
         else this.queue.push({ run: this.run, id: event.id, kind: 'cancel', operation: event.kind as 'read' | 'write' });
       };
       this.pending.set(event.id, { event, delivered: false, cancelling: false, resolve,
         remove: () => signal.removeEventListener('abort', cancel) });
       signal.addEventListener('abort', cancel, { once: true }); this.queue.push(event);
     });
   }
   take(): Event[] {
     const out = this.queue; this.queue = [];
     for (const event of out) this.pending.get(event.id)!.delivered = true;
     // Never expose the mutable validation record or its byte array to the peer.
     return out.map(event => event.kind === 'write' ? { ...event, bytes: [...event.bytes] } : { ...event });
   }
   private finish(id: number, value: Completion): void {
     const entry = this.pending.get(id)!;
     this.pending.delete(id); this.queue = this.queue.filter(event => event.id !== id); entry.remove();
     entry.resolve(entry.cancelling ? { kind: 'cancelled' } : value);
   }
   // Wrong run/kind/bounds and stale/duplicate replies never consume a live wait.
   // A valid normal completion can race with cancellation: it acknowledges work,
   // but its data is discarded. No extra cancellation acknowledgement is needed.
   reply(value: unknown): boolean {
     if (!value || typeof value !== 'object') return false;
     const v = value as Record<string, unknown>;
     if (v.run !== this.run || typeof v.id !== 'number' || !Number.isSafeInteger(v.id)) return false;
     const entry = this.pending.get(v.id); if (!entry || !entry.delivered) return false;
     let completion: Completion;
     if (v.kind === 'cancelled' && entry.cancelling) completion = { kind: 'cancelled' };
     else if (v.kind === 'error') completion = { kind: 'error' };
     else if (v.kind === 'read' && entry.event.kind === 'read' && bytes(v.bytes, entry.event.limit) &&
         typeof v.eof === 'boolean' && (v.eof ? v.bytes.length === 0 : v.bytes.length > 0))
       completion = { kind: 'read', bytes: [...v.bytes], eof: v.eof };
     else if (v.kind === 'write' && entry.event.kind === 'write' && typeof v.count === 'number' &&
         Number.isInteger(v.count) && v.count > 0 && v.count <= entry.event.bytes.length)
       completion = { kind: 'write', count: v.count };
     else return false;
     this.finish(v.id, completion); return true;
   }
 }