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