diff --git a/packages/provider-muse/src/server/adapter.test.ts b/packages/provider-muse/src/server/adapter.test.ts index 7950ef28cb93a..c02da189df0d8 100644 --- a/packages/provider-muse/src/server/adapter.test.ts +++ b/packages/provider-muse/src/server/adapter.test.ts @@ -1386,6 +1386,82 @@ describe("MuseAdapterV2", () => { }).pipe(Effect.scoped, Effect.provide(testLayer)), ); + 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); + const acknowledgement = pendingPromise>(); + fake.queueResponse("turn/steer", acknowledgement.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); + acknowledgement.resolve({ turnId: turn.nativeId }); + assert.strictEqual(fake.calls.filter((call) => call.method === "turn/steer").length, 1); + + 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)), + ); + + 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(); diff --git a/packages/provider-muse/src/server/adapter.ts b/packages/provider-muse/src/server/adapter.ts index b568bbd2d8f96..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"; @@ -347,7 +348,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, @@ -375,7 +376,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) => @@ -2205,10 +2213,24 @@ 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 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, - }).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"); },