diff --git a/packages/plugin-browser/src/connection.ts b/packages/plugin-browser/src/connection.ts index 4c63e2eb72b4..32983c51d446 100644 --- a/packages/plugin-browser/src/connection.ts +++ b/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 @@ -41,7 +43,7 @@ type Pending = { type Attachment = { connectionID: string state: Browser.State - closed: Deferred.Deferred<"closed" | "replaced"> + closed: Deferred.Deferred /** Completed by the first `state` publication, which is the first real tab list. */ inventory: Deferred.Deferred inventoryReady: boolean @@ -66,8 +68,9 @@ export const make = Effect.fn("BrowserConnection.make")(function* ( const browsers = new Map() // In-flight commands waiting for an ordinary reconnect, not a replacement that is already installed. const subscribers = new Map>>() - // Set while a delete or move is settling, so that close cannot park a new waiter. - const retired = new Set() + // 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() // 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() 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(), inventory: yield* Deferred.make(), 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, - adopt: ( - previous: Attachment, - reason: "closed" | "replaced", - ) => Effect.Effect, + adopt: (previous: Attachment, reason: CloseReason) => Effect.Effect, 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 })), diff --git a/packages/plugin-browser/test/connection.test.ts b/packages/plugin-browser/test/connection.test.ts index d26fe3531296..57fb31736fd7 100644 --- a/packages/plugin-browser/test/connection.test.ts +++ b/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>() + 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>() + 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>() + 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( diff --git a/packages/plugin-browser/test/rpc.test.ts b/packages/plugin-browser/test/rpc.test.ts index af1d5b53f05b..d0e0c017217f 100644 --- a/packages/plugin-browser/test/rpc.test.ts +++ b/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([