Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
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
116 changes: 107 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,59 @@ 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 }),
])
const mine = (qs.error ? [] : (qs.data ?? [])).filter((value) => value.sessionID === input.sessionID)
Comment thread
marius-kilocode marked this conversation as resolved.
Outdated
const first = mine[0]
if (first) {
const info = first.questions[0]
Comment thread
marius-kilocode marked this conversation as resolved.
Outdated
const labels = (info?.options ?? []).slice(0, 8).map((option) => option.label)
const detail = info
? ` asking "${truncate(info.question, 200)}"${labels.length ? ` (options: ${labels.join(", ")})` : ""}`
: ""
const more = mine.length > 1 ? ` ${mine.length} questions are pending; answer each in order.` : ""
return `The managed session ${input.sessionID} ("${name}") is waiting for an answer to question ${first.id}${detail}. Call agent_manager with action "answer", sessionID "${input.sessionID}", questionID "${first.id}", and one label array per question before prompting.${more}`
}
if ((perms.error ? [] : (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 +408,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