diff --git a/packages/codemode/codemode.md b/packages/codemode/codemode.md index d75e64e14c..ce08add6e8 100644 --- a/packages/codemode/codemode.md +++ b/packages/codemode/codemode.md @@ -121,8 +121,6 @@ closing. Timeout and external interruption cancel immediately instead. ### Confirmed defects -- [ ] Make every `await` continuation asynchronous. Awaiting a plain or already-settled value currently resumes in the - same scheduling turn and can reorder state mutation relative to JavaScript. - [ ] Return rejected promises for invalid `Promise.all`/`allSettled`/`race` inputs instead of throwing during the call. - [ ] Align handler callability with the values CodeMode reports as functions, or document the narrower callback allowlist. For example, unsupported constructor-like callables are currently treated as absent handlers. @@ -151,13 +149,14 @@ closing. Timeout and external interruption cancel immediately instead. loser. - Timeouts interrupt all in-flight promise fibers with parallel teardown, while host interruption propagates instead of becoming a diagnostic. -- Awaiting the same promise twice settles it once, and basic `then` handlers run after synchronous statements. +- Promise reactions and plain, pending, or settled `await` continuations start in deterministic FIFO order. Nested + reactions preserve enqueue order, while an async reaction can suspend without blocking the next queued reaction. +- Awaiting the same promise twice settles it once. ### Missing coverage - Nested unreturned work from `catch` and `finally` handlers. - Abandoned chained and combinator rejections. -- Plain-value and already-settled `await` ordering. - External interruption while handled pending work remains. - Never-settling race losers and fail-fast `Promise.all` siblings under an explicit timeout. - Shared or duplicate promises across combinators, discarded inner chains, and detailed reaction ordering. diff --git a/packages/codemode/src/interpreter/runtime.ts b/packages/codemode/src/interpreter/runtime.ts index fd3c79c61c..369df4e2fc 100644 --- a/packages/codemode/src/interpreter/runtime.ts +++ b/packages/codemode/src/interpreter/runtime.ts @@ -1,5 +1,5 @@ import { parse } from "acorn" -import { Cause, Effect, Exit, Fiber, Scope, Semaphore } from "effect" +import { Cause, Deferred, Effect, Exit, Fiber, Queue, Scope, Semaphore } from "effect" import { DiagnosticCategory, ModuleKind, ScriptTarget, flattenDiagnosticMessageText, transpileModule } from "typescript" import { copyIn, @@ -148,6 +148,11 @@ const parseProgram = (code: string): ProgramNode => { return parsed as ProgramNode } +type PromiseReaction = { + readonly effect: Effect.Effect + readonly settlement: Deferred.Deferred +} + const publicErrorMessage = (message: string): string => message.replace(/\/(?:Users|home|private|tmp|var\/folders)\/[^\s"'`]+/g, "") @@ -610,6 +615,7 @@ class Interpreter { // Every promise fiber belongs to the execution rather than the async function or handler // that happened to create it. The execution scope still interrupts all work on teardown. private readonly promiseScope: Scope.Scope + private readonly reactionQueue: Queue.Queue> // Caps how many eagerly forked tool calls run at once (the parallel-call concurrency cap). private readonly callPermits: Semaphore.Semaphore // Fiber-backed promises whose settlement no program construct has observed yet. Ordinary @@ -621,6 +627,7 @@ class Interpreter { invokeTool: (path: ReadonlyArray, args: Array) => Effect.Effect, toolKeys: (path: ReadonlyArray) => ReadonlyArray, promiseScope: Scope.Scope, + reactionQueue: Queue.Queue>, logs: Array = [], shared?: { callPermits: Semaphore.Semaphore; pendingSettlements: Set }, ) { @@ -631,6 +638,7 @@ class Interpreter { this.logs = logs this.lastValue = undefined this.promiseScope = promiseScope + this.reactionQueue = reactionQueue this.callPermits = shared?.callPermits ?? Semaphore.makeUnsafe(TOOL_CALL_CONCURRENCY) this.pendingSettlements = shared?.pendingSettlements ?? new Set() globalScope.set("tools", { mutable: false, value: new ToolReference([]) }) @@ -775,6 +783,42 @@ class Interpreter { return Effect.flatMap(this.observePromise(promise), (exit) => self.unwrapPromiseExit(exit)) } + private createReaction( + source: Effect.Effect, never, R>, + reaction: (exit: Exit.Exit, chained: SandboxPromise) => Effect.Effect, + ): Effect.Effect { + const settlement = Deferred.makeUnsafe() + const chained = new SandboxPromise(undefined, Deferred.await(settlement)) + this.pendingSettlements.add(chained) + return Effect.as( + Effect.forkIn( + Effect.flatMap(source, (exit) => + Effect.sync(() => + Queue.offerUnsafe(this.reactionQueue, { + settlement, + effect: Effect.suspend(() => reaction(exit, chained)), + }), + ), + ), + this.promiseScope, + { startImmediately: true }, + ), + chained, + ) + } + + private awaitValue(value: unknown): Effect.Effect { + return Effect.flatMap( + this.createReaction( + value instanceof SandboxPromise + ? Effect.exit(this.settlePromise(value)) + : Effect.succeed(Exit.succeed(value)), + (exit) => this.unwrapPromiseExit(exit), + ), + (continuation) => this.settlePromise(continuation), + ) + } + private unwrapPromiseExit(exit: Exit.Exit): Effect.Effect { if (Exit.isSuccess(exit)) return Effect.succeed(exit.value) return Effect.failCause(exit.cause) @@ -1532,12 +1576,9 @@ class Interpreter { case "UpdateExpression": return this.evaluateUpdateExpression(node) case "AwaitExpression": { - // `await` resolves a promise value; awaiting anything else is a passthrough no-op, - // matching real JS semantics for non-thenables. - const self = this - return Effect.flatMap(this.evaluateExpression(getNode(node, "argument")), (value) => - value instanceof SandboxPromise ? self.settlePromise(value) : Effect.succeed(value), - ) + // Even an already-settled value resumes in a later reaction turn, after work that was + // already queued, matching JavaScript's await continuation ordering. + return Effect.flatMap(this.evaluateExpression(getNode(node, "argument")), (value) => this.awaitValue(value)) } case "NewExpression": return this.evaluateNewExpression(node) @@ -2328,7 +2369,7 @@ class Interpreter { } private invokeFunction(fn: CodeModeFunction, args: Array): Effect.Effect { - const invocation = new Interpreter(this.invokeTool, this.toolKeys, this.promiseScope, this.logs, { + const invocation = new Interpreter(this.invokeTool, this.toolKeys, this.promiseScope, this.reactionQueue, this.logs, { callPermits: this.callPermits, pendingSettlements: this.pendingSettlements, }) @@ -2428,7 +2469,7 @@ class Interpreter { !(value instanceof ErrorConstructorReference) ) return undefined - return (callbackArgs: Array) => + return (callbackArgs: Array, chained: SandboxPromise) => Effect.flatMap( value instanceof ToolReference ? this.createToolCallPromise(value.path, callbackArgs) @@ -2469,27 +2510,24 @@ class Interpreter { }, ) } - let chained: SandboxPromise | undefined const onFulfilled = name === "then" ? callback(args[0]) : undefined const onRejected = name === "then" ? callback(args[1]) : name === "catch" ? callback(args[0]) : undefined const onFinally = name === "finally" ? callback(args[0]) : undefined const settlement = Effect.exit(this.settlePromise(promise)) - return Effect.map( - this.createPromise( + return this.createReaction( + settlement, + (exit, chained) => Effect.gen(function* () { - yield* Effect.yieldNow - const exit = yield* settlement - if (onFinally !== undefined) yield* onFinally([]) + if (onFinally !== undefined) yield* onFinally([], chained) if (Exit.isSuccess(exit)) { if (onFulfilled === undefined) return exit.value - return yield* onFulfilled([exit.value]) + return yield* onFulfilled([exit.value], chained) } - if (onRejected !== undefined) return yield* onRejected([caughtErrorValue(Cause.squash(exit.cause))]) + if (onRejected !== undefined) + return yield* onRejected([caughtErrorValue(Cause.squash(exit.cause))], chained) return yield* Effect.failCause(exit.cause) }), - ), - (promise) => (chained = promise), ) } @@ -3580,8 +3618,20 @@ export const executeWithLimits = >( const operation = Effect.scoped( Effect.gen(function* () { const scope = yield* Effect.acquireRelease(Scope.make("parallel"), (scope, exit) => Scope.close(scope, exit)) + const reactionQueue = yield* Queue.unbounded>>() + yield* Effect.forever( + Effect.gen(function* () { + const reaction = yield* Queue.take(reactionQueue) + yield* Effect.yieldNow + yield* Effect.forkIn( + Effect.flatMap(Effect.exit(reaction.effect), (exit) => Deferred.done(reaction.settlement, exit)), + scope, + { startImmediately: true }, + ) + }), + ).pipe(Effect.forkIn(scope, { startImmediately: true })) const program = parseProgram(options.code) - const interpreter = new Interpreter>(tools.invoke, tools.keys, scope, logs) + const interpreter = new Interpreter>(tools.invoke, tools.keys, scope, reactionQueue, logs) const value = yield* interpreter.run(program) const result = copyOut(copyIn(value, "Execution result"), true) as DataValue return { diff --git a/packages/codemode/test/promise.test.ts b/packages/codemode/test/promise.test.ts index 522ada193c..6e5a5e07e8 100644 --- a/packages/codemode/test/promise.test.ts +++ b/packages/codemode/test/promise.test.ts @@ -699,6 +699,65 @@ describe("promise chaining", () => { ).toEqual(["sync", "then"]) }) + test("nested reactions run before downstream reactions queued later", async () => { + expect( + await value(` + const order = [] + await Promise.resolve() + .then(() => { + order.push(1) + Promise.resolve().then(() => order.push(2)) + }) + .then(() => order.push(3)) + return order + `), + ).toEqual([1, 2, 3]) + }) + + test("plain and settled await resume after reactions that are already queued", async () => { + expect( + await value(` + const order = [] + Promise.resolve().then(() => order.push(1)) + await 0 + order.push(2) + Promise.resolve().then(() => order.push(3)) + await Promise.resolve() + order.push(4) + return order + `), + ).toEqual([1, 2, 3, 4]) + }) + + test("reactions registered on the same pending promise preserve order", async () => { + expect( + await value(` + const order = [] + const pending = tools.host.sleepy({ id: 1 }) + pending.then(() => order.push(1)) + pending.then(() => order.push(2)) + await pending + return order + `), + ).toEqual([1, 2]) + }) + + test("an async reaction does not block the next queued reaction", async () => { + expect( + await value(` + const order = [] + const first = Promise.resolve().then(async () => { + order.push(1) + await tools.host.sleepy({ id: 1 }) + order.push(3) + }) + Promise.resolve().then(() => order.push(2)) + await first + return order + `), + ).toEqual([1, 2, 3]) + }) + test("catch receives normalized errors and recovers the chain", async () => { expect(await value(`return tools.host.fail({}).catch((error) => error.message)`)).toBe("Lookup refused") expect(