Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
5 changes: 5 additions & 0 deletions .changeset/agent-manager-answer-action.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
"kilo-code": minor
---

Add an agent_manager answer action so orchestrating agents can resolve a managed session's pending question instead of only stopping it. Prompting a session that waits on input now fails immediately with the pending question named.
49 changes: 43 additions & 6 deletions packages/kilo-vscode/src/agent-manager/orchestration-bridge.ts
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ import type { PRStatus } from "./types"
import type { WorktreeStateManager } from "./WorktreeStateManager"
import {
OrchestrationError,
answer,
move,
overview,
prompt,
Expand All @@ -28,12 +29,14 @@ type Request =
| (RequestBase & { operation: "prompt"; targetSessionID: string; prompt: string })
| (RequestBase & { operation: "stop"; targetSessionID: string })
| (RequestBase & { operation: "move"; targetSessionID: string; sectionID: string | null })
| (RequestBase & { operation: "answer"; targetSessionID: string; questionID?: string; answers: string[][] })

type Result =
| { operation: "overview"; overview: Overview }
| { operation: "prompt"; sessionID: string; delivered: true }
| { operation: "stop"; sessionID: string; stopped: true }
| { operation: "move"; sessionID: string; sectionID: string | null; moved: true }
| { operation: "answer"; sessionID: string; questionID: string; resolved: true }

interface Failure {
code: FailureCode | "cancelled" | "disconnected" | "timeout"
Expand Down Expand Up @@ -288,6 +291,9 @@ export class AgentManagerOrchestrationBridge {
if (this.disposed || active.cancelled) return
return { result: { operation: "prompt", sessionID: request.targetSessionID, delivered: true } }
}
if (request.operation === "answer") {
return await this.resolveQuestion(client, root, state, request, origin, active)
}
if (request.operation === "move") {
move({ state, sessionID: request.targetSessionID, sectionID: request.sectionID })
this.options.push(origin.directory)
Expand All @@ -301,18 +307,49 @@ export class AgentManagerOrchestrationBridge {
},
}
}
if (!this.options.managed(request.targetSessionID, origin.directory)) {
throw new OrchestrationError("unknown_session", "The session is not managed by this Agent Manager workspace")
}
await this.options.close(request.targetSessionID, origin.directory)
if (this.disposed || active.cancelled) return
return { result: { operation: "stop", sessionID: request.targetSessionID, stopped: true } }
return await this.deactivate(request.targetSessionID, origin.directory, active)
} catch (error) {
if (this.disposed || active.cancelled) return
return { error: failure(error) }
}
}

private async deactivate(sessionID: string, originDirectory: string, active: Active): Promise<Outcome | undefined> {
if (!this.options.managed(sessionID, originDirectory)) {
throw new OrchestrationError("unknown_session", "The session is not managed by this Agent Manager workspace")
}
await this.options.close(sessionID, originDirectory)
if (this.disposed || active.cancelled) return
return { result: { operation: "stop", sessionID, stopped: true } }
}

private async resolveQuestion(
client: KiloClient,
root: string,
state: WorktreeStateManager,
request: Extract<Request, { operation: "answer" }>,
origin: Origin,
active: Active,
): Promise<Outcome | undefined> {
const resolved = await answer({
client,
root,
state,
sessionID: request.targetSessionID,
questionID: request.questionID,
answers: request.answers,
})
if (this.disposed || active.cancelled) return
return {
result: {
operation: "answer",
sessionID: request.targetSessionID,
questionID: resolved.questionID,
resolved: true,
},
}
}

private async reply(requestID: string, directory: string, result: Result): Promise<boolean> {
try {
const response = await this.connection.getClient().kilocode.agentManager.reply({ requestID, directory, result })
Expand Down
125 changes: 116 additions & 9 deletions packages/kilo-vscode/src/agent-manager/orchestration-domain.ts
Original file line number Diff line number Diff line change
Expand Up @@ -311,17 +311,21 @@ export async function overview(input: OverviewInput): Promise<Overview> {
return grouped(input, sessions, summaries, worktreeSummaries(input, summaries, filters))
}

export async function prompt(input: {
interface Target {
client: KiloClient
root: string
state: WorktreeStateManager
sessionID: string
text: string
messageID: string
signal?: AbortSignal
idleTimeoutMs?: number
}): Promise<void> {
if (input.signal?.aborted) return
}

interface Located {
dir: string
name: string
}

// Verify the target is a live managed session of this workspace and return its authoritative
// directory plus display name, so error messages can echo exact IDs back to the caller.
async function locate(input: Target): Promise<Located> {
const managed = input.state.getSession(input.sessionID)
if (!managed)
throw new OrchestrationError("unknown_session", "The session is not managed by this Agent Manager workspace")
Expand All @@ -343,12 +347,68 @@ export async function prompt(input: {
if (!(await sameManagedDirectory(response.data.directory, dir))) {
throw new OrchestrationError("cross_workspace", "The managed session belongs to a different workspace directory")
}
await waitForIdle(input.client, dir, input.sessionID, input.signal, input.idleTimeoutMs ?? 30_000)
return { dir, name: response.data.title.trim() || input.sessionID }
}

function truncate(value: string, max: number): string {
return value.length <= max ? value : `${value.slice(0, max - 1)}…`
}

// Name what keeps the target from accepting a prompt. A session blocked on a question
// never becomes idle on its own, so naming the blocker here lets an orchestrating agent
// answer it instead of waiting out the idle timeout. The message echoes the exact session
// and question IDs so a follow-up answer call can copy them without guessing.
async function blocked(input: Target, dir: string, name: string): Promise<string | undefined> {
const [perms, qs] = await Promise.all([
input.client.permission.list({ directory: dir }),
input.client.question.list({ directory: dir }),
])
if (perms.error || qs.error)
throw new OrchestrationError("host_error", "The managed session blockers could not be read")
const mine = (qs.data ?? []).filter((value) => value.sessionID === input.sessionID)
const first = mine[0]
if (first) {
const detail = mine
.map((value) => {
const list = value.questions
.map((info, index) => {
const labels = info.options.slice(0, 8).map((option) => option.label)
return `${index + 1}. "${truncate(info.question, 200)}"${labels.length ? ` (options: ${labels.join(", ")})` : ""}`
})
.join("; ")
return `questionID "${value.id}": ${list}`
})
.join(" | ")
const count = first.questions.length
const more = mine.length > 1 ? ` Pending request IDs: ${mine.map((value) => value.id).join(", ")}.` : ""
return `The managed session ${input.sessionID} ("${name}") is waiting for input. Pending question requests: ${detail}. Call agent_manager with action "answer", sessionID "${input.sessionID}", questionID "${first.id}", and one label array per question in that request (${count} total), in order, before prompting.${more}`
}
if ((perms.data ?? []).some((value) => value.sessionID === input.sessionID)) {
return `The managed session ${input.sessionID} ("${name}") has a pending permission request; resolve it in Agent Manager before prompting`
}
return undefined
}

export async function prompt(input: {
client: KiloClient
root: string
state: WorktreeStateManager
sessionID: string
text: string
messageID: string
signal?: AbortSignal
idleTimeoutMs?: number
}): Promise<void> {
if (input.signal?.aborted) return
const target = await locate(input)
const blocker = await blocked(input, target.dir, target.name)
if (blocker) throw new OrchestrationError("unavailable_session", blocker)
await waitForIdle(input.client, target.dir, input.sessionID, input.signal, input.idleTimeoutMs ?? 30_000)
if (input.signal?.aborted) return
await input.client.session.promptAsync(
{
sessionID: input.sessionID,
directory: dir,
directory: target.dir,
messageID: `msg_agent_manager_${input.messageID}`,
parts: [{ type: "text", text: input.text }],
snapshotInitialization: SNAPSHOT_INITIALIZATION,
Expand All @@ -357,6 +417,53 @@ export async function prompt(input: {
)
}

export async function answer(input: {
client: KiloClient
root: string
state: WorktreeStateManager
sessionID: string
questionID?: string
answers: string[][]
}): Promise<{ questionID: string }> {
const dir = (await locate(input)).dir
const listed = await input.client.question.list({ directory: dir })
if (listed.error) throw new OrchestrationError("host_error", "The managed session questions could not be read")
const mine = (listed.data ?? []).filter((value) => value.sessionID === input.sessionID)
if (mine.length === 0) {
// The caller may have mixed up lookalike session IDs. Point at the sessions that
// actually hold pending questions so one retry with the right ID resolves it.
const others = (listed.data ?? []).map((value) => `${value.sessionID} (question ${value.id})`)
const hint = others.length ? ` Sessions with pending questions: ${others.join(", ")}.` : ""
throw new OrchestrationError("unavailable_session", `The managed session has no pending question to answer.${hint}`)
}
let target = mine[0]
if (input.questionID) {
const found = mine.find((value) => value.id === input.questionID)
if (!found)
throw new OrchestrationError(
"unavailable_session",
`The session has no pending question ${input.questionID}; pending: ${mine.map((value) => value.id).join(", ")}`,
)
target = found
} else if (mine.length > 1) {
throw new OrchestrationError(
"unavailable_session",
`Several questions are pending: ${mine.map((value) => value.id).join(", ")}. Name one with questionID.`,
)
}
if (input.answers.length !== target.questions.length) {
throw new OrchestrationError(
"unavailable_session",
`Question ${target.id} expects one answer array per question (${target.questions.length}), received ${input.answers.length}`,
)
}
await input.client.question.reply(
{ requestID: target.id, answers: input.answers, directory: dir },
{ throwOnError: true },
)
return { questionID: target.id }
}

async function waitForIdle(
client: KiloClient,
directory: string,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,7 @@ describe("AgentManagerOrchestrationBridge", () => {
const promptAsync = mock(async () => ({ data: undefined }))
const close = mock(async () => undefined)
const push = mock(() => undefined)
const questionReply = mock(async () => ({ data: true }))
const client = {
session: {
get: mock(async ({ sessionID, directory }: { sessionID?: string; directory?: string }) => ({
Expand All @@ -62,6 +63,7 @@ describe("AgentManagerOrchestrationBridge", () => {
},
question: {
list: mock(async () => ({ data: [] })),
reply: questionReply,
},
kilocode: {
agentManager: {
Expand Down Expand Up @@ -129,6 +131,7 @@ describe("AgentManagerOrchestrationBridge", () => {
managed,
promptAsync,
push,
questionReply,
rejections,
replies,
request,
Expand Down Expand Up @@ -305,6 +308,62 @@ describe("AgentManagerOrchestrationBridge", () => {
test.bridge.dispose()
})

it("answers a managed session's pending question through the backend reply route", async () => {
const test = harness()
;(test.client.question.list as ReturnType<typeof mock>).mockImplementation(async () => ({
data: [
{
id: "que_1",
sessionID: "ses_target",
questions: [{ header: "Approve", question: "Proceed?", options: [{ label: "Yes", description: "go" }] }],
},
],
}))

test.request({
id: "amr_answer",
sessionID: "ses_caller",
operation: "answer",
targetSessionID: "ses_target",
answers: [["Yes"]],
})
await waitFor(() => test.replies.length === 1)

expect(test.questionReply).toHaveBeenCalledTimes(1)
expect(test.questionReply).toHaveBeenCalledWith(
{ requestID: "que_1", answers: [["Yes"]], directory: dir },
{ throwOnError: true },
)
expect(test.replies[0]).toEqual({
requestID: "amr_answer",
directory: root,
result: { operation: "answer", sessionID: "ses_target", questionID: "que_1", resolved: true },
})
test.bridge.dispose()
})

it("rejects an answer when the target has no pending question", async () => {
const test = harness()

test.request({
id: "amr_answer_none",
sessionID: "ses_caller",
operation: "answer",
targetSessionID: "ses_target",
questionID: "que_gone",
answers: [["Yes"]],
})
await waitFor(() => test.rejections.length === 1)

expect(test.questionReply).not.toHaveBeenCalled()
expect(test.rejections[0]).toEqual({
requestID: "amr_answer_none",
directory: root,
error: { code: "unavailable_session", message: expect.stringContaining("no pending question") },
})
test.bridge.dispose()
})

it("rejects stopping a session not managed by the current workspace", async () => {
const test = harness()
test.request({
Expand Down
Loading
Loading