io-core.ts (13904B)
1 // Private resumable tree interpreter. No public session dispatch or capabilities. 2 // Adapters must retain roots() between turns and keep temporary evaluator handles 3 // alive for a whole turn. Provider promises acknowledge physical completion; 4 // providers return deferred Result encoders, never task outcomes. Encoding runs 5 // in an interpreter turn, not in a promise/GC gap; cancelled encoders never run. 6 import { canonicalNat, decodeAction, ProtocolError, type Reader, type Action } from './io-protocol.js'; 7 import { lifecycle, type Resource } from './io-resources.js'; 8 import { HostWork } from './io-host-work.js'; 9 import type { Token, Outcome, Diagnostics } from './io-tasks.js'; 10 export interface Trees<H> extends Reader<H> { 11 readonly leaf: H; fork(a: H, b: H): H; stem(a: H): H; app(f: H, x: H): H; 12 natural(value: bigint): H; text(value: string): H; 13 } 14 type Request<H> = Extract<Action<H>, { kind: 'request' }>; 15 type Completion<H> = { kind: 'body'; encode: () => H } | { kind: 'release'; errors: Diagnostics<H> }; 16 type Body<H> = { action: H; frames: H[]; mode: 'runnable' | 'lexical' | 'join' | 'host' | 'done' }; 17 const operationErrors = new Set(['stale-task', 'join-cycle', 'out-of-scope', 'stale-handle', 'not-owner', 18 'borrowed-resource', 'internal-resource', 'closed-handle', 'busy', 'duplicate-handle', 'closed-scope', 'invalid-argument']); 19 export class OperationError extends Error {} 20 function logical<T>(operation: () => T): T { 21 try { return operation(); } 22 catch (error) { 23 if (error instanceof Error && operationErrors.has(error.message)) throw new OperationError(error.message); 24 throw error; 25 } 26 } 27 export class Wire<H> { 28 constructor(readonly tree: Trees<H>, readonly identity: bigint) {} 29 pair(value: H): [H, H] { 30 const node = this.tree.observe(value).node; 31 if (node.kind !== 'fork') throw new OperationError('invalid-argument'); 32 return [node.left, node.right]; 33 } 34 list(values: readonly H[]): H { let out = this.tree.leaf; for (let i = values.length - 1; i >= 0; i--) out = this.tree.fork(values[i]!, out); return out; } 35 result(ok: boolean, value: H): H { return this.tree.fork(ok ? this.tree.stem(this.tree.leaf) : this.tree.leaf, this.tree.fork(value, this.tree.leaf)); } 36 problem(category: string, operation: string, detail = '', progress: 'none' | 'unknown' = 'none'): H { 37 return this.list([category, operation, detail, progress].map(s => this.tree.text(s))); 38 } 39 marker(name: string, key: number): H { return this.list([this.tree.text(name), this.tree.natural(this.identity), this.tree.natural(BigInt(key))]); } 40 token(value: H, name: string): Token { 41 const [tag, rest] = this.pair(value), [run, tail] = this.pair(rest), [key, end] = this.pair(tail); 42 if (this.tree.observe(end).node.kind !== 'leaf') throw new OperationError('invalid-argument'); 43 let cursor = tag; 44 for (const char of name) { 45 const [head, next] = this.pair(cursor); 46 if (canonicalNat(this.tree, head, 7) !== BigInt(char.charCodeAt(0))) throw new OperationError('invalid-argument'); 47 cursor = next; 48 } 49 if (this.tree.observe(cursor).node.kind !== 'leaf') throw new OperationError('invalid-argument'); 50 const r = canonicalNat(this.tree, run, 64), k = canonicalNat(this.tree, key, 64); 51 if (r === undefined || k === undefined) throw new OperationError('invalid-argument'); 52 // Unrepresentable keys cannot alias a live JS identity. Keep the ordinary 53 // stale-handle/task path rather than rounding into a different live handle. 54 return { run: r, key: k > BigInt(Number.MAX_SAFE_INTEGER) ? -1 : Number(k) }; 55 } 56 outcome(value: Outcome<H>): H { 57 const name = this.tree.text(value.kind); 58 if (value.kind === 'completed') return this.list([name, value.value]); 59 const errors = [...value.cleanup?.items ?? []], omitted = value.cleanup?.omitted ?? 0; 60 if (omitted) errors.push(this.problem('cleanup-omitted', 'close', String(omitted))); 61 return this.list([name, value.value, this.list(errors)]); 62 } 63 } 64 export interface Provider<H, B> { 65 request(engine: Core<H, B>, scope: number, request: Request<H>, signal: AbortSignal): (() => H) | Promise<() => H>; 66 release(resource: Resource<B>): void | Promise<void>; 67 roots?(): Iterable<H>; 68 } 69 export class Core<H, B> { 70 readonly wire: Wire<H>; 71 readonly lifecycle: ReturnType<typeof lifecycle<H, B>>; 72 readonly work: HostWork<H, Completion<H>>; 73 private readonly bodies = new Map<number, Body<H>>(); 74 private readonly constants: { stopped: H; cycle: H; fault: H; cleanup: H; budget: H }; 75 private readonly pendingArguments = new Map<number, H>(); 76 private cursor = -1; 77 private actions = 0; 78 result: Outcome<H> | null = null; 79 // Embedding/evaluator failure is retained separately and never stringified or 80 // mistaken for an ordinary Result. Drain cleanup before rethrowing at boundary. 81 fault: unknown = undefined; 82 hasFault = false; 83 constructor(readonly tree: Trees<H>, identity: bigint, action: H, standard: readonly [B, B, B], 84 private readonly provider: Provider<H, B>, private readonly actionLimit = 65536) { 85 if (!Number.isSafeInteger(actionLimit) || actionLimit < 1) throw new Error('invalid action limit'); 86 this.wire = new Wire(tree, identity); 87 this.constants = { 88 stopped: this.wire.problem('cancelled', 'scope', 'owner stopped'), 89 cycle: this.wire.problem('join-cycle', 'scope'), fault: this.wire.problem('runtime-error', 'core'), 90 cleanup: this.wire.problem('cleanup-failed', 'resource/release', '', 'unknown'), budget: this.wire.problem('limit', 'core', 'IO action limit exceeded') }; 91 this.lifecycle = lifecycle(identity, this.constants.stopped, this.constants.cycle, standard); 92 this.work = new HostWork(this.lifecycle.tasks); 93 this.bodies.set(0, this.body(action)); 94 } 95 private body(action: H): Body<H> { return { action, frames: [], mode: 'runnable' }; } 96 stop(reason: H): void { if (!this.result) { this.lifecycle.tasks.stop(0, { kind: 'cancelled', value: reason }); this.work.cancelStopped(); } } 97 private done(scope: number, outcome: Outcome<H>): void { 98 this.lifecycle.tasks.bodyDone(scope, outcome); 99 const body = this.bodies.get(scope)!; body.mode = 'done'; body.action = this.tree.leaf; body.frames = []; 100 } 101 private value(scope: number, value: H): void { 102 const body = this.bodies.get(scope)!; 103 if (body.frames.length) { body.action = this.tree.app(body.frames.pop()!, value); body.mode = 'runnable'; } 104 else this.done(scope, { kind: 'completed', value }); 105 } 106 private operation(scope: number, name: string, action: () => void): void { 107 try { action(); } 108 catch (error) { 109 if (!(error instanceof OperationError)) throw error; 110 this.value(scope, this.wire.result(false, this.wire.problem(error.message, name))); 111 } 112 } 113 private step(scope: number): boolean { 114 const { tasks, resources } = this.lifecycle, body = this.bodies.get(scope)!, state = tasks.state(scope); 115 if (body.mode === 'done' || body.mode === 'lexical') return false; 116 if (state.reason) { this.done(scope, state.reason); return true; } 117 if (body.mode === 'host') return false; 118 if (body.mode === 'join') { 119 const outcome = tasks.pollJoin(scope); if (!outcome) return false; 120 this.value(scope, this.wire.result(true, this.wire.outcome(outcome))); return true; 121 } 122 if (++this.actions > this.actionLimit) { tasks.stop(0, { kind: 'failed', value: this.constants.budget }); return true; } 123 let action: Action<H>; 124 try { action = decodeAction(this.tree, body.action); } 125 catch (error) { 126 if (!(error instanceof ProtocolError)) throw error; 127 this.done(scope, { kind: 'failed', value: this.wire.problem('protocol-error', 'core/decode', error.message) }); return true; 128 } 129 if (action.kind === 'request') { 130 // Keep action/arguments rooted until provider acknowledgement, including 131 // cancellation. The abandoned continuation is cleared by done(). 132 body.mode = 'host'; 133 const request = action; this.pendingArguments.set(scope, request.arguments); 134 this.work.start(scope, async signal => ({ kind: 'body', encode: await this.provider.request(this, scope, request, signal) })); 135 return true; 136 } 137 const p = action.payload; 138 switch (action.code) { 139 case 0: this.value(scope, p); break; 140 case 1: { 141 const node = this.tree.observe(p).node; 142 if (node.kind !== 'fork') this.done(scope, { kind: 'failed', value: this.wire.problem('protocol-error', 'core/decode', 'invalid IO computation') }); 143 else { body.frames.push(node.right); body.action = node.left; } 144 break; 145 } 146 case 2: this.done(scope, { kind: 'failed', value: p }); break; 147 case 9: this.done(scope, { kind: 'cancelled', value: p }); break; 148 case 3: { 149 const child = tasks.nested(scope); this.bodies.set(child, this.body(this.tree.leaf)); body.mode = 'lexical'; body.action = this.tree.leaf; 150 this.bodies.get(child)!.action = this.tree.app(p, this.wire.marker('io.v2.scope', child)); break; 151 } 152 case 4: this.value(scope, this.wire.marker('io.v2.scope', scope)); break; 153 case 5: this.operation(scope, 'core/spawn', () => { 154 const [sup, a] = this.wire.pair(p), [keep, b] = this.wire.pair(a), [handles, callback] = this.wire.pair(b); 155 const supervised = canonicalNat(this.tree, sup, 1), retained = canonicalNat(this.tree, keep, 1); 156 if (supervised === undefined || retained === undefined) throw new OperationError('invalid-argument'); 157 const tokens: Token[] = []; let rest = handles; 158 while (this.tree.observe(rest).node.kind !== 'leaf') { 159 if (tokens.length === 256) throw new OperationError('invalid-argument'); 160 const [head, tail] = this.wire.pair(rest); tokens.push(this.wire.token(head, 'io.v2.resource')); rest = tail; 161 } 162 const moved = logical(() => resources.spawn(scope, supervised === 1n, retained === 1n, tokens)), child = moved.task.key; 163 this.bodies.set(child, this.body(this.tree.leaf)); 164 this.bodies.get(child)!.action = this.tree.app(callback, this.wire.list(moved.handles.map(t => this.wire.marker('io.v2.resource', t.key)))); 165 this.value(scope, this.wire.result(true, this.wire.list([this.wire.marker('io.v2.task', child), retained === 1n ? this.wire.marker('io.v2.join', child) : this.tree.leaf]))); 166 }); break; 167 case 6: case 7: this.operation(scope, action.code === 6 ? 'core/join' : 'core/discard', () => { 168 const token = this.wire.token(p, 'io.v2.join'); 169 if (action.code === 7) { logical(() => tasks.discard(scope, token)); this.value(scope, this.wire.result(true, this.tree.leaf)); } 170 else { logical(() => tasks.claimJoin(scope, token)); body.mode = 'join'; body.action = this.tree.leaf; } 171 }); break; 172 case 8: this.operation(scope, 'core/cancel', () => { 173 const [handle, reason] = this.wire.pair(p), token = this.wire.token(handle, 'io.v2.task'); logical(() => tasks.cancel(scope, token, reason)); 174 this.value(scope, this.wire.result(true, this.tree.leaf)); 175 }); break; 176 } 177 return true; 178 } 179 private settled(scope: number, errors: Diagnostics<H>): void { 180 const { tasks } = this.lifecycle, state = tasks.state(scope), outcome = tasks.settle(scope, errors); 181 this.bodies.delete(scope); 182 if (scope === 0) this.result = outcome; 183 else if (state.kind === 'lexical') { 184 const parent = state.parent!, reason = tasks.state(parent).reason; 185 this.bodies.get(parent)!.mode = 'runnable'; 186 if (reason) this.done(parent, reason); 187 else if (outcome.kind === 'completed') this.value(parent, outcome.value); 188 else this.done(parent, outcome); 189 } 190 } 191 // One progress turn, never waits for a provider. False means wait for host 192 // acknowledgement. The embedding must yield to its event loop between batches. 193 advance(): boolean { 194 if (this.result) return false; 195 try { 196 this.work.cancelStopped(); 197 for (const scope of this.pendingArguments.keys()) 198 if (this.lifecycle.tasks.state(scope).hostCalls === 0) this.pendingArguments.delete(scope); 199 const reply = this.work.take(); 200 if (reply) { 201 if (!reply.outcome.ok) throw reply.outcome.error; 202 const value = reply.outcome.value; 203 if (value.kind === 'release') this.settled(reply.token.scope, value.errors); 204 else this.value(reply.token.scope, value.encode()); 205 return true; 206 } 207 const { tasks, resources } = this.lifecycle; 208 for (const scope of tasks.scopeIds()) if (tasks.state(scope).phase === 'joining' && tasks.beginRelease(scope)) { 209 this.work.release(scope, async () => ({ kind: 'release', errors: await resources.releaseScope(scope, 210 resource => this.provider.release(resource), () => this.constants.cleanup, this.constants.cleanup) })); 211 return true; 212 } 213 const ids = tasks.scopeIds(), ordered = [...ids.filter(id => id > this.cursor), ...ids.filter(id => id <= this.cursor)]; 214 for (const scope of ordered) if (this.step(scope)) { this.cursor = scope; return true; } 215 return false; 216 } catch (error) { 217 if (!this.hasFault) this.fault = error; 218 this.hasFault = true; 219 this.lifecycle.tasks.stop(0, { kind: 'failed', value: this.constants.fault }); 220 this.work.cancelStopped(); return true; 221 } 222 } 223 wait(): Promise<void> { return this.work.wait(); } 224 *roots(): Iterable<H> { 225 yield this.tree.leaf; yield* Object.values(this.constants); yield* this.lifecycle.tasks.payloads(); 226 yield* this.pendingArguments.values(); yield* this.provider.roots?.() ?? []; 227 for (const body of this.bodies.values()) { yield body.action; yield* body.frames; } 228 for (const value of this.work.values()) { 229 if (value.kind === 'release') yield* value.errors.items; 230 } 231 if (this.result) { yield this.result.value; yield* this.result.cleanup?.items ?? []; } 232 } 233 }