From e176c3958746e93439b19831ca63c6b86822604e Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Fri, 23 Oct 2026 17:18:22 -0500 Subject: [PATCH 1/3] test(muse): a slow or self-timed-out steer ack is delivered Muse acks an accepted turn/steer only once the steer's records are on disk. On a slow disk that ack can take more than 30 s, and Muse Code 1.4.12 may also answer with its own -32603 internal "runtime command acknowledgement timed out after 30s" error and then apply the steer anyway. The adapter treats both as failures: its 30 s request timeout closes the host and fails the active run, and Muse's error fails the steer effect, which the server retries with a fresh command id, so Muse runs the steer twice. Co-Authored-By: Claude Opus 5.5 --- .../provider-muse/src/server/adapter.test.ts | 67 +++++++++++++++++++ 1 file changed, 67 insertions(+) diff --git a/packages/provider-muse/src/server/adapter.test.ts b/packages/provider-muse/src/server/adapter.test.ts index 63e0a1779d205..c6225333fe4a9 100644 --- a/packages/provider-muse/src/server/adapter.test.ts +++ b/packages/provider-muse/src/server/adapter.test.ts @@ -1395,6 +1395,73 @@ describe("MuseAdapterV2", () => { }).pipe(Effect.scoped, Effect.provide(testLayer)), ); + // Muse acks an accepted steer only once its records are on disk; a slow disk has + // held that ack for 33 s. + it.effect("reports a steer delivered and keeps the host while Muse holds its ack", () => + Effect.gen(function* () { + const fake = yield* makeFakeMuse(); + const harness = yield* makeHarness(fake); + const turn = yield* startConversation(harness, fake); + const input = yield* turnInput(harness.providerThread); + fake.queueResponse("turn/steer", pendingPromise>().promise); + const steer = yield* harness.runtime.steerTurn!({ + threadId: THREAD_ID, + runId: input.runId, + providerThread: harness.providerThread, + providerTurnId: turn.providerTurn.id, + message: { ...input.message, text: "Delegated task finished" }, + }).pipe(Effect.forkScoped); + yield* fake.takeCall("turn/steer"); + yield* TestClock.adjust("31 seconds"); + yield* Fiber.join(steer); + assert.strictEqual(fake.closeCount(), 0); + + yield* fake.emit("turn/completed", { turnId: turn.nativeId, terminal: "completed" }); + assert.strictEqual((yield* harness.takeEvent("turn.terminal")).status, "completed"); + }).pipe(Effect.scoped, Effect.provide(testLayer)), + ); + + // Muse Code 1.4.12 answers a steer whose records are not durable within 30 s with this + // error, then applies the steer anyway. Retrying it would deliver the steer twice. + it.effect("reports a steer delivered when Muse times out its own ack", () => + Effect.gen(function* () { + const fake = yield* makeFakeMuse(); + const harness = yield* makeHarness(fake); + const turn = yield* startConversation(harness, fake); + const input = yield* turnInput(harness.providerThread); + const steer = (text: string) => + harness.runtime.steerTurn!({ + threadId: THREAD_ID, + runId: input.runId, + providerThread: harness.providerThread, + providerTurnId: turn.providerTurn.id, + message: { ...input.message, text }, + }); + fake.queueResponse( + "turn/steer", + new MspError({ + code: -32603, + message: "runtime submit failed: turn engine unavailable", + data: { kind: "internal", retryable: false }, + }), + ); + assert.strictEqual((yield* steer("Unrelated failure").pipe(Effect.exit))._tag, "Failure"); + fake.queueResponse( + "turn/steer", + new MspError({ + code: -32603, + message: "runtime submit failed: runtime command acknowledgement timed out after 30s", + data: { kind: "internal", retryable: false }, + }), + ); + yield* steer("Delegated task finished"); + assert.strictEqual(fake.closeCount(), 0); + + yield* fake.emit("turn/completed", { turnId: turn.nativeId, terminal: "completed" }); + assert.strictEqual((yield* harness.takeEvent("turn.terminal")).status, "completed"); + }).pipe(Effect.scoped, Effect.provide(testLayer)), + ); + it.effect("resumes the requested native session and reads the same session snapshot", () => Effect.gen(function* () { const fake = yield* makeFakeMuse(); From 2a545ab4a5b34ea992207bb5373014405a45ca87 Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Fri, 23 Oct 2026 17:18:22 -0500 Subject: [PATCH 2/3] fix(muse): a slow steer acknowledgement no longer closes the host Muse refuses a steer it will not take within milliseconds, and keeps an accepted steer even across a Stop, running it as a new turn. So turn/steer now waits up to the request timeout and, if the ack is still pending, reports the steer delivered and keeps the host. Muse's own "runtime command acknowledgement timed out" error is handled the same way, so the server does not retry it. Every other steer error still fails. The wait stays bounded at the request timeout because Stop and approval answers queue behind the steer effect. Co-Authored-By: Claude Opus 5.5 --- packages/provider-muse/src/server/adapter.ts | 34 +++++++++++++++++--- 1 file changed, 30 insertions(+), 4 deletions(-) diff --git a/packages/provider-muse/src/server/adapter.ts b/packages/provider-muse/src/server/adapter.ts index 4ea37a2cc9eba..04482e2bf6956 100644 --- a/packages/provider-muse/src/server/adapter.ts +++ b/packages/provider-muse/src/server/adapter.ts @@ -354,7 +354,7 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* ( Schema.decodeUnknownEffect(schema)(data).pipe( Effect.mapError((cause) => protocolError("Invalid Muse protocol response", cause)), ); - const request = ( + const send = ( method: string, params: Record, command = true, @@ -382,7 +382,14 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* ( : `Muse ${method} failed`, cause, ), - }).pipe( + }); + const request = ( + method: string, + params: Record, + command = true, + commandId?: string, + ) => + send(method, params, command, commandId).pipe( Effect.timeout(options.requestTimeoutMs ?? 30_000), Effect.catchTags({ TimeoutError: (cause) => @@ -2214,10 +2221,29 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* ( ) return yield* protocolError("This Muse turn cannot be steered"); const parts = yield* prompt(args.message); - const result = yield* request("turn/steer", { + // Muse refuses a steer it will not take within milliseconds, but acks an + // accepted one only once its records are on disk, which a slow disk can + // hold past 30 s. An accepted steer survives the turn (Muse runs it as a + // new turn after a Stop), so an ack still pending after the wait means + // delivered, not a dead host. Muse also gives up on the ack itself after + // 30 s and still applies the steer, so a retry of that error would steer + // twice. + const ack = yield* send("turn/steer", { expectedTurnId: turn.nativeId, input: parts, - }).pipe(Effect.flatMap((value) => decode(MuseTurnStartResult, value))); + }).pipe( + Effect.timeoutOption(options.requestTimeoutMs ?? 30_000), + Effect.catchIf( + (error) => + error.payload instanceof MspError && + error.payload.code === -32603 && + error.payload.kind === "internal" && + error.payload.message.includes("runtime command acknowledgement timed out"), + () => Effect.succeedNone, + ), + ); + if (Option.isNone(ack)) return; + const result = yield* decode(MuseTurnStartResult, ack.value); if (result.turnId !== turn.nativeId) return yield* protocolError("Muse steering targeted a different turn"); }, From bd788bd4827ab84dd8ea6ded434bc268aa778d1b Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Mon, 26 Oct 2026 07:42:52 +0000 Subject: [PATCH 3/3] fix(muse): verify slow steering against current main --- .../provider-muse/src/server/adapter.test.ts | 23 +++++++++++++------ packages/provider-muse/src/server/adapter.ts | 10 +++----- 2 files changed, 19 insertions(+), 14 deletions(-) diff --git a/packages/provider-muse/src/server/adapter.test.ts b/packages/provider-muse/src/server/adapter.test.ts index 15201e6c133b6..c02da189df0d8 100644 --- a/packages/provider-muse/src/server/adapter.test.ts +++ b/packages/provider-muse/src/server/adapter.test.ts @@ -1386,15 +1386,14 @@ describe("MuseAdapterV2", () => { }).pipe(Effect.scoped, Effect.provide(testLayer)), ); - // Muse acks an accepted steer only once its records are on disk; a slow disk has - // held that ack for 33 s. it.effect("reports a steer delivered and keeps the host while Muse holds its ack", () => Effect.gen(function* () { const fake = yield* makeFakeMuse(); const harness = yield* makeHarness(fake); const turn = yield* startConversation(harness, fake); const input = yield* turnInput(harness.providerThread); - fake.queueResponse("turn/steer", pendingPromise>().promise); + const acknowledgement = pendingPromise>(); + fake.queueResponse("turn/steer", acknowledgement.promise); const steer = yield* harness.runtime.steerTurn!({ threadId: THREAD_ID, runId: input.runId, @@ -1406,14 +1405,24 @@ describe("MuseAdapterV2", () => { yield* TestClock.adjust("31 seconds"); yield* Fiber.join(steer); assert.strictEqual(fake.closeCount(), 0); + acknowledgement.resolve({ turnId: turn.nativeId }); + assert.strictEqual(fake.calls.filter((call) => call.method === "turn/steer").length, 1); - yield* fake.emit("turn/completed", { turnId: turn.nativeId, terminal: "completed" }); - assert.strictEqual((yield* harness.takeEvent("turn.terminal")).status, "completed"); + fake.queueResponse("turn/interrupt", pendingPromise>().promise); + const interrupt = yield* harness.runtime + .interruptTurn({ + providerThread: harness.providerThread, + providerTurnId: turn.providerTurn.id, + }) + .pipe(Effect.exit, Effect.forkScoped); + yield* fake.takeCall("turn/interrupt"); + yield* TestClock.adjust("31 seconds"); + assert.strictEqual((yield* Fiber.join(interrupt))._tag, "Failure"); + assert.strictEqual(fake.closeCount(), 1); + assert.strictEqual((yield* harness.takeEvent("turn.terminal")).status, "failed"); }).pipe(Effect.scoped, Effect.provide(testLayer)), ); - // Muse Code 1.4.12 answers a steer whose records are not durable within 30 s with this - // error, then applies the steer anyway. Retrying it would deliver the steer twice. it.effect("reports a steer delivered when Muse times out its own ack", () => Effect.gen(function* () { const fake = yield* makeFakeMuse(); diff --git a/packages/provider-muse/src/server/adapter.ts b/packages/provider-muse/src/server/adapter.ts index bf4f7c6b4f371..7ec0a454f30cb 100644 --- a/packages/provider-muse/src/server/adapter.ts +++ b/packages/provider-muse/src/server/adapter.ts @@ -31,6 +31,7 @@ import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; +import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; import * as Schema from "effect/Schema"; import * as Semaphore from "effect/Semaphore"; @@ -2212,13 +2213,8 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* ( ) return yield* protocolError("This Muse turn cannot be steered"); const parts = yield* prompt(args.message); - // Muse refuses a steer it will not take within milliseconds, but acks an - // accepted one only once its records are on disk, which a slow disk can - // hold past 30 s. An accepted steer survives the turn (Muse runs it as a - // new turn after a Stop), so an ack still pending after the wait means - // delivered, not a dead host. Muse also gives up on the ack itself after - // 30 s and still applies the steer, so a retry of that error would steer - // twice. + // Muse can apply a steer after its durable-write acknowledgement times out. + // Retrying can duplicate it; only other command timeouts close the host. const ack = yield* send("turn/steer", { expectedTurnId: turn.nativeId, input: parts,