io-session.ts (9875B)
1 // Cloneable event session shared by private probes and public native adapters. 2 import { Core, OperationError, type Trees, type Provider } from './io-core.js'; 3 import { available, capabilities, inspect } from './io-profile.js'; 4 import { ProtocolError, decodeAction, canonicalNat } from './io-protocol.js'; 5 import { encodeUTF8, decodeUTF8, UTF8Error } from './io-utf8.js'; 6 import { Events, type Completion } from './io-events.js'; 7 interface Slot { name: 'input' | 'output' | 'error'; poisoned: boolean; eof: boolean; } 8 const faults = new Set(['stale-handle', 'out-of-scope', 'closed-handle', 'wrong-kind', 'internal-resource', 'borrowed-resource', 'busy']); 9 function logical<T>(f: () => T): T { 10 try { return f(); } catch (error) { 11 if (error instanceof Error && faults.has(error.message)) throw new OperationError(error.message); 12 throw error; 13 } 14 } 15 function provider<H>(events: Events, backend: string): Provider<H, Slot> { 16 let outputBudget = 16384; // Attempted bytes across both slots; no replay/refund on uncertainty. 17 return { 18 release: () => { throw new Error('unexpected owned terminal resource'); }, 19 async request(engine, scope, request, signal) { 20 const { tree, wire } = engine, registry = engine.lifecycle.resources, op = `${request.group}/${request.operation}`; 21 const bad = (category: string, progress: 'none' | 'unknown' = 'none') => () => wire.result(false, wire.problem(category, op, '', progress)); 22 // Unsupported names/revisions leave arguments completely opaque. 23 if (!available(request.group, request.revision, request.operation)) return bad('unsupported'); 24 try { 25 if (request.group === 'runtime') { 26 if (request.operation === 'capabilities') { 27 if (tree.observe(request.arguments).node.kind !== 'leaf') return bad('invalid-argument'); 28 return () => wire.result(true, capabilities(wire, backend)); 29 } 30 // Classify in a live turn; defer construction so no newly allocated 31 // handles can be swept between promise resolution and consumption. 32 try { decodeAction(tree, request.arguments); } 33 catch (error) { if (error instanceof ProtocolError) return bad('invalid-argument'); throw error; } 34 return () => wire.result(true, inspect(wire, request.arguments)); 35 } 36 if (request.group === 'utf8') { 37 const readList = (value: H, limit: number, bits: number, scalar: boolean): number[] => { 38 const values: number[] = []; let rest = value; 39 for (;;) { 40 const node = tree.observe(rest).node; 41 if (node.kind === 'leaf') return values; 42 // Reject the excess spine cell before observing its head or tail. 43 if (values.length === limit || node.kind !== 'fork') throw new OperationError('invalid-argument'); 44 const number = canonicalNat(tree, node.left, bits); 45 if (number === undefined || scalar && (number > 0x10ffffn || number >= 0xd800n && number <= 0xdfffn)) throw new OperationError('invalid-argument'); 46 values.push(Number(number)); rest = node.right; 47 } 48 }; 49 if (request.operation === 'encode') { 50 const scalars = readList(request.arguments, 1024, 21, true); 51 const bytes = encodeUTF8(String.fromCodePoint(...scalars)); 52 return () => wire.result(true, wire.list(bytes.map(byte => tree.natural(BigInt(byte))))); 53 } 54 const [final, rest] = wire.pair(request.arguments), [bytes, end] = wire.pair(rest); 55 if (tree.observe(end).node.kind !== 'leaf') throw new OperationError('invalid-argument'); 56 const last = canonicalNat(tree, final, 1); 57 if (last === undefined) throw new OperationError('invalid-argument'); 58 const input = readList(bytes, 4096, 8, false); 59 try { 60 const decoded = decodeUTF8(input, last === 1n); 61 return () => wire.result(true, wire.list([tree.text(decoded.text), wire.list(decoded.carry.map(byte => tree.natural(BigInt(byte))))])); 62 } catch (error) { 63 if (!(error instanceof UTF8Error)) throw error; 64 return () => wire.result(false, wire.problem('invalid-data', op, error.message)); 65 } 66 } 67 if (request.group === 'standard') { 68 if (tree.observe(request.arguments).node.kind !== 'leaf') throw new OperationError('invalid-argument'); 69 const key = ['input', 'output', 'error'].indexOf(request.operation) + 1; 70 return () => wire.result(true, wire.marker('io.v2.resource', key)); 71 } 72 if (request.group === 'resource') { 73 const token = wire.token(request.arguments, 'io.v2.resource'); 74 logical(() => registry.claimClose(scope, token)); 75 throw new Error('unexpected owned terminal close'); 76 } 77 const [handle, argument] = wire.pair(request.arguments), token = wire.token(handle, 'io.v2.resource'); 78 const read = request.operation === 'read'; 79 const lease = logical(() => registry.beginLease(scope, token, read ? 'reader' : 'writer', read ? 'read' : 'write')); 80 try { 81 const slot = lease.backend; 82 if (slot.poisoned) return bad('invalid-state', 'unknown'); 83 let reply: Completion; 84 if (read) { 85 const limit = canonicalNat(tree, argument, 16); 86 if (limit === undefined) throw new OperationError('invalid-argument'); 87 if (limit < 1n || limit > 4096n) return bad('invalid-bound'); 88 if (slot.eof) return () => wire.result(true, wire.list([tree.text('eof')])); 89 reply = await events.read(Number(limit), signal); 90 if (reply.kind === 'read') { 91 const received = reply; slot.eof = received.eof; 92 return () => wire.result(true, received.eof ? wire.list([tree.text('eof')]) : 93 tree.fork(tree.text('data'), wire.list(received.bytes.map(byte => tree.natural(BigInt(byte)))))); 94 } 95 } else { 96 const bytes: number[] = []; let rest = argument; 97 while (tree.observe(rest).node.kind !== 'leaf') { 98 if (bytes.length === 4096) throw new OperationError('invalid-argument'); 99 const [head, tail] = wire.pair(rest), byte = canonicalNat(tree, head, 8); 100 if (byte === undefined) throw new OperationError('invalid-argument'); 101 bytes.push(Number(byte)); rest = tail; 102 } 103 if (!bytes.length) return bad('invalid-bound'); 104 if (bytes.length > outputBudget) return bad('invalid-bound'); 105 outputBudget -= bytes.length; 106 reply = await events.write(slot.name === 'error' ? 'error' : 'output', bytes, signal); 107 if (reply.kind === 'write') { const count = reply.count; return () => wire.result(true, tree.natural(BigInt(count))); } 108 } 109 if (!read) slot.poisoned = true; 110 return bad(reply.kind === 'cancelled' ? 'cancelled' : 'io-error', 'unknown'); 111 } finally { registry.endLease(lease.token); } 112 } catch (error) { 113 if (!(error instanceof OperationError)) throw error; 114 return bad(error.message); 115 } 116 } 117 }; 118 } 119 export type Command = { run: string; kind: 'poll' | 'stop' } | { run: string; kind: 'reply'; reply: unknown }; 120 export class Session<H> { 121 readonly events: Events; 122 readonly core: Core<H, Slot>; 123 private busy = false; 124 private finished = false; 125 constructor(readonly tree: Trees<H>, action: H, readonly run: string, private readonly sweep: (roots: Iterable<H>) => void, private readonly options: { summarize?: boolean; steps?: number; backend?: string } = {}) { 126 if (options.steps !== undefined && (!Number.isInteger(options.steps) || options.steps < 1 || options.steps > 128)) throw new Error('invalid session step bound'); 127 this.events = new Events(run); 128 this.core = new Core(tree, BigInt(run), action, 129 ['input', 'output', 'error'].map(name => ({ name, poisoned: false, eof: false })) as [Slot, Slot, Slot], provider(this.events, options.backend ?? 'browser')); 130 sweep(this.core.roots()); 131 } 132 async turn(command: Command) { 133 if (this.busy) throw new Error('session turn already active'); 134 if (this.finished || command.run !== this.run) throw new Error('stale session run'); 135 if (!['poll', 'stop', 'reply'].includes(command.kind)) throw new Error('invalid session command'); 136 this.busy = true; 137 try { 138 const accepted = command.kind === 'reply' ? this.events.reply(command.reply) : null; 139 if (command.kind === 'stop') this.core.stop(this.tree.leaf); 140 // Never await a terminal operation. Bounded progress batches leave the 141 // transport available for multiple replies, polls and structured stop. 142 let progress = false; 143 for (let turns = 0; turns < (this.options.steps ?? 128) && !this.core.result; turns++) { 144 progress = this.core.advance(); this.sweep(this.core.roots()); 145 await Promise.resolve(); 146 if (!progress) break; 147 } 148 const outcome = this.core.result; 149 if (outcome) this.finished = true; 150 return { run: this.run, accepted, waiting: !progress && this.events.counts.pending > 0 && this.core.work.counts.pending === this.events.counts.pending && this.core.work.counts.replies === 0, done: !!outcome, events: this.events.take(), 151 outcome: outcome ? { kind: outcome.kind, value: outcome.kind === 'completed' && this.options.summarize !== false ? canonicalNat(this.tree, outcome.value, 64)?.toString() ?? null : null, 152 cleanup: outcome.cleanup?.items.length ?? 0, fault: this.core.hasFault } : null, 153 counts: { tasks: this.core.lifecycle.tasks.counts, resources: this.core.lifecycle.resources.counts, 154 work: this.core.work.counts, events: this.events.counts } }; 155 } finally { this.busy = false; } 156 } 157 }