ceos/opencode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(core): drop synchronous drain start from the coordinator

Browse source 
Drains now uniformly start one tick after wake or run, removing the
ready-deferred release that ran the first drain synchronously inside
the caller. The synchronous start was observable only by tests that
asserted state immediately after fire-and-forget calls, coupling them
to scheduler internals; those tests now synchronize explicitly through
session.wait or stream-start signals.
This commit is contained in:
Kit Langton 2026-07-02 21:15:15 -04:00
commit 7d598b5610
2 changed files with 10 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

14
packages/core/src/session/run-coordinator.ts
View file

@ -49,29 +49,27 @@ export const make = <Key, E>(options: {
),
)
const start = (key: Key, force: boolean, defer = false) => {
const start = (key: Key, force: boolean) => {
const execution: Execution<E> = { done: Deferred.makeUnsafe<void, E>(), pendingWake: false, stopping: false }
executions.set(key, execution)
const ready = Deferred.makeUnsafe<void>()
// The leading yield lets `owner` be assigned before the drain can settle, and keeps
// failing self-waking executions from growing the stack across successor starts.
// Drains start one tick after wake; callers observe progress through events or run.
execution.owner = fork(
(defer ? Effect.yieldNow : Deferred.await(ready)).pipe(
Effect.yieldNow.pipe(
Effect.andThen(loop(key, execution, force)),
Effect.onExit((exit) => Effect.sync(() => settle(key, execution, exit))),
Effect.exit,
Effect.asVoid,
),
)
// Releasing after `owner` is assigned starts the drain synchronously (a prompt's wake
// reaches the provider in the same tick) without racing ownership. Successor starts
// defer one tick instead so failing self-waking executions cannot grow the stack.
if (!defer) Deferred.doneUnsafe(ready, Effect.void)
return execution
}
// A doorbell that survives the execution loop (rung after the loop decided to end, or
// during failure or interruption cleanup) starts a fresh execution for the remaining work.
const settle = (key: Key, execution: Execution<E>, exit: Exit.Exit<void, E>) => {
if (execution.pendingWake) start(key, false, true)
if (execution.pendingWake) start(key, false)
else executions.delete(key)
Deferred.doneUnsafe(execution.done, exit)
}

5
packages/core/test/session-runner.test.ts
View file

@ -655,6 +655,7 @@ describe("SessionRunnerLLM", () => {
response = []
const message = yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Run automatically" }) })
yield* session.wait(sessionID)
expect(requests).toHaveLength(1)
expect(yield* session.messages({ sessionID })).toMatchObject([
@ -714,6 +715,7 @@ describe("SessionRunnerLLM", () => {
systemUnavailable = false
yield* session.prompt({ id: messageID, sessionID, prompt: Prompt.make({ text: "First" }) })
yield* session.wait(sessionID)
expect(requests).toHaveLength(1)
expect(requests[0]?.messages.map((message) => message.role)).toEqual(["user"])
@ -2537,8 +2539,9 @@ describe("SessionRunnerLLM", () => {
const first = yield* session.resume(sessionID).pipe(Effect.forkChild)
yield* Deferred.await(streamStarted)
streamStarted = yield* Deferred.make<void>()
const second = yield* session.resume(otherSessionID).pipe(Effect.forkChild)
yield* Effect.yieldNow
yield* Deferred.await(streamStarted)
expect(requests).toHaveLength(2)
expect(requests.map((request) => request.providerOptions?.openai?.promptCacheKey)).toEqual([