fix(plugin-browser): finish browser replay follow-ups

packages/plugin-browser/src/connection.ts

@@ -4,7 +4,7 @@ import type { Context } from "@opencode/plugin/effect/plugin"
 import type { RpcRegistration } from "@opencode/plugin/effect/rpc"
 import type { Session } from "@opencode/schema/session"
 import { Tool } from "@opencode/schema/tool"
-import { Deferred, Effect, Option, Schema, Stream } from "effect"
+import { Clock, Deferred, Duration, Effect, Option, Schema, Stream } from "effect"
 import { Browser } from "./rpc.js"
 import { BrowserSkill } from "./skill.js"
 import { BrowserTunnel } from "./tunnel.js"
@@ -31,6 +31,8 @@ export const attached =
 export const replayGrace = "2 seconds"
 export const inventoryGrace = "2 seconds"
 
+type CloseReason = "closed" | "replaced" | "retired"
+
 type Pending = {
   command: Browser.Command
   result: Deferred.Deferred<Browser.Result, Tool.Error>
@@ -41,7 +43,7 @@ type Pending = {
 type Attachment = {
   connectionID: string
   state: Browser.State
-  closed: Deferred.Deferred<"closed" | "replaced">
+  closed: Deferred.Deferred<CloseReason>
   /** Completed by the first `state` publication, which is the first real tab list. */
   inventory: Deferred.Deferred<void>
   inventoryReady: boolean
@@ -66,8 +68,9 @@ export const make = Effect.fn("BrowserConnection.make")(function* (
   const browsers = new Map<Session.ID, Attachment>()
   // In-flight commands waiting for an ordinary reconnect, not a replacement that is already installed.
   const subscribers = new Map<Session.ID, Array<Deferred.Deferred<Attachment, Tool.Error>>>()
-  // Set while a delete or move is settling, so that close cannot park a new waiter.
-  const retired = new Set<Session.ID>()
+  // Session ids that must not replay, and the time that ban lifts. A move or delete
+  // extends the deadline instead of clearing another event's entry.
+  const retiredUntil = new Map<Session.ID, number>()
   // The browser tools and their skill, which an attachment allows together.
   const access = [
     { action: "browser", resource: "*" },
@@ -137,7 +140,7 @@ export const make = Effect.fn("BrowserConnection.make")(function* (
       { discard: true },
     )
   }
-  const close = (sessionID: Session.ID, reason: "closed" | "replaced" = "closed") =>
+  const close = (sessionID: Session.ID, reason: CloseReason = "closed") =>
     Effect.gen(function* () {
       // Drop anyone still parked from an earlier grace window before this close wakes them.
       yield* forget(sessionID)
@@ -153,18 +156,44 @@ export const make = Effect.fn("BrowserConnection.make")(function* (
   }
   // `replaced` already installed the next attachment. `closed` is an ordinary drop:
   // give the desktop one reconnect attempt before reporting that the action did not run.
-  // Attach carries a fresh connection id, not a stable desktop id, so this replay can
-  // land on a different desktop that attaches to the same session.
-  const adopt = (sessionID: Session.ID, previous: Attachment, reason: "closed" | "replaced") =>
+  // Attach carries sessionID, connectionID, and version. connectionID is new on every
+  // registration, and the desktop does not send a stable id, so replay follows the session.
+  const isRetired = (sessionID: Session.ID) =>
+    Effect.gen(function* () {
+      const deadline = retiredUntil.get(sessionID)
+      if (deadline === undefined) return false
+      if ((yield* Clock.currentTimeMillis) < deadline) return true
+      retiredUntil.delete(sessionID)
+      return false
+    })
+  // Holds the ban for one reconnect grace. A later move or delete only lengthens it.
+  const markRetired = (sessionID: Session.ID) =>
+    Effect.gen(function* () {
+      const until = (yield* Clock.currentTimeMillis) + Duration.toMillis(replayGrace)
+      const current = retiredUntil.get(sessionID)
+      if (current === undefined || until > current) retiredUntil.set(sessionID, until)
+      yield* Effect.sleep(replayGrace).pipe(
+        Effect.andThen(
+          Effect.gen(function* () {
+            const deadline = retiredUntil.get(sessionID)
+            if (deadline !== undefined && (yield* Clock.currentTimeMillis) >= deadline) retiredUntil.delete(sessionID)
+          }),
+        ),
+        Effect.forkScoped({ startImmediately: true }),
+      )
+    })
+  const adopt = (sessionID: Session.ID, previous: Attachment, reason: CloseReason) =>
     Effect.gen(function* () {
-      if (retired.has(sessionID)) return undefined
+      // The reason travels with the wake. close only schedules this command, so a
+      // flag cleared by the handler would already be gone by the time we run.
+      if (reason === "retired" || (yield* isRetired(sessionID))) return undefined
       const installed = browsers.get(sessionID)
       if (installed && installed !== previous) return installed
       if (reason === "replaced") return undefined
       const waiter = yield* Deferred.make<Attachment, Tool.Error>()
       subscribers.set(sessionID, [...(subscribers.get(sessionID) ?? []), waiter])
       const raced = browsers.get(sessionID)
-      if (retired.has(sessionID)) {
+      if (yield* isRetired(sessionID)) {
         drop(sessionID, waiter)
         return undefined
       }
@@ -210,14 +239,14 @@ export const make = Effect.fn("BrowserConnection.make")(function* (
               const browser: Attachment = {
                 connectionID: input.connectionID,
                 state: { tabs: [], focusedTabID: null },
-                closed: yield* Deferred.make<"closed" | "replaced">(),
+                closed: yield* Deferred.make<CloseReason>(),
                 inventory: yield* Deferred.make<void>(),
                 inventoryReady: false,
                 pending: new Map(),
                 tunnels: BrowserTunnel.make(),
               }
               browsers.set(input.sessionID, browser)
-              retired.delete(input.sessionID)
+              // A new attachment does not lift a delete or move ban early. The deadline does.
               // Wake reconnect waiters before the previous command observes `replaced` or `closed`.
               yield* announce(input.sessionID, browser)
               if (previous) yield* Deferred.succeed(previous.closed, "replaced")
@@ -229,7 +258,9 @@ export const make = Effect.fn("BrowserConnection.make")(function* (
           yield* rpc.events
             .emit("control", { type: "attached", connectionID: input.connectionID, version: 4 })
             .pipe(Effect.orDie)
-          return yield* Deferred.await(browser.closed)
+          // In-flight commands see "retired". The desktop protocol only has closed and replaced.
+          const outcome = yield* Deferred.await(browser.closed)
+          return outcome === "retired" ? "closed" : outcome
         }).pipe(Effect.scoped),
       state: (input, call) =>
         Effect.gen(function* () {
@@ -290,12 +321,13 @@ export const make = Effect.fn("BrowserConnection.make")(function* (
   yield* ctx.event.subscribe().pipe(
     Stream.filter((event) => event.type === "session.deleted" || event.type === "session.moved"),
     Stream.runForEach((event) =>
-      Effect.sync(() => {
-        retired.add(event.data.sessionID)
-      }).pipe(
-        Effect.andThen(close(event.data.sessionID)),
-        // close wakes an in-flight command, which may park a waiter before this line.
+      markRetired(event.data.sessionID).pipe(
+        Effect.andThen(close(event.data.sessionID, "retired")),
+        // close only schedules the command it wakes. That command reads the retired
+        // reason, so it does not park a reconnect after this handler returns.
         Effect.andThen(forget(event.data.sessionID)),
+        // The id stays retired for replayGrace. Another move or delete extends that
+        // deadline; clearing it here would let a late adopt replay onto a session that came back.
       ),
     ),
     Effect.forkScoped({ startImmediately: true }),
@@ -317,6 +349,7 @@ export const make = Effect.fn("BrowserConnection.make")(function* (
   return {
     status,
     waiting: (sessionID: Session.ID) => subscribers.get(sessionID)?.length ?? 0,
+    retired: (sessionID: Session.ID) => retiredUntil.has(sessionID),
     // Fails before a browser tool reads files or dispatches. Status stays available either way.
     require: Effect.fn("BrowserConnection.require")(function* (sessionID: Session.ID) {
       if (browsers.has(sessionID)) return
@@ -376,10 +409,7 @@ export const make = Effect.fn("BrowserConnection.make")(function* (
 
 const request = Effect.fn("BrowserConnection.request")(function* (
   rpc: RpcRegistration<typeof Browser.Definition>,
-  adopt: (
-    previous: Attachment,
-    reason: "closed" | "replaced",
-  ) => Effect.Effect<Attachment | undefined, Tool.Error>,
+  adopt: (previous: Attachment, reason: CloseReason) => Effect.Effect<Attachment | undefined, Tool.Error>,
   browser: Attachment,
   action: Browser.Action,
   tab: Browser.Tab | undefined,
@@ -463,7 +493,11 @@ const tabAfterInventory = Effect.fn("BrowserConnection.tabAfterInventory")(funct
   action: Browser.Action,
 ) {
   if (!("tabID" in action)) return undefined
-  yield* Deferred.await(next.inventory).pipe(
+  // A desktop that closes while we wait must not hold the command for the whole grace.
+  yield* Effect.raceFirst(
+    Deferred.await(next.inventory),
+    Deferred.await(next.closed).pipe(Effect.andThen(Effect.fail(new Tool.Error({ message: closedBefore })))),
+  ).pipe(
     Effect.timeoutOrElse({
       duration: inventoryGrace,
       orElse: () => Effect.fail(new Tool.Error({ message: closedBefore })),

packages/plugin-browser/test/connection.test.ts

@@ -1,5 +1,5 @@
 import { expect, test } from "bun:test"
-import { Cause, Effect, Exit, Fiber, Queue, Stream } from "effect"
+import { Cause, Deferred, Effect, Exit, Fiber, Option, Queue, Stream } from "effect"
 import { TestClock } from "effect/testing"
 import type { Context } from "@opencode/plugin/effect/plugin"
 import { AbsolutePath } from "@opencode/schema/schema"
@@ -376,7 +376,34 @@ test("a replacement replays a tab command after the new desktop publishes its ta
   )
 })
 
-test("an ordinary reconnect replays a command the desktop never took", async () => {
+test("closing the new desktop ends the tab-list wait", async () => {
+  await Effect.runPromise(
+    Effect.scoped(
+      Effect.gen(function* () {
+        const app = yield* open
+        yield* app.connect("desktop-a")
+        yield* app.methods.state({ sessionID, connectionID: "desktop-a", state }, call)
+        const pending = yield* app
+          .invoke("browser.snapshot", { tabID })
+          .pipe(Effect.forkScoped({ startImmediately: true }))
+        const settled = yield* Deferred.make<Exit.Exit<unknown, unknown>>()
+        yield* Effect.exit(Fiber.join(pending)).pipe(
+          Effect.flatMap((exit) => Deferred.succeed(settled, exit)),
+          Effect.forkScoped({ startImmediately: true }),
+        )
+        yield* app.take("command", "desktop-a")
+        const next = yield* app.connect("desktop-b")
+        yield* Fiber.interrupt(next)
+        yield* Effect.yieldNow
+        const ready = yield* Deferred.poll(settled)
+        if (Option.isNone(ready)) return yield* Effect.die(new Error("tab-list wait ignored the desktop closing"))
+        expect(message(yield* ready.value)).toContain("did not run")
+      }),
+    ).pipe(Effect.provide(TestClock.layer())),
+  )
+})
+
+test("an ordinary reconnect replays onto the new connection id", async () => {
   await Effect.runPromise(
     Effect.scoped(
       Effect.gen(function* () {
@@ -388,6 +415,7 @@ test("an ordinary reconnect replays a command the desktop never took", async ()
         yield* app.connect("desktop-b")
         const replayed = yield* app.take("command", "desktop-b")
         if (replayed.type !== "command") return yield* Effect.die(new Error("expected the reconnected command"))
+        expect(replayed.connectionID).toBe("desktop-b")
         yield* app.methods.command({ sessionID, connectionID: "desktop-b", requestID: replayed.requestID }, call)
         yield* app.methods.result(
           {
@@ -459,6 +487,7 @@ for (const type of ["session.deleted", "session.moved"] as const) {
           yield* app.end(type)
           yield* Effect.yieldNow
           expect(app.connection.waiting(sessionID)).toBe(0)
+          expect(app.connection.retired(sessionID)).toBe(true)
           const failure = message(yield* Effect.exit(Fiber.join(pending)))
           expect(failure).toContain("did not run")
         }),
@@ -467,6 +496,85 @@ for (const type of ["session.deleted", "session.moved"] as const) {
   })
 }
 
+for (const type of ["session.deleted", "session.moved"] as const) {
+  test(`${type} fails an untaken command while the desktop is still attached`, async () => {
+    await Effect.runPromise(
+      Effect.scoped(
+        Effect.gen(function* () {
+          const app = yield* open
+          yield* app.connect("desktop-a")
+          const pending = yield* app.invoke("browser.tabs.list", {}).pipe(Effect.forkScoped({ startImmediately: true }))
+          const settled = yield* Deferred.make<Exit.Exit<unknown, unknown>>()
+          yield* Effect.exit(Fiber.join(pending)).pipe(
+            Effect.flatMap((exit) => Deferred.succeed(settled, exit)),
+            Effect.forkScoped({ startImmediately: true }),
+          )
+          // Seen by the desktop, but not retrieved, so the close reaches adopt.
+          yield* app.take("command", "desktop-a")
+          yield* app.end(type)
+          yield* Effect.yieldNow
+          const ready = yield* Deferred.poll(settled)
+          if (Option.isNone(ready)) return yield* Effect.die(new Error("command waited after the session ended"))
+          const failure = message(yield* ready.value)
+          expect(failure).toContain("did not run")
+          expect(failure).not.toContain("may already have run")
+          expect(app.connection.waiting(sessionID)).toBe(0)
+          // The ban outlives the handler, so this command cannot be replayed onto a later attach.
+          expect(app.connection.retired(sessionID)).toBe(true)
+          const again = yield* app.connect("desktop-b")
+          yield* Effect.yieldNow
+          const replayed: string[] = []
+          let queued = yield* Queue.poll(app.events)
+          while (Option.isSome(queued)) {
+            if (queued.value.type === "command") replayed.push(queued.value.connectionID)
+            queued = yield* Queue.poll(app.events)
+          }
+          expect(replayed).toEqual([])
+          const next = yield* app.invoke("browser.tabs.list", {}).pipe(Effect.forkScoped({ startImmediately: true }))
+          const nextSettled = yield* Deferred.make<Exit.Exit<unknown, unknown>>()
+          yield* Effect.exit(Fiber.join(next)).pipe(
+            Effect.flatMap((exit) => Deferred.succeed(nextSettled, exit)),
+            Effect.forkScoped({ startImmediately: true }),
+          )
+          yield* app.take("command", "desktop-b")
+          yield* Fiber.interrupt(again)
+          yield* Effect.yieldNow
+          const dropped = yield* Deferred.poll(nextSettled)
+          if (Option.isNone(dropped)) return yield* Effect.die(new Error("re-attach drop waited out the ban"))
+          expect(message(yield* dropped.value)).toContain("did not run")
+          expect(app.connection.waiting(sessionID)).toBe(0)
+        }),
+      ).pipe(Effect.provide(TestClock.layer())),
+    )
+  })
+}
+
+test("a later delete extends the ban started by a move", async () => {
+  await Effect.runPromise(
+    Effect.scoped(
+      Effect.gen(function* () {
+        const app = yield* open
+        yield* app.connect("desktop-a")
+        const pending = yield* app.invoke("browser.tabs.list", {}).pipe(Effect.forkScoped({ startImmediately: true }))
+        yield* app.take("command", "desktop-a")
+        yield* app.end("session.moved")
+        yield* Effect.yieldNow
+        expect(message(yield* Effect.exit(Fiber.join(pending)))).toContain("did not run")
+        expect(app.connection.waiting(sessionID)).toBe(0)
+        yield* TestClock.adjust("1 second")
+        yield* app.end("session.deleted")
+        yield* Effect.yieldNow
+        yield* TestClock.adjust("1 second")
+        yield* Effect.yieldNow
+        expect(app.connection.retired(sessionID)).toBe(true)
+        yield* TestClock.adjust("1 second")
+        yield* Effect.yieldNow
+        expect(app.connection.retired(sessionID)).toBe(false)
+      }),
+    ).pipe(Effect.provide(TestClock.layer())),
+  )
+})
+
 test("a close that already has the result keeps that result", async () => {
   await Effect.runPromise(
     Effect.scoped(

packages/plugin-browser/test/rpc.test.ts

@@ -1,6 +1,7 @@
 import { expect, test } from "bun:test"
-import { Browser } from "../src/rpc.js"
 import { Schema } from "effect"
+import { Session } from "@opencode/schema/session"
+import { Browser } from "../src/rpc.js"
 
 const tabID = Browser.TabID.make(`tab_${crypto.randomUUID()}`)
 
@@ -59,6 +60,20 @@ test("network lifecycle and RPC version are explicit", () => {
   expect(Schema.decodeUnknownSync(Browser.Definition.methods.attach.output)("replaced")).toBe("replaced")
 })
 
+test("attach keeps session, connection, and version, and drops a desktop id", () => {
+  const decoded = Schema.decodeUnknownSync(Browser.Definition.methods.attach.input)({
+    sessionID: Session.ID.make("ses_browser"),
+    connectionID: "connection-1",
+    version: 4,
+    desktopID: "desk-1",
+  })
+  expect(decoded.sessionID).toBe(Session.ID.make("ses_browser"))
+  expect(decoded.connectionID).toBe("connection-1")
+  expect(decoded.version).toBe(4)
+  expect("desktopID" in decoded).toBe(false)
+  expect(Object.keys(decoded).sort()).toEqual(["connectionID", "sessionID", "version"])
+})
+
 // Tool search matches query words as substrings of the description, folding a trailing "s"/"es".
 // Each phrase an agent is likely to search for when it wants the user to see a file must hit.
 test.each([