fix(opencode): publish permission cleanup replies

Author: 李冠辰

packages/opencode/src/permission/index.ts

@@ -157,17 +157,34 @@ export const layer = Layer.effect(
 
         yield* Effect.addFinalizer(() =>
           Effect.gen(function* () {
-            for (const item of state.pending.values()) {
+            for (const id of Array.from(state.pending.keys())) {
+              const item = yield* removePending(state.pending, id, "reject")
+              if (!item) continue
               yield* Deferred.fail(item.deferred, new RejectedError())
             }
-            state.pending.clear()
           }),
         )
 
         return state
       }),
     )
 
+    const removePending = Effect.fn("Permission.removePending")(function* (
+      pending: Map<PermissionID, PendingEntry>,
+      id: PermissionID,
+      reply: Reply,
+    ) {
+      const item = pending.get(id)
+      if (!item) return undefined
+      pending.delete(id)
+      yield* bus.publish(Event.Replied, {
+        sessionID: item.info.sessionID,
+        requestID: item.info.id,
+        reply,
+      })
+      return item
+    })
+
     const ask = Effect.fn("Permission.ask")(function* (input: AskInput) {
       const { approved, pending } = yield* InstanceState.get(state)
       const { ruleset, ...request } = input
@@ -202,26 +219,14 @@ export const layer = Layer.effect(
       const deferred = yield* Deferred.make<void, RejectedError | CorrectedError>()
       pending.set(id, { info, deferred })
       yield* bus.publish(Event.Asked, info)
-      return yield* Effect.ensuring(
-        Deferred.await(deferred),
-        Effect.sync(() => {
-          pending.delete(id)
-        }),
-      )
+      return yield* Effect.ensuring(Deferred.await(deferred), removePending(pending, id, "reject").pipe(Effect.ignore))
     })
 
     const reply = Effect.fn("Permission.reply")(function* (input: ReplyInput) {
       const { approved, pending } = yield* InstanceState.get(state)
-      const existing = pending.get(input.requestID)
+      const existing = yield* removePending(pending, input.requestID, input.reply)
       if (!existing) return yield* new NotFoundError({ requestID: input.requestID })
 
-      pending.delete(input.requestID)
-      yield* bus.publish(Event.Replied, {
-        sessionID: existing.info.sessionID,
-        requestID: existing.info.id,
-        reply: input.reply,
-      })
-
       if (input.reply === "reject") {
         yield* Deferred.fail(
           existing.deferred,
@@ -230,13 +235,8 @@ export const layer = Layer.effect(
 
         for (const [id, item] of pending.entries()) {
           if (item.info.sessionID !== existing.info.sessionID) continue
-          pending.delete(id)
-          yield* bus.publish(Event.Replied, {
-            sessionID: item.info.sessionID,
-            requestID: item.info.id,
-            reply: "reject",
-          })
-          yield* Deferred.fail(item.deferred, new RejectedError())
+          const removed = yield* removePending(pending, id, "reject")
+          if (removed) yield* Deferred.fail(removed.deferred, new RejectedError())
         }
         return
       }
@@ -258,13 +258,8 @@ export const layer = Layer.effect(
           (pattern) => evaluate(item.info.permission, pattern, approved).action === "allow",
         )
         if (!ok) continue
-        pending.delete(id)
-        yield* bus.publish(Event.Replied, {
-          sessionID: item.info.sessionID,
-          requestID: item.info.id,
-          reply: "always",
-        })
-        yield* Deferred.succeed(item.deferred, undefined)
+        const removed = yield* removePending(pending, id, "always")
+        if (removed) yield* Deferred.succeed(removed.deferred, undefined)
       }
     })
 

packages/opencode/test/permission/next.test.ts

@@ -9,6 +9,7 @@ import { InstanceBootstrap } from "../../src/project/bootstrap-service"
 import { InstanceStore } from "../../src/project/instance-store"
 import { TestInstance, tmpdirScoped } from "../fixture/fixture"
 import { testEffect } from "../lib/effect"
+import { waitGlobalBusEvent } from "../server/global-bus"
 import { MessageID, SessionID } from "../../src/session/schema"
 
 const bus = Bus.layer
@@ -50,6 +51,22 @@ const waitForPending = (count: number) =>
     )
   })
 
+const waitForReplied = (requestID: PermissionID) =>
+  Effect.gen(function* () {
+    const bus = yield* Bus.Service
+    const seen = yield* Deferred.make<{ sessionID: SessionID; requestID: PermissionID; reply: Permission.Reply }>()
+    const unsub = yield* bus.subscribeCallback(Permission.Event.Replied, (event) => {
+      if (event.properties.requestID === requestID) Deferred.doneUnsafe(seen, Effect.succeed(event.properties))
+    })
+    yield* Effect.addFinalizer(() => Effect.sync(unsub))
+    return yield* Deferred.await(seen).pipe(
+      Effect.timeoutOrElse({
+        duration: "1 second",
+        orElse: () => Effect.fail(new Error(`timed out waiting for permission replied event: ${requestID}`)),
+      }),
+    )
+  })
+
 const fail = <A, E, R>(self: Effect.Effect<A, E, R>) =>
   Effect.gen(function* () {
     const exit = yield* self.pipe(Effect.exit)
@@ -1003,6 +1020,31 @@ it.live("permission requests stay isolated by directory", () =>
   }),
 )
 
+it.instance(
+  "ask - publishes replied event when pending request is interrupted",
+  () =>
+    Effect.gen(function* () {
+      const requestID = PermissionID.make("per_interrupt_event")
+      const sessionID = SessionID.make("session_interrupt_event")
+      const replied = yield* waitForReplied(requestID).pipe(Effect.forkScoped)
+      const fiber = yield* ask({
+        id: requestID,
+        sessionID,
+        permission: "bash",
+        patterns: ["ls"],
+        metadata: {},
+        always: [],
+        ruleset: [],
+      }).pipe(Effect.forkScoped)
+
+      expect(yield* waitForPending(1)).toHaveLength(1)
+      yield* Fiber.interrupt(fiber)
+      expect(yield* Fiber.join(replied)).toEqual({ sessionID, requestID, reply: "reject" })
+      expect(yield* list()).toHaveLength(0)
+    }),
+  { git: true },
+)
+
 it.instance(
   "pending permission rejects on instance dispose",
   () =>
@@ -1030,6 +1072,42 @@ it.instance(
   { git: true },
 )
 
+it.instance(
+  "pending permission publishes global replied event on instance dispose",
+  () =>
+    Effect.gen(function* () {
+      const requestID = PermissionID.make("per_dispose_event")
+      const sessionID = SessionID.make("session_dispose_event")
+      const test = yield* TestInstance
+      const store = yield* InstanceStore.Service
+      const replied = yield* waitGlobalBusEvent({
+        message: "timed out waiting for global permission replied event",
+        predicate: (event) =>
+          event.directory === test.directory &&
+          event.payload.type === "permission.replied" &&
+          event.payload.properties.requestID === requestID,
+      }).pipe(Effect.forkScoped)
+      const fiber = yield* ask({
+        id: requestID,
+        sessionID,
+        permission: "bash",
+        patterns: ["ls"],
+        metadata: {},
+        always: [],
+        ruleset: [],
+      }).pipe(Effect.forkScoped)
+
+      expect(yield* waitForPending(1)).toHaveLength(1)
+      const ctx = yield* store.load({ directory: test.directory })
+      yield* store.dispose(ctx)
+
+      const exit = yield* Fiber.await(fiber)
+      expect(Exit.isFailure(exit)).toBe(true)
+      expect((yield* Fiber.join(replied)).payload.properties).toEqual({ sessionID, requestID, reply: "reject" })
+    }),
+  { git: true },
+)
+
 it.instance(
   "pending permission rejects on instance reload",
   () =>