Skip to content

fix(session): route nested blockers to ancestor clients #30639

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
literally-dan wants to merge 1 commit into
base: dev
Choose a base branch
from
Open

fix(session): route nested blockers to ancestor clients #30639

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions packages/opencode/src/acp/event.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,7 @@ export class Subscription {

stop() {
this.abort.abort()
this.permission.stop()
}

async handle(event: Event) {
Expand All @@ -74,6 +75,9 @@ export class Subscription {
return this.handlePartUpdated(event)
case "message.part.delta":
return this.handlePartDelta(event)
case "session.deleted":
this.permission.forgetSession(event.properties.sessionID)
return
}
}

Expand Down
234 changes: 203 additions & 31 deletions packages/opencode/src/acp/permission.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ import { applyPatch } from "diff"
import { exists, readText } from "@/util/filesystem"
import type { ACPSession } from "./session"
import { pendingToolCall, toLocations, type ToolInput } from "./tool"
import { SessionAncestry } from "@/session/ancestry"
import { Effect } from "effect"

type PermissionEvent = Extract<Event, { type: "permission.asked" }>
Expand All @@ -23,8 +24,17 @@ const permissionOptions: PermissionOption[] = [
{ optionId: "reject", kind: "reject_once", name: "Reject" },
]

type Ownership = SessionAncestry.Ownership<ACPSession.Info>

export class Handler {
private readonly queues = new Map<string, Promise<void>>()
private readonly pending = new Map<string, { sessionID: string; controller: AbortController }>()
private stopped = false
// Memoise sessionID -> parentID lookups so events streaming from a
// deeply nested subagent don't re-fetch the whole chain per event.
// `parentID` is immutable per session row, so cache entries never go
// stale; we drop them on `session.deleted` to bound memory.
private readonly parentIDCache = new Map<string, string | undefined>()

constructor(
private readonly input: {
Expand All @@ -34,46 +44,170 @@ export class Handler {
},
) {}

// Invoked from the event subscription on `session.deleted` so the
// parentID cache tracks live sessions only.
forgetSession(sessionID: string) {
this.parentIDCache.delete(sessionID)
for (const pending of this.pending.values()) {
if (pending.sessionID === sessionID) pending.controller.abort()
}
}

stop() {
this.stopped = true
for (const pending of this.pending.values()) pending.controller.abort()
this.pending.clear()
this.queues.clear()
}

handle(event: PermissionEvent) {
const permission = event.properties
const previous = this.queues.get(permission.sessionID) ?? Promise.resolve()
if (this.stopped || this.pending.has(permission.id)) return
const controller = new AbortController()
this.pending.set(permission.id, { sessionID: permission.sessionID, controller })
void this.route(event, controller.signal)
.catch((error) =>
Effect.runSync(
Effect.logError("failed to route permission request", {
requestID: permission.id,
sessionID: permission.sessionID,
error,
}),
),
)
.finally(() => {
if (this.pending.get(permission.id)?.controller === controller) this.pending.delete(permission.id)
})
}

private async route(event: PermissionEvent, signal: AbortSignal) {
const permission = event.properties
const ownership = await this.resolveOwnership(permission, signal)
if (ownership.type === "foreign" || ownership.type === "cancelled") return
if (ownership.type === "exhausted") {
// A cycle or a deleted parent never resolves. Without a reply the nested
// tool call waits on an untimed deferred forever, so answer it.
Effect.runSync(
Effect.logError("permission ancestry unresolvable; rejecting", {
requestID: permission.id,
sessionID: permission.sessionID,
error: ownership.error,
}),
)
await this.reply(permission.id, "reject").catch(() => {})
return
}
const session = ownership.value

// The ACP client displays nested prompts against the managed ancestor.
// Queue by that same ID so sibling leaves cannot open competing prompts
// for one displayed session.
const previous = this.queues.get(session.id) ?? Promise.resolve()
const next = previous
.then(() => this.process(event))
.catch(() => {})
.then(() => (signal.aborted ? undefined : this.process(event, session)))
.catch((error) => this.rejectOwnedPermission(event, error, session.cwd))
.finally(() => {
if (this.queues.get(permission.sessionID) === next) {
this.queues.delete(permission.sessionID)
if (this.queues.get(session.id) === next) {
this.queues.delete(session.id)
}
})
this.queues.set(permission.sessionID, next)
this.queues.set(session.id, next)
await next
}

private async process(event: PermissionEvent) {
const permission = event.properties
const session = await Effect.runPromise(this.input.session.tryGet(permission.sessionID))
if (!session) return
private resolveOwnership(permission: PermissionEvent["properties"], signal: AbortSignal) {
return SessionAncestry.resolve({
signal,
lookup: (current) => this.lookupOwnership(permission.sessionID, current),
onRetry: (retry) => {
if (retry.attempt !== 1 && retry.attempt % 5 !== 0) return
Effect.runSync(
Effect.logWarning("permission ownership unresolved; retrying ancestry lookup", {
requestID: permission.id,
sessionID: permission.sessionID,
attempt: retry.attempt,
retryIn: retry.retryIn,
error: retry.error,
}),
)
},
})
}

private async lookupOwnership(sessionID: string, signal: AbortSignal): Promise<Ownership> {
try {
const session = await this.resolveManagedAncestor(sessionID, signal)
return session ? { type: "owned", value: session } : { type: "foreign" }
} catch (error) {
return { type: "unknown", error }
}
}

// Climb the parent chain from `sessionID` and return the first ACP
// session the client created (root or load). Used to route permission
// prompts for nested subagent sessions — the ACP client never sees the
// leaf id, so we report against the managed ancestor instead. Returns
// undefined if the chain reaches the root without hitting a managed
// session, in which case the prompt is foreign (e.g. another ACP
// agent's session).
private async resolveManagedAncestor(sessionID: string, signal: AbortSignal) {
const direct = await Effect.runPromise(this.input.session.tryGet(sessionID))
if (direct) return direct
let current: string | undefined = sessionID
const seen = new Set<string>()
while (current && !seen.has(current)) {
seen.add(current)
const parentID = await this.lookupParentID(current, signal)
if (!parentID) return undefined
const parent = await Effect.runPromise(this.input.session.tryGet(parentID))
if (parent) return parent
current = parentID
}
throw new Error(`parent chain cycle detected for ${sessionID}: ${Array.from(seen).join(" -> ")}`)
}

private async lookupParentID(sessionID: string, signal: AbortSignal): Promise<string | undefined> {
if (this.parentIDCache.has(sessionID)) return this.parentIDCache.get(sessionID)
// Only cache successful lookups. Caching `undefined` on a transient
// SDK failure (network blip, server restart mid-flight) would poison
// the chain — a retained permission would re-use the false negative,
// fail to find an ancestor, and the tool call would hang forever. That
// is exactly the symptom this fix is meant to prevent, so each retry
// pays the SDK round-trip again. `throwOnError`
// is required so an HTTP-level failure (5xx during a server restart)
// rejects into the catch instead of resolving to an empty-data envelope
// whose `undefined` parentID would otherwise be cached as a false
// negative.
const res = await this.input.sdk.session.get({ sessionID }, { throwOnError: true, signal })
if (!res.data) throw new Error(`session lookup returned no data for ${sessionID}`)
const parentID = res.data.parentID
this.parentIDCache.set(sessionID, parentID)
return parentID
}

private async process(event: PermissionEvent, session: ACPSession.Info) {
const permission = event.properties
// Permissions raised inside a nested subagent carry the deepest
// session's ID, which the ACP client never explicitly created. Walk
// up to the managed ancestor so the prompt is routed somewhere the
// client knows about; otherwise the deferred on the server never
// resolves and the nested tool call (including MCP calls) hangs.
if (!this.input.connection.requestPermission) {
await this.reply(permission.id, "reject", session.cwd)
return
}

const result = await this.input.connection
.requestPermission({
sessionId: permission.sessionID,
toolCall: await permissionToolCall({
toolCallId: permission.tool?.callID ?? permission.id,
toolName: permission.permission,
input: permission.metadata,
}),
options: permissionOptions,
})
.catch(async () => {
await this.reply(permission.id, "reject", session.cwd)
return undefined
})

if (!result) return
const result = await this.input.connection.requestPermission({
// Route to the managed ancestor's id (resolved above), not the
// leaf subagent id the client never created.
sessionId: session.id,
toolCall: await permissionToolCall({
toolCallId: permission.tool?.callID ?? permission.id,
toolName: permission.permission,
input: permission.metadata,
}),
options: permissionOptions,
})

const reply = selectedReply(result)
if (reply !== "once" && reply !== "always") {
Expand All @@ -88,14 +222,48 @@ export class Handler {
await this.reply(permission.id, reply, session.cwd)
}

private async reply(requestID: string, reply: Reply, directory: string) {
await this.input.sdk.permission.reply({
requestID,
reply,
directory,
private async rejectOwnedPermission(event: PermissionEvent, error: unknown, directory: string) {
const permission = event.properties
Effect.runSync(
Effect.logError("failed to process owned permission request", {
requestID: permission.id,
sessionID: permission.sessionID,
error,
}),
)
await this.reply(permission.id, "reject", directory).catch((replyError) => {
Effect.runSync(
Effect.logError("failed to reject owned permission request", {
requestID: permission.id,
sessionID: permission.sessionID,
error: replyError,
}),
)
})
}

private async reply(requestID: string, reply: Reply, directory?: string) {
try {
const result: unknown = await this.input.sdk.permission.reply(
{
requestID,
reply,
...(directory ? { directory } : {}),
},
{ throwOnError: true },
)
if (typeof result === "object" && result !== null && "error" in result && result.error !== undefined) {
throw result.error
}
} catch (error) {
// The server drops a pending request when its session is aborted and
// rejects the cascade itself, so a reply that lost that race is already
// answered. Nothing left to do.
if (isPermissionNotFound(error)) return
throw error
}
}

private async writeProposedEdit(sessionId: string, metadata: ToolInput) {
const filepath = stringValue(metadata.filepath)
const diff = stringValue(metadata.diff)
Expand All @@ -115,6 +283,10 @@ export class Handler {
}
}

function isPermissionNotFound(error: unknown) {
return typeof error === "object" && error !== null && "_tag" in error && error._tag === "PermissionNotFoundError"
}

async function permissionToolCall(input: {
readonly toolCallId: string
readonly toolName: string
Expand Down
Loading
Loading