fix: harden OpenCode event correlation

cursoragent · shukieshah · cursoragent · commit 18c0b0898b86 · 2026-07-18T23:10:15.000Z
Co-authored-by: Shukan Shah &lt;shukanshah14@gmail.com&gt;
diff --git a/cli/beacon-hooks/cmd/opencode_event.go b/cli/beacon-hooks/cmd/opencode_event.go
@@ -172,8 +172,10 @@ func opencodeEndpointEvents(input map[string]interface{}, sessionID string) []op
 		if tool := opencodeToolName(input); tool != "" {
 			fields["tool"] = map[string]interface{}{"name": tool}
 		}
+		mergeMap(fields, opencodePermissionCorrelation(properties))
 		return one("approval.requested", "approval", "info", "opencode permission requested", fields)
 	case "permission.replied", "permission.updated", "permission.v2.replied":
+		properties := opencodeProperties(input)
 		decision := opencodeDecision(input)
 		if decision == "" {
 			decision = "unknown"
@@ -182,6 +184,7 @@ func opencodeEndpointEvents(input map[string]interface{}, sessionID string) []op
 		if tool := opencodeToolName(input); tool != "" {
 			fields["tool"] = map[string]interface{}{"name": tool}
 		}
+		mergeMap(fields, opencodePermissionCorrelation(properties))
 		action := opencodeApprovalAction(decision)
 		return one(action, "approval", "info", "opencode permission "+decision, fields)
 	default:
@@ -196,6 +199,7 @@ func supportedOpenCodeEventTypes() []string {
 		"command.executed",
 		"file.edited",
 		"file.watcher.updated",
+		"message.part.delta",
 		"message.part.updated",
 		"message.updated",
 		"permission.asked",
@@ -613,6 +617,28 @@ func opencodeApprovalAction(decision string) string {
 	}
 }
 
+func opencodePermissionCorrelation(properties map[string]interface{}) map[string]interface{} {
+	toolRef := opencodeMap(properties, "tool")
+	callID := getFirstStr(toolRef, "callID", "call_id")
+	if callID == "" {
+		return nil
+	}
+	toolName := getFirstStr(properties, "permission")
+	fields := map[string]interface{}{
+		"gen_ai": map[string]interface{}{
+			"operation": map[string]interface{}{"name": "execute_tool"},
+			"tool": map[string]interface{}{
+				"name": toolName,
+				"call": map[string]interface{}{"id": callID},
+			},
+		},
+	}
+	if toolName != "" {
+		fields["tool"] = map[string]interface{}{"name": toolName}
+	}
+	return fields
+}
+
 func opencodeProperties(input map[string]interface{}) map[string]interface{} {
 	if properties := opencodeMap(input, "properties", "data"); len(properties) > 0 {
 		return properties
diff --git a/cli/beacon-hooks/cmd/opencode_event_test.go b/cli/beacon-hooks/cmd/opencode_event_test.go
@@ -332,7 +332,10 @@ func TestOpenCodePermissionRepliesMapAllowAndDeny(t *testing.T) {
 		action, _, _, _, _ := opencodeEndpointEvent(map[string]interface{}{
 			"type":       "permission.replied",
 			"session_id": "ses_test",
-			"properties": map[string]interface{}{"reply": tt.reply},
+			"properties": map[string]interface{}{
+				"reply": tt.reply, "permission": "bash",
+				"tool": map[string]interface{}{"messageID": "msg_1", "callID": "call_permission"},
+			},
 		}, "ses_test")
 		if action != tt.action {
 			t.Fatalf("reply %q action = %q, want %q", tt.reply, action, tt.action)
diff --git a/cli/beacon/internal/endpoint/hooks/assets/opencode/beacon.ts b/cli/beacon/internal/endpoint/hooks/assets/opencode/beacon.ts
@@ -47,6 +47,11 @@ async function debugLog(client, message, extra) {
 }
 
 async function sendToBeacon(client, payload) {
+  const testSender = globalThis[Symbol.for("beacon.opencode.testSender")]
+  if (typeof testSender === "function") {
+    await testSender(payload)
+    return
+  }
   let proc
   try {
     proc = Bun.spawn(["/bin/sh", "-lc", beaconCommand], {
@@ -87,6 +92,7 @@ function sessionID(value) {
 }
 
 function modelName(value) {
+  if (value?.providerID && value?.modelID) return value.providerID + "/" + value.modelID
   const model = value?.model || value?.modelInfo
   if (!model || typeof model === "string") return model || ""
   if (model.providerID && model.modelID) return model.providerID + "/" + model.modelID
@@ -118,6 +124,9 @@ export const BeaconEndpointPlugin = async ({ project, directory, worktree, clien
   const pendingParts = new Map()
   const partDeltas = new Map()
   const messageModels = new Map()
+  const completedMessages = new Set()
+  const permissionRequests = new Map()
+  const sessionStates = new Map()
   const recentFilePaths = new Map()
   let eventQueue = Promise.resolve()
 
@@ -128,10 +137,10 @@ export const BeaconEndpointPlugin = async ({ project, directory, worktree, clien
       .catch((err) => debugLog(client, "Beacon event queue failed", { error: String(err), type: value?.type }))
     return eventQueue
   }
-  const rememberFilePath = (tool, args) => {
+  const rememberFilePath = (tool, args, sid, callID) => {
     if (!isFileMutationTool(tool)) return
     const path = filePath(args)
-    if (path) recentFilePaths.set(path, Date.now())
+    if (path) recentFilePaths.set(path, { time: Date.now(), sessionID: sid, callID })
   }
   const partKey = (part) => `${part?.sessionID || ""}:${part?.messageID || ""}:${part?.id || ""}`
   const emitPart = (part, sid, model) => {
@@ -208,7 +217,7 @@ export const BeaconEndpointPlugin = async ({ project, directory, worktree, clien
     "tool.execute.after": async (input, output) => {
       const active = activeCalls.get(input?.callID)
       const args = input?.args || active?.args || {}
-      rememberFilePath(input?.tool, args)
+      rememberFilePath(input?.tool, args, sessionID(input), input?.callID)
       await sendToBeacon(
         client,
         payload("tool.execute.after", {
@@ -235,6 +244,10 @@ export const BeaconEndpointPlugin = async ({ project, directory, worktree, clien
         const model = modelName(info)
         if (info?.id && model) messageModels.set(info.id, model)
         if (!info?.time?.completed && !info?.error) return
+        if (info?.id) {
+          if (completedMessages.has(info.id)) return
+          completedMessages.add(info.id)
+        }
       }
       if (type === "message.part.delta") {
         const key = `${sid}:${properties?.messageID || ""}:${properties?.partID || ""}`
@@ -246,27 +259,51 @@ export const BeaconEndpointPlugin = async ({ project, directory, worktree, clien
         emitPart(part, sid, messageModels.get(part?.messageID) || "")
         return
       }
-      if (type === "session.status" && properties?.status?.type === "idle") flushParts(sid)
-      if (type === "session.idle") flushParts(sid)
+      if (type === "session.status") {
+        const status = properties?.status?.type || ""
+        sessionStates.set(sid, status)
+        if (status === "idle") flushParts(sid)
+      }
+      if (type === "session.idle") {
+        flushParts(sid)
+        if (sessionStates.get(sid) === "idle") return
+        sessionStates.set(sid, "idle")
+      }
+      if (type === "permission.asked" || type === "permission.v2.asked") {
+        if (properties?.id) {
+          permissionRequests.set(properties.id, { tool: properties?.tool, permission: properties?.permission })
+        }
+      }
+      if (type === "permission.replied" || type === "permission.v2.replied") {
+        const request = permissionRequests.get(properties?.requestID)
+        if (request?.tool && !properties.tool) properties.tool = request.tool
+        if (request?.permission && !properties.permission) properties.permission = request.permission
+        permissionRequests.delete(properties?.requestID)
+      }
       if (type === "session.diff") {
         const diffs = Array.isArray(properties?.diff) ? properties.diff : []
         const now = Date.now()
         const filtered = diffs.filter((item) => {
           const path = filePath(item)
           if (!path) return false
           const seen = recentFilePaths.get(path)
-          if (seen && now - seen < 5000) return false
+          if (seen && now - seen.time < 5000) return false
           return item?.before !== item?.after || item?.additions || item?.deletions
         })
         if (filtered.length === 0) return
         properties.diff = filtered
       }
       if (type === "file.edited" || type === "file.watcher.updated") {
-        if (!sid) return
+        if (!sid) {
+          const seen = recentFilePaths.get(filePath(properties))
+          if (!seen || Date.now() - seen.time >= 5000) return
+          properties.sessionID = seen.sessionID
+          properties.callID = seen.callID
+        }
       }
       void enqueue(
         payload(type, {
-          session_id: sid,
+          session_id: sessionID(properties) || sid,
           model: modelName(info || properties),
           properties,
           event,
diff --git a/cli/beacon/internal/endpoint/hooks/opencode_test.go b/cli/beacon/internal/endpoint/hooks/opencode_test.go
@@ -132,6 +132,9 @@ func TestOpenCodeInstalledUsesManagedMarker(t *testing.T) {
 	if isOpenCodeInstalledAt(path) {
 		t.Fatal("unmarked plugin should not be detected as installed")
 	}
+	if err := os.Remove(path); err != nil {
+		t.Fatalf("remove unmarked plugin: %v", err)
+	}
 	if err := installOpenCodePlugin(path, "/tmp/beacon-hooks", "/tmp/runtime.jsonl", "/tmp/config.json"); err != nil {
 		t.Fatalf("installOpenCodePlugin returned error: %v", err)
 	}
diff --git a/plugins/opencode-beacon/src/beacon.test.ts b/plugins/opencode-beacon/src/beacon.test.ts
@@ -1,28 +1,16 @@
 import { afterEach, beforeEach, describe, expect, test } from "bun:test"
 import { BeaconEndpointPlugin } from "./beacon"
 
-const originalSpawn = Bun.spawn
 const payloads: any[] = []
+const senderKey = Symbol.for("beacon.opencode.testSender")
 
 beforeEach(() => {
   payloads.length = 0
-  Object.defineProperty(Bun, "spawn", {
-    configurable: true,
-    value: () => ({
-      stdin: {
-        write(value: string) {
-          payloads.push(JSON.parse(value))
-        },
-        end() {},
-      },
-      exited: Promise.resolve(0),
-      kill() {},
-    }),
-  })
+  ;(globalThis as any)[senderKey] = async (payload: any) => payloads.push(structuredClone(payload))
 })
 
 afterEach(() => {
-  Object.defineProperty(Bun, "spawn", { configurable: true, value: originalSpawn })
+  delete (globalThis as any)[senderKey]
 })
 
 async function plugin() {
diff --git a/plugins/opencode-beacon/src/beacon.ts b/plugins/opencode-beacon/src/beacon.ts
@@ -47,6 +47,11 @@ async function debugLog(client, message, extra) {
 }
 
 async function sendToBeacon(client, payload) {
+  const testSender = globalThis[Symbol.for("beacon.opencode.testSender")]
+  if (typeof testSender === "function") {
+    await testSender(payload)
+    return
+  }
   let proc
   try {
     proc = Bun.spawn(["/bin/sh", "-lc", beaconCommand], {
@@ -87,6 +92,7 @@ function sessionID(value) {
 }
 
 function modelName(value) {
+  if (value?.providerID && value?.modelID) return value.providerID + "/" + value.modelID
   const model = value?.model || value?.modelInfo
   if (!model || typeof model === "string") return model || ""
   if (model.providerID && model.modelID) return model.providerID + "/" + model.modelID
@@ -118,6 +124,9 @@ export const BeaconEndpointPlugin = async ({ project, directory, worktree, clien
   const pendingParts = new Map()
   const partDeltas = new Map()
   const messageModels = new Map()
+  const completedMessages = new Set()
+  const permissionRequests = new Map()
+  const sessionStates = new Map()
   const recentFilePaths = new Map()
   let eventQueue = Promise.resolve()
 
@@ -128,10 +137,10 @@ export const BeaconEndpointPlugin = async ({ project, directory, worktree, clien
       .catch((err) => debugLog(client, "Beacon event queue failed", { error: String(err), type: value?.type }))
     return eventQueue
   }
-  const rememberFilePath = (tool, args) => {
+  const rememberFilePath = (tool, args, sid, callID) => {
     if (!isFileMutationTool(tool)) return
     const path = filePath(args)
-    if (path) recentFilePaths.set(path, Date.now())
+    if (path) recentFilePaths.set(path, { time: Date.now(), sessionID: sid, callID })
   }
   const partKey = (part) => `${part?.sessionID || ""}:${part?.messageID || ""}:${part?.id || ""}`
   const emitPart = (part, sid, model) => {
@@ -208,7 +217,7 @@ export const BeaconEndpointPlugin = async ({ project, directory, worktree, clien
     "tool.execute.after": async (input, output) => {
       const active = activeCalls.get(input?.callID)
       const args = input?.args || active?.args || {}
-      rememberFilePath(input?.tool, args)
+      rememberFilePath(input?.tool, args, sessionID(input), input?.callID)
       await sendToBeacon(
         client,
         payload("tool.execute.after", {
@@ -235,6 +244,10 @@ export const BeaconEndpointPlugin = async ({ project, directory, worktree, clien
         const model = modelName(info)
         if (info?.id && model) messageModels.set(info.id, model)
         if (!info?.time?.completed && !info?.error) return
+        if (info?.id) {
+          if (completedMessages.has(info.id)) return
+          completedMessages.add(info.id)
+        }
       }
       if (type === "message.part.delta") {
         const key = `${sid}:${properties?.messageID || ""}:${properties?.partID || ""}`
@@ -246,27 +259,51 @@ export const BeaconEndpointPlugin = async ({ project, directory, worktree, clien
         emitPart(part, sid, messageModels.get(part?.messageID) || "")
         return
       }
-      if (type === "session.status" && properties?.status?.type === "idle") flushParts(sid)
-      if (type === "session.idle") flushParts(sid)
+      if (type === "session.status") {
+        const status = properties?.status?.type || ""
+        sessionStates.set(sid, status)
+        if (status === "idle") flushParts(sid)
+      }
+      if (type === "session.idle") {
+        flushParts(sid)
+        if (sessionStates.get(sid) === "idle") return
+        sessionStates.set(sid, "idle")
+      }
+      if (type === "permission.asked" || type === "permission.v2.asked") {
+        if (properties?.id) {
+          permissionRequests.set(properties.id, { tool: properties?.tool, permission: properties?.permission })
+        }
+      }
+      if (type === "permission.replied" || type === "permission.v2.replied") {
+        const request = permissionRequests.get(properties?.requestID)
+        if (request?.tool && !properties.tool) properties.tool = request.tool
+        if (request?.permission && !properties.permission) properties.permission = request.permission
+        permissionRequests.delete(properties?.requestID)
+      }
       if (type === "session.diff") {
         const diffs = Array.isArray(properties?.diff) ? properties.diff : []
         const now = Date.now()
         const filtered = diffs.filter((item) => {
           const path = filePath(item)
           if (!path) return false
           const seen = recentFilePaths.get(path)
-          if (seen && now - seen < 5000) return false
+          if (seen && now - seen.time < 5000) return false
           return item?.before !== item?.after || item?.additions || item?.deletions
         })
         if (filtered.length === 0) return
         properties.diff = filtered
       }
       if (type === "file.edited" || type === "file.watcher.updated") {
-        if (!sid) return
+        if (!sid) {
+          const seen = recentFilePaths.get(filePath(properties))
+          if (!seen || Date.now() - seen.time >= 5000) return
+          properties.sessionID = seen.sessionID
+          properties.callID = seen.callID
+        }
       }
       void enqueue(
         payload(type, {
-          session_id: sid,
+          session_id: sessionID(properties) || sid,
           model: modelName(info || properties),
           properties,
           event,

Original file line number	Diff line number	Diff line change
`@@ -132,6 +132,9 @@ func TestOpenCodeInstalledUsesManagedMarker(t *testing.T) {`
`132`	`132`	`if isOpenCodeInstalledAt(path) {`
`133`	`133`	`t.Fatal("unmarked plugin should not be detected as installed")`
`134`	`134`	`}`
	`135`	`+ if err := os.Remove(path); err != nil {`
	`136`	`+ t.Fatalf("remove unmarked plugin: %v", err)`
	`137`	`+ }`
`135`	`138`	`if err := installOpenCodePlugin(path, "/tmp/beacon-hooks", "/tmp/runtime.jsonl", "/tmp/config.json"); err != nil {`
`136`	`139`	`t.Fatalf("installOpenCodePlugin returned error: %v", err)`
`137`	`140`	`}`