fix(muse): a slow steer acknowledgement no longer fails the run

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<Record<string, unknown>>();
+      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<Record<string, unknown>>().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();

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<string, unknown>,
         command = true,
@@ -375,7 +376,14 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* (
                 : `Muse ${method} failed`,
               cause,
             ),
-        }).pipe(
+        });
+      const request = (
+        method: string,
+        params: Record<string, unknown>,
+        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");
           },