scriptoriumadmin/Roo-Code
1
0
Fork 0
mirror of https://github.com/RooVetGit/Roo-Code.git synced 2026-08-28 05:27:24 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Fix ACP task cancellation

Browse source
This commit is contained in:
cte 2026-01-11 18:40:45 -08:00
parent 7492abb9d0
commit 4ac8a460db
20 changed files with 1458 additions and 784 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

397
apps/cli/src/acp/__tests__/plan-translator.test.ts Normal file
View file

@ -0,0 +1,397 @@
import type { TodoItem } from "@roo-code/types"
import {
todoItemToPlanEntry,
todoListToPlanUpdate,
parseTodoListFromMessage,
isTodoListMessage,
extractTodoListFromMessage,
createPlanUpdateFromMessage,
type PriorityConfig,
} from "../translator/plan-translator.js"
describe("Plan Translator", () => {
// ===========================================================================
// Test Data
// ===========================================================================
const createTodoItem = (
content: string,
status: "pending" | "in_progress" | "completed",
id?: string,
): TodoItem => ({
id: id ?? `todo-${Date.now()}`,
content,
status,
})
// ===========================================================================
// todoItemToPlanEntry
// ===========================================================================
describe("todoItemToPlanEntry", () => {
it("converts a todo item to a plan entry with default config", () => {
const todo = createTodoItem("Implement feature X", "pending")
const entry = todoItemToPlanEntry(todo)
expect(entry).toEqual({
content: "Implement feature X",
priority: "medium",
status: "pending",
})
})
it("assigns high priority to in_progress items by default", () => {
const todo = createTodoItem("Working on feature", "in_progress")
const entry = todoItemToPlanEntry(todo)
expect(entry.priority).toBe("high")
expect(entry.status).toBe("in_progress")
})
it("preserves completed status", () => {
const todo = createTodoItem("Done task", "completed")
const entry = todoItemToPlanEntry(todo)
expect(entry.status).toBe("completed")
})
it("respects custom priority config", () => {
const todo = createTodoItem("Low priority task", "pending")
const config: PriorityConfig = {
defaultPriority: "low",
prioritizeInProgress: false,
prioritizeByOrder: false,
highPriorityCount: 3,
}
const entry = todoItemToPlanEntry(todo, 0, 1, config)
expect(entry.priority).toBe("low")
})
it("uses order-based priority when enabled", () => {
const config: PriorityConfig = {
defaultPriority: "medium",
prioritizeInProgress: false,
prioritizeByOrder: true,
highPriorityCount: 2,
}
// First 2 items should be high priority
const first = todoItemToPlanEntry(createTodoItem("First", "pending"), 0, 6, config)
const second = todoItemToPlanEntry(createTodoItem("Second", "pending"), 1, 6, config)
expect(first.priority).toBe("high")
expect(second.priority).toBe("high")
// Items 3-4 (first half) should be medium
const third = todoItemToPlanEntry(createTodoItem("Third", "pending"), 2, 6, config)
expect(third.priority).toBe("medium")
// Items past the halfway point should be low
const fifth = todoItemToPlanEntry(createTodoItem("Fifth", "pending"), 4, 6, config)
expect(fifth.priority).toBe("low")
})
it("prioritizes in_progress over order when both enabled", () => {
const config: PriorityConfig = {
defaultPriority: "low",
prioritizeInProgress: true,
prioritizeByOrder: true,
highPriorityCount: 1,
}
// Even at the end of the list, in_progress should be high
const inProgress = todoItemToPlanEntry(createTodoItem("In progress", "in_progress"), 5, 6, config)
expect(inProgress.priority).toBe("high")
})
})
// ===========================================================================
// todoListToPlanUpdate
// ===========================================================================
describe("todoListToPlanUpdate", () => {
it("converts an empty array to a plan with no entries", () => {
const update = todoListToPlanUpdate([])
expect(update).toEqual({
sessionUpdate: "plan",
entries: [],
})
})
it("converts a list of todos to a plan update", () => {
const todos: TodoItem[] = [
createTodoItem("Task 1", "completed"),
createTodoItem("Task 2", "in_progress"),
createTodoItem("Task 3", "pending"),
]
const update = todoListToPlanUpdate(todos)
expect(update.sessionUpdate).toBe("plan")
expect(update.entries).toHaveLength(3)
expect(update.entries[0]).toEqual({
content: "Task 1",
priority: "medium",
status: "completed",
})
expect(update.entries[1]).toEqual({
content: "Task 2",
priority: "high", // in_progress gets high priority
status: "in_progress",
})
expect(update.entries[2]).toEqual({
content: "Task 3",
priority: "medium",
status: "pending",
})
})
it("accepts partial config overrides", () => {
const todos = [createTodoItem("Task", "pending")]
const update = todoListToPlanUpdate(todos, { defaultPriority: "high" })
expect(update.entries[0]?.priority).toBe("high")
})
})
// ===========================================================================
// parseTodoListFromMessage
// ===========================================================================
describe("parseTodoListFromMessage", () => {
it("parses valid todo list JSON", () => {
const text = JSON.stringify({
tool: "updateTodoList",
todos: [
{ id: "1", content: "Task 1", status: "pending" },
{ id: "2", content: "Task 2", status: "completed" },
],
})
const result = parseTodoListFromMessage(text)
expect(result).toEqual([
{ id: "1", content: "Task 1", status: "pending" },
{ id: "2", content: "Task 2", status: "completed" },
])
})
it("returns null for invalid JSON", () => {
expect(parseTodoListFromMessage("not json")).toBeNull()
expect(parseTodoListFromMessage("{invalid}")).toBeNull()
})
it("returns null for JSON without updateTodoList tool", () => {
expect(parseTodoListFromMessage(JSON.stringify({ tool: "other" }))).toBeNull()
expect(parseTodoListFromMessage(JSON.stringify({ todos: [] }))).toBeNull()
})
it("returns null for JSON with non-array todos", () => {
expect(parseTodoListFromMessage(JSON.stringify({ tool: "updateTodoList", todos: "not array" }))).toBeNull()
})
})
// ===========================================================================
// isTodoListMessage
// ===========================================================================
describe("isTodoListMessage", () => {
it("detects tool ask messages with updateTodoList", () => {
const message = {
type: "ask",
ask: "tool",
text: JSON.stringify({ tool: "updateTodoList", todos: [] }),
}
expect(isTodoListMessage(message)).toBe(true)
})
it("detects user_edit_todos say messages", () => {
const message = {
type: "say",
say: "user_edit_todos",
text: JSON.stringify({ tool: "updateTodoList", todos: [] }),
}
expect(isTodoListMessage(message)).toBe(true)
})
it("returns false for other ask types", () => {
const message = {
type: "ask",
ask: "command",
text: JSON.stringify({ tool: "updateTodoList", todos: [] }),
}
expect(isTodoListMessage(message)).toBe(false)
})
it("returns false for other say types", () => {
const message = {
type: "say",
say: "text",
text: JSON.stringify({ tool: "updateTodoList", todos: [] }),
}
expect(isTodoListMessage(message)).toBe(false)
})
it("returns false for messages without text", () => {
const message = {
type: "ask",
ask: "tool",
}
expect(isTodoListMessage(message)).toBe(false)
})
it("returns false for tool messages with other tools", () => {
const message = {
type: "ask",
ask: "tool",
text: JSON.stringify({ tool: "read_file", path: "/some/path" }),
}
expect(isTodoListMessage(message)).toBe(false)
})
})
// ===========================================================================
// extractTodoListFromMessage
// ===========================================================================
describe("extractTodoListFromMessage", () => {
it("extracts todos from tool ask message", () => {
const todos = [{ id: "1", content: "Task", status: "pending" }]
const message = {
type: "ask",
ask: "tool",
text: JSON.stringify({ tool: "updateTodoList", todos }),
}
expect(extractTodoListFromMessage(message)).toEqual(todos)
})
it("extracts todos from user_edit_todos say message", () => {
const todos = [{ id: "1", content: "Task", status: "completed" }]
const message = {
type: "say",
say: "user_edit_todos",
text: JSON.stringify({ tool: "updateTodoList", todos }),
}
expect(extractTodoListFromMessage(message)).toEqual(todos)
})
it("returns null for non-todo messages", () => {
expect(extractTodoListFromMessage({ type: "say", say: "text", text: "Hello" })).toBeNull()
})
it("returns null for messages without text", () => {
expect(extractTodoListFromMessage({ type: "ask", ask: "tool" })).toBeNull()
})
})
// ===========================================================================
// createPlanUpdateFromMessage
// ===========================================================================
describe("createPlanUpdateFromMessage", () => {
it("creates plan update from valid todo message", () => {
const todos = [
{ id: "1", content: "First task", status: "in_progress" as const },
{ id: "2", content: "Second task", status: "pending" as const },
]
const message = {
type: "ask",
ask: "tool",
text: JSON.stringify({ tool: "updateTodoList", todos }),
}
const update = createPlanUpdateFromMessage(message)
expect(update).not.toBeNull()
expect(update?.sessionUpdate).toBe("plan")
expect(update?.entries).toHaveLength(2)
expect(update?.entries[0]).toEqual({
content: "First task",
priority: "high", // in_progress
status: "in_progress",
})
})
it("returns null for non-todo messages", () => {
const message = {
type: "say",
say: "text",
text: "Just some text",
}
expect(createPlanUpdateFromMessage(message)).toBeNull()
})
it("returns null for empty todo list", () => {
const message = {
type: "ask",
ask: "tool",
text: JSON.stringify({ tool: "updateTodoList", todos: [] }),
}
expect(createPlanUpdateFromMessage(message)).toBeNull()
})
it("accepts custom priority config", () => {
const todos = [{ id: "1", content: "Task", status: "pending" as const }]
const message = {
type: "ask",
ask: "tool",
text: JSON.stringify({ tool: "updateTodoList", todos }),
}
const update = createPlanUpdateFromMessage(message, { defaultPriority: "low" })
expect(update?.entries[0]?.priority).toBe("low")
})
})
// ===========================================================================
// Edge Cases
// ===========================================================================
describe("edge cases", () => {
it("handles todos with special characters in content", () => {
const todo = createTodoItem('Task with "quotes" and <html>', "pending")
const entry = todoItemToPlanEntry(todo)
expect(entry.content).toBe('Task with "quotes" and <html>')
})
it("handles todos with unicode content", () => {
const todo = createTodoItem("Task with emoji 🚀 and unicode ñ", "pending")
const entry = todoItemToPlanEntry(todo)
expect(entry.content).toBe("Task with emoji 🚀 and unicode ñ")
})
it("handles very long content", () => {
const longContent = "A".repeat(10000)
const todo = createTodoItem(longContent, "pending")
const entry = todoItemToPlanEntry(todo)
expect(entry.content).toBe(longContent)
})
it("handles malformed JSON gracefully", () => {
const message = {
type: "ask",
ask: "tool",
text: '{"tool": "updateTodoList", "todos": [{"broken',
}
expect(isTodoListMessage(message)).toBe(false)
expect(extractTodoListFromMessage(message)).toBeNull()
expect(createPlanUpdateFromMessage(message)).toBeNull()
})
})
})

397
apps/cli/src/acp/__tests__/session-plan-integration.test.ts Normal file
View file

@ -0,0 +1,397 @@
/**
* Integration tests for ACP Plan updates via session-event-handler.
*
* Tests the end-to-end flow of:
* 1. Extension sending todo list update messages
* 2. Session-event-handler detecting and translating them
* 3. ACP plan updates being sent to the connection
*/
import type { ClineMessage } from "@roo-code/types"
import {
SessionEventHandler,
createSessionEventHandler,
type SessionEventHandlerDeps,
} from "../session-event-handler.js"
import type { IAcpLogger, IDeltaTracker, IPromptStateMachine } from "../interfaces.js"
import { ToolHandlerRegistry } from "../tool-handler.js"
// =============================================================================
// Mock Setup
// =============================================================================
const createMockLogger = (): IAcpLogger => ({
debug: vi.fn(),
info: vi.fn(),
warn: vi.fn(),
error: vi.fn(),
request: vi.fn(),
response: vi.fn(),
notification: vi.fn(),
})
const createMockDeltaTracker = (): IDeltaTracker => ({
getDelta: vi.fn().mockReturnValue(null),
peekDelta: vi.fn().mockReturnValue(null),
reset: vi.fn(),
resetId: vi.fn(),
})
const createMockPromptState = (): IPromptStateMachine => ({
getState: vi.fn().mockReturnValue("processing"),
getAbortSignal: vi.fn().mockReturnValue(null),
getPromptText: vi.fn().mockReturnValue(""),
canStartPrompt: vi.fn().mockReturnValue(false),
isProcessing: vi.fn().mockReturnValue(true), // Return true so messages are processed
startPrompt: vi.fn().mockReturnValue(Promise.resolve({ stopReason: "end_turn" })),
complete: vi.fn().mockReturnValue("end_turn"),
transitionToComplete: vi.fn(),
cancel: vi.fn(),
reset: vi.fn(),
})
const createMockCommandStreamManager = () => ({
handleExecutionOutput: vi.fn(),
handleCommandOutput: vi.fn(),
isCommandOutputMessage: vi.fn().mockReturnValue(false),
trackCommand: vi.fn(),
reset: vi.fn(),
})
const createMockToolContentStreamManager = () => ({
handleToolContentStreaming: vi.fn(),
isToolAskMessage: vi.fn().mockReturnValue(false),
reset: vi.fn(),
})
const createMockExtensionClient = () => {
const handlers: Record<string, ((data: unknown) => void)[]> = {}
return {
on: vi.fn((event: string, handler: (data: unknown) => void) => {
handlers[event] = handlers[event] || []
handlers[event]!.push(handler)
return { on: vi.fn(), off: vi.fn() }
}),
off: vi.fn(),
emit: (event: string, data: unknown) => {
handlers[event]?.forEach((h) => h(data))
},
respond: vi.fn(),
approve: vi.fn(),
reject: vi.fn(),
}
}
const createMockExtensionHost = () => ({
on: vi.fn(),
off: vi.fn(),
client: createMockExtensionClient(),
activate: vi.fn().mockResolvedValue(undefined),
dispose: vi.fn().mockResolvedValue(undefined),
sendToExtension: vi.fn(),
})
// =============================================================================
// Tests
// =============================================================================
describe("Session Plan Integration", () => {
let eventHandler: SessionEventHandler
let mockSendUpdate: ReturnType<typeof vi.fn>
let mockClient: ReturnType<typeof createMockExtensionClient>
let deps: SessionEventHandlerDeps
beforeEach(() => {
mockSendUpdate = vi.fn()
mockClient = createMockExtensionClient()
deps = {
logger: createMockLogger(),
client: mockClient,
extensionHost: createMockExtensionHost(),
promptState: createMockPromptState(),
deltaTracker: createMockDeltaTracker(),
commandStreamManager: createMockCommandStreamManager(),
toolContentStreamManager: createMockToolContentStreamManager(),
toolHandlerRegistry: new ToolHandlerRegistry(),
sendUpdate: mockSendUpdate,
approveAction: vi.fn(),
respondWithText: vi.fn(),
sendToExtension: vi.fn(),
workspacePath: "/test/workspace",
initialModeId: "code",
isCancelling: vi.fn().mockReturnValue(false),
}
eventHandler = createSessionEventHandler(deps)
eventHandler.setupEventHandlers()
})
describe("todo list message detection", () => {
it("detects and sends plan update for updateTodoList tool ask message", () => {
const todoMessage: ClineMessage = {
type: "ask",
ask: "tool",
ts: Date.now(),
text: JSON.stringify({
tool: "updateTodoList",
todos: [
{ id: "1", content: "First task", status: "completed" },
{ id: "2", content: "Second task", status: "in_progress" },
{ id: "3", content: "Third task", status: "pending" },
],
}),
}
// Emit the message through the mock client
mockClient.emit("message", todoMessage)
// Verify plan update was sent
expect(mockSendUpdate).toHaveBeenCalledWith(
expect.objectContaining({
sessionUpdate: "plan",
entries: expect.arrayContaining([
expect.objectContaining({
content: "First task",
status: "completed",
priority: expect.any(String),
}),
expect.objectContaining({
content: "Second task",
status: "in_progress",
priority: "high", // in_progress gets high priority
}),
expect.objectContaining({
content: "Third task",
status: "pending",
priority: expect.any(String),
}),
]),
}),
)
})
it("detects and sends plan update for user_edit_todos say message", () => {
const editMessage: ClineMessage = {
type: "say",
say: "user_edit_todos",
ts: Date.now(),
text: JSON.stringify({
tool: "updateTodoList",
todos: [{ id: "1", content: "Edited task", status: "completed" }],
}),
}
mockClient.emit("message", editMessage)
expect(mockSendUpdate).toHaveBeenCalledWith(
expect.objectContaining({
sessionUpdate: "plan",
entries: [
expect.objectContaining({
content: "Edited task",
status: "completed",
}),
],
}),
)
})
it("does not send plan update for other tool ask messages", () => {
const otherToolMessage: ClineMessage = {
type: "ask",
ask: "tool",
ts: Date.now(),
text: JSON.stringify({
tool: "read_file",
path: "/some/file.txt",
}),
}
mockClient.emit("message", otherToolMessage)
// Should not have sent a plan update (but may send other updates)
const planUpdateCalls = mockSendUpdate.mock.calls.filter(
(call) => (call[0] as { sessionUpdate?: string })?.sessionUpdate === "plan",
)
expect(planUpdateCalls).toHaveLength(0)
})
it("does not send plan update for empty todo list", () => {
const emptyTodoMessage: ClineMessage = {
type: "ask",
ask: "tool",
ts: Date.now(),
text: JSON.stringify({
tool: "updateTodoList",
todos: [],
}),
}
mockClient.emit("message", emptyTodoMessage)
// Should not have sent a plan update for empty list
const planUpdateCalls = mockSendUpdate.mock.calls.filter(
(call) => (call[0] as { sessionUpdate?: string })?.sessionUpdate === "plan",
)
expect(planUpdateCalls).toHaveLength(0)
})
})
describe("priority assignment", () => {
it("assigns high priority to in_progress items", () => {
const todoMessage: ClineMessage = {
type: "ask",
ask: "tool",
ts: Date.now(),
text: JSON.stringify({
tool: "updateTodoList",
todos: [
{ id: "1", content: "Pending task", status: "pending" },
{ id: "2", content: "In progress task", status: "in_progress" },
{ id: "3", content: "Completed task", status: "completed" },
],
}),
}
mockClient.emit("message", todoMessage)
const planUpdateCall = mockSendUpdate.mock.calls.find(
(call) => (call[0] as { sessionUpdate?: string })?.sessionUpdate === "plan",
)
expect(planUpdateCall).toBeDefined()
const entries = (planUpdateCall![0] as { entries: Array<{ content: string; priority: string }> }).entries
const inProgressEntry = entries.find((e) => e.content === "In progress task")
expect(inProgressEntry?.priority).toBe("high")
})
it("assigns medium priority to pending and completed items by default", () => {
const todoMessage: ClineMessage = {
type: "ask",
ask: "tool",
ts: Date.now(),
text: JSON.stringify({
tool: "updateTodoList",
todos: [
{ id: "1", content: "Pending task", status: "pending" },
{ id: "2", content: "Completed task", status: "completed" },
],
}),
}
mockClient.emit("message", todoMessage)
const planUpdateCall = mockSendUpdate.mock.calls.find(
(call) => (call[0] as { sessionUpdate?: string })?.sessionUpdate === "plan",
)
expect(planUpdateCall).toBeDefined()
const entries = (planUpdateCall![0] as { entries: Array<{ content: string; priority: string }> }).entries
const pendingEntry = entries.find((e) => e.content === "Pending task")
const completedEntry = entries.find((e) => e.content === "Completed task")
expect(pendingEntry?.priority).toBe("medium")
expect(completedEntry?.priority).toBe("medium")
})
})
describe("message updates (streaming)", () => {
it("sends plan update when message is updated", () => {
const todoMessage: ClineMessage = {
type: "ask",
ask: "tool",
ts: Date.now(),
text: JSON.stringify({
tool: "updateTodoList",
todos: [{ id: "1", content: "Initial task", status: "pending" }],
}),
}
// First message
mockClient.emit("message", todoMessage)
// Updated message with more todos
const updatedMessage: ClineMessage = {
...todoMessage,
text: JSON.stringify({
tool: "updateTodoList",
todos: [
{ id: "1", content: "Initial task", status: "completed" },
{ id: "2", content: "New task", status: "pending" },
],
}),
}
mockClient.emit("messageUpdated", updatedMessage)
// Should have sent 2 plan updates (one for each message)
const planUpdateCalls = mockSendUpdate.mock.calls.filter(
(call) => (call[0] as { sessionUpdate?: string })?.sessionUpdate === "plan",
)
expect(planUpdateCalls.length).toBeGreaterThanOrEqual(2)
})
})
describe("logging", () => {
it("sends plan updates without verbose logging", () => {
const todoMessage: ClineMessage = {
type: "ask",
ask: "tool",
ts: Date.now(),
text: JSON.stringify({
tool: "updateTodoList",
todos: [{ id: "1", content: "Test task", status: "pending" }],
}),
}
mockClient.emit("message", todoMessage)
// Plan update should be sent without verbose logging
const planUpdateCalls = mockSendUpdate.mock.calls.filter(
(call) => (call[0] as { sessionUpdate?: string })?.sessionUpdate === "plan",
)
expect(planUpdateCalls).toHaveLength(1)
})
})
describe("reset behavior", () => {
it("continues to detect plan updates after reset", () => {
const todoMessage: ClineMessage = {
type: "ask",
ask: "tool",
ts: Date.now(),
text: JSON.stringify({
tool: "updateTodoList",
todos: [{ id: "1", content: "Task 1", status: "pending" }],
}),
}
mockClient.emit("message", todoMessage)
mockSendUpdate.mockClear()
// Reset the event handler
eventHandler.reset()
// Send another todo message
const anotherMessage: ClineMessage = {
...todoMessage,
ts: Date.now() + 1,
text: JSON.stringify({
tool: "updateTodoList",
todos: [{ id: "2", content: "Task 2", status: "pending" }],
}),
}
mockClient.emit("message", anotherMessage)
// Should still detect and send plan update
const planUpdateCalls = mockSendUpdate.mock.calls.filter(
(call) => (call[0] as { sessionUpdate?: string })?.sessionUpdate === "plan",
)
expect(planUpdateCalls).toHaveLength(1)
})
})
})

42
apps/cli/src/acp/__tests__/session.test.ts
View file

@ -1,12 +1,28 @@
import type * as acp from "@agentclientprotocol/sdk"
import { AgentLoopState } from "@/agent/agent-state.js"
// Track registered event handlers for simulation
type EventHandler = (data: unknown) => void
const clientEventHandlers: Map<string, EventHandler[]> = new Map()
vi.mock("@/agent/extension-host.js", () => {
const mockClient = {
on: vi.fn().mockReturnThis(),
on: vi.fn().mockImplementation((event: string, handler: EventHandler) => {
const handlers = clientEventHandlers.get(event) || []
handlers.push(handler)
clientEventHandlers.set(event, handlers)
return mockClient
}),
off: vi.fn().mockReturnThis(),
respond: vi.fn(),
approve: vi.fn(),
reject: vi.fn(),
getAgentState: vi.fn().mockReturnValue({
state: AgentLoopState.RUNNING,
isRunning: true,
isStreaming: false,
currentAsk: null,
}),
}
return {
@ -22,6 +38,19 @@ vi.mock("@/agent/extension-host.js", () => {
}
})
/**
* Simulate the extension responding to a cancel by emitting a state change to a terminal state.
*/
function simulateExtensionCancelResponse(): void {
const handlers = clientEventHandlers.get("stateChange") || []
handlers.forEach((handler) => {
handler({
previousState: { state: AgentLoopState.RUNNING, isRunning: true, isStreaming: false },
currentState: { state: AgentLoopState.IDLE, isRunning: false, isStreaming: false },
})
})
}
import { AcpSession, type AcpSessionOptions } from "../session.js"
import { ExtensionHost } from "@/agent/extension-host.js"
@ -37,6 +66,9 @@ describe("AcpSession", () => {
}
beforeEach(() => {
// Clear registered event handlers between tests
clientEventHandlers.clear()
mockConnection = {
sessionUpdate: vi.fn().mockResolvedValue(undefined),
requestPermission: vi.fn().mockResolvedValue({
@ -56,6 +88,7 @@ describe("AcpSession", () => {
afterEach(() => {
vi.clearAllMocks()
clientEventHandlers.clear()
})
describe("create", () => {
@ -140,8 +173,9 @@ describe("AcpSession", () => {
}),
)
// Cancel to resolve the promise
// Cancel to resolve the promise - simulate extension responding to cancel
session.cancel()
simulateExtensionCancelResponse()
const result = await promptPromise
expect(result.stopReason).toBe("cancelled")
})
@ -174,6 +208,7 @@ describe("AcpSession", () => {
)
session.cancel()
simulateExtensionCancelResponse()
await promptPromise
})
})
@ -196,8 +231,9 @@ describe("AcpSession", () => {
prompt: [{ type: "text", text: "Hello" }],
})
// Cancel
// Cancel and simulate extension responding
session.cancel()
simulateExtensionCancelResponse()
expect(mockHostInstance.sendToExtension).toHaveBeenCalledWith({ type: "cancelTask" })

374
apps/cli/src/acp/__tests__/update-buffer.test.ts
View file

@ -1,374 +0,0 @@
import type * as acp from "@agentclientprotocol/sdk"
import { UpdateBuffer } from "../update-buffer.js"
type SessionUpdate = acp.SessionNotification["update"]
describe("UpdateBuffer", () => {
let sentUpdates: Array<{ sessionUpdate: string; content?: unknown }>
let sendUpdate: (update: SessionUpdate) => Promise<void>
beforeEach(() => {
sentUpdates = []
sendUpdate = vi.fn(async (update) => {
sentUpdates.push(update)
})
vi.useFakeTimers()
})
afterEach(() => {
vi.useRealTimers()
})
describe("text chunk buffering", () => {
it("should buffer agent_message_chunk updates", async () => {
const buffer = new UpdateBuffer(sendUpdate, {
minBufferSize: 100,
flushDelayMs: 50,
})
await buffer.queueUpdate({
sessionUpdate: "agent_message_chunk",
content: { type: "text", text: "Hello" },
})
// Should not be sent immediately
expect(sentUpdates).toHaveLength(0)
expect(buffer.getBufferSizes().message).toBe(5)
})
it("should buffer agent_thought_chunk updates", async () => {
const buffer = new UpdateBuffer(sendUpdate, {
minBufferSize: 100,
flushDelayMs: 50,
})
await buffer.queueUpdate({
sessionUpdate: "agent_thought_chunk",
content: { type: "text", text: "Thinking..." },
})
// Should not be sent immediately
expect(sentUpdates).toHaveLength(0)
expect(buffer.getBufferSizes().thought).toBe(11)
})
it("should batch multiple text chunks together", async () => {
const buffer = new UpdateBuffer(sendUpdate, {
minBufferSize: 100,
flushDelayMs: 50,
})
await buffer.queueUpdate({
sessionUpdate: "agent_message_chunk",
content: { type: "text", text: "Hello " },
})
await buffer.queueUpdate({
sessionUpdate: "agent_message_chunk",
content: { type: "text", text: "World" },
})
expect(sentUpdates).toHaveLength(0)
expect(buffer.getBufferSizes().message).toBe(11)
// Flush and check combined content
await buffer.flush()
expect(sentUpdates).toHaveLength(1)
expect(sentUpdates[0]).toEqual({
sessionUpdate: "agent_message_chunk",
content: { type: "text", text: "Hello World" },
})
})
})
describe("size threshold flushing", () => {
it("should flush when buffer reaches minBufferSize", async () => {
const buffer = new UpdateBuffer(sendUpdate, {
minBufferSize: 10,
flushDelayMs: 1000, // Long delay to ensure size triggers flush
})
await buffer.queueUpdate({
sessionUpdate: "agent_message_chunk",
content: { type: "text", text: "Hello World!" }, // 12 chars, exceeds 10
})
// Should have flushed due to size
expect(sentUpdates).toHaveLength(1)
expect(sentUpdates[0]).toEqual({
sessionUpdate: "agent_message_chunk",
content: { type: "text", text: "Hello World!" },
})
})
it("should consider combined buffer sizes", async () => {
const buffer = new UpdateBuffer(sendUpdate, {
minBufferSize: 15,
flushDelayMs: 1000,
})
await buffer.queueUpdate({
sessionUpdate: "agent_message_chunk",
content: { type: "text", text: "Hello" }, // 5 chars
})
await buffer.queueUpdate({
sessionUpdate: "agent_thought_chunk",
content: { type: "text", text: "Thinking!" }, // 9 chars, total 14
})
// Not flushed yet (14 < 15)
expect(sentUpdates).toHaveLength(0)
await buffer.queueUpdate({
sessionUpdate: "agent_message_chunk",
content: { type: "text", text: "X" }, // 1 more, total 15
})
// Should have flushed (15 >= 15)
expect(sentUpdates).toHaveLength(2) // message and thought
})
})
describe("time threshold flushing", () => {
it("should flush after flushDelayMs", async () => {
const buffer = new UpdateBuffer(sendUpdate, {
minBufferSize: 1000,
flushDelayMs: 50,
})
await buffer.queueUpdate({
sessionUpdate: "agent_message_chunk",
content: { type: "text", text: "Hello" },
})
expect(sentUpdates).toHaveLength(0)
// Advance time past the flush delay
await vi.advanceTimersByTimeAsync(60)
expect(sentUpdates).toHaveLength(1)
expect(sentUpdates[0]).toEqual({
sessionUpdate: "agent_message_chunk",
content: { type: "text", text: "Hello" },
})
})
it("should reset timer on new content", async () => {
const buffer = new UpdateBuffer(sendUpdate, {
minBufferSize: 1000,
flushDelayMs: 50,
})
await buffer.queueUpdate({
sessionUpdate: "agent_message_chunk",
content: { type: "text", text: "A" },
})
// Advance 30ms (not enough to flush)
await vi.advanceTimersByTimeAsync(30)
expect(sentUpdates).toHaveLength(0)
// Add more content (should NOT reset timer in current impl)
await buffer.queueUpdate({
sessionUpdate: "agent_message_chunk",
content: { type: "text", text: "B" },
})
// Advance another 30ms (total 60ms from first queue)
await vi.advanceTimersByTimeAsync(30)
// Should have flushed
expect(sentUpdates).toHaveLength(1)
expect(sentUpdates[0]!.content).toEqual({ type: "text", text: "AB" })
})
})
describe("non-bufferable updates", () => {
it("should send tool_call updates immediately", async () => {
const buffer = new UpdateBuffer(sendUpdate, {
minBufferSize: 1000,
flushDelayMs: 1000,
})
await buffer.queueUpdate({
sessionUpdate: "tool_call",
toolCallId: "test-123",
title: "Test Tool",
kind: "read",
status: "in_progress",
})
// Should be sent immediately
expect(sentUpdates).toHaveLength(1)
expect(sentUpdates[0]!.sessionUpdate).toBe("tool_call")
})
it("should send tool_call_update updates immediately", async () => {
const buffer = new UpdateBuffer(sendUpdate, {
minBufferSize: 1000,
flushDelayMs: 1000,
})
await buffer.queueUpdate({
sessionUpdate: "tool_call_update",
toolCallId: "test-123",
status: "completed",
})
expect(sentUpdates).toHaveLength(1)
expect(sentUpdates[0]!.sessionUpdate).toBe("tool_call_update")
})
it("should flush buffered content before sending non-bufferable update", async () => {
const buffer = new UpdateBuffer(sendUpdate, {
minBufferSize: 1000,
flushDelayMs: 1000,
})
// Buffer some text first
await buffer.queueUpdate({
sessionUpdate: "agent_message_chunk",
content: { type: "text", text: "Before tool" },
})
// Send tool call - should flush text first
await buffer.queueUpdate({
sessionUpdate: "tool_call",
toolCallId: "test-123",
title: "Test Tool",
kind: "read",
status: "in_progress",
})
// Text should come first, then tool call
expect(sentUpdates).toHaveLength(2)
expect(sentUpdates[0]!.sessionUpdate).toBe("agent_message_chunk")
expect(sentUpdates[1]!.sessionUpdate).toBe("tool_call")
})
})
describe("flush method", () => {
it("should flush all buffered content", async () => {
const buffer = new UpdateBuffer(sendUpdate, {
minBufferSize: 1000,
flushDelayMs: 1000,
})
await buffer.queueUpdate({
sessionUpdate: "agent_message_chunk",
content: { type: "text", text: "Message" },
})
await buffer.queueUpdate({
sessionUpdate: "agent_thought_chunk",
content: { type: "text", text: "Thought" },
})
expect(sentUpdates).toHaveLength(0)
await buffer.flush()
expect(sentUpdates).toHaveLength(2)
expect(sentUpdates[0]!.sessionUpdate).toBe("agent_message_chunk")
expect(sentUpdates[1]!.sessionUpdate).toBe("agent_thought_chunk")
})
it("should be idempotent when buffer is empty", async () => {
const buffer = new UpdateBuffer(sendUpdate, {
minBufferSize: 1000,
flushDelayMs: 1000,
})
await buffer.flush()
await buffer.flush()
await buffer.flush()
expect(sentUpdates).toHaveLength(0)
})
})
describe("reset method", () => {
it("should clear all buffered content", async () => {
const buffer = new UpdateBuffer(sendUpdate, {
minBufferSize: 1000,
flushDelayMs: 1000,
})
await buffer.queueUpdate({
sessionUpdate: "agent_message_chunk",
content: { type: "text", text: "Hello" },
})
expect(buffer.getBufferSizes().message).toBe(5)
buffer.reset()
expect(buffer.getBufferSizes().message).toBe(0)
expect(buffer.getBufferSizes().thought).toBe(0)
// Flushing should send nothing
await buffer.flush()
expect(sentUpdates).toHaveLength(0)
})
it("should cancel pending flush timer", async () => {
const buffer = new UpdateBuffer(sendUpdate, {
minBufferSize: 1000,
flushDelayMs: 50,
})
await buffer.queueUpdate({
sessionUpdate: "agent_message_chunk",
content: { type: "text", text: "Hello" },
})
buffer.reset()
// Advance past flush delay
await vi.advanceTimersByTimeAsync(100)
// Nothing should have been sent
expect(sentUpdates).toHaveLength(0)
})
})
describe("default options", () => {
it("should use defaults (200 chars, 500ms)", async () => {
const buffer = new UpdateBuffer(sendUpdate)
// Default minBufferSize is 200
const longText = "A".repeat(199)
await buffer.queueUpdate({
sessionUpdate: "agent_message_chunk",
content: { type: "text", text: longText },
})
// Not flushed yet (199 < 200)
expect(sentUpdates).toHaveLength(0)
await buffer.queueUpdate({
sessionUpdate: "agent_message_chunk",
content: { type: "text", text: "B" }, // 200 total
})
// Should have flushed (200 >= 200)
expect(sentUpdates).toHaveLength(1)
})
it("should flush after 500ms by default", async () => {
const buffer = new UpdateBuffer(sendUpdate)
await buffer.queueUpdate({
sessionUpdate: "agent_message_chunk",
content: { type: "text", text: "Hello" },
})
// Not flushed at 400ms
await vi.advanceTimersByTimeAsync(400)
expect(sentUpdates).toHaveLength(0)
// Flushed at 500ms
await vi.advanceTimersByTimeAsync(150)
expect(sentUpdates).toHaveLength(1)
})
})
})

23
apps/cli/src/acp/command-stream.ts
View file

@ -91,7 +91,6 @@ export class CommandStreamManager {
*/
trackCommand(toolCallId: string, command: string, ts: number): void {
this.pendingCommandCalls.set(toolCallId, { toolCallId, command, ts })
this.logger.debug("CommandStream", `Tracking command: ${toolCallId}`)
}
/**
@ -107,11 +106,6 @@ export class CommandStreamManager {
const output = message.text || ""
const isPartial = message.partial === true
this.logger.debug(
"CommandStream",
`handleCommandOutput: partial=${message.partial}, text length=${output.length}`,
)
// Skip partial updates - streaming is handled by handleExecutionOutput()
if (isPartial) {
return
@ -121,12 +115,9 @@ export class CommandStreamManager {
const pendingCall = this.findMostRecentPendingCommand()
if (pendingCall) {
this.logger.debug("CommandStream", `Command completed: ${pendingCall.toolCallId}`)
// Send closing code fence as agent_message_chunk if we had streaming output
const hadStreamingOutput = this.commandCodeFencesSent.has(pendingCall.toolCallId)
if (hadStreamingOutput) {
this.logger.debug("CommandStream", "Sending closing code fence via agent_message_chunk")
this.sendUpdate({
sessionUpdate: "agent_message_chunk",
content: { type: "text", text: "```\n" },
@ -161,11 +152,6 @@ export class CommandStreamManager {
* Uses executionId → toolCallId mapping for robust routing.
*/
handleExecutionOutput(executionId: string, output: string): void {
this.logger.debug(
"CommandStream",
`handleExecutionOutput: executionId=${executionId}, output length=${output.length}`,
)
// Find or establish the toolCallId for this executionId
let toolCallId = this.executionToToolCallId.get(executionId)
@ -173,12 +159,10 @@ export class CommandStreamManager {
// First output for this executionId - establish the mapping
const pendingCall = this.findMostRecentPendingCommand()
if (!pendingCall) {
this.logger.debug("CommandStream", "No pending command, skipping execution output")
return
}
toolCallId = pendingCall.toolCallId
this.executionToToolCallId.set(executionId, toolCallId)
this.logger.debug("CommandStream", `Mapped executionId ${executionId} → toolCallId ${toolCallId}`)
}
// Use executionId as the message key for delta tracking
@ -191,7 +175,6 @@ export class CommandStreamManager {
const isFirstChunk = !this.commandCodeFencesSent.has(toolCallId)
if (isFirstChunk) {
this.commandCodeFencesSent.add(toolCallId)
this.logger.debug("CommandStream", `Sending opening code fence for toolCallId ${toolCallId}`)
this.sendUpdate({
sessionUpdate: "agent_message_chunk",
content: { type: "text", text: "```\n" },
@ -199,7 +182,6 @@ export class CommandStreamManager {
}
// Send the delta as agent_message_chunk for Zed visibility
this.logger.debug("CommandStream", `Streaming command output via agent_message_chunk: ${delta.length} chars`)
this.sendUpdate({
sessionUpdate: "agent_message_chunk",
content: { type: "text", text: delta },
@ -220,14 +202,9 @@ export class CommandStreamManager {
reset(): void {
// Clear all pending commands - any from previous prompts are now stale
// and would cause duplicate completion messages if not cleaned up
const staleCount = this.pendingCommandCalls.size
if (staleCount > 0) {
this.logger.debug("CommandStream", `Clearing ${staleCount} stale pending commands`)
}
this.pendingCommandCalls.clear()
this.commandCodeFencesSent.clear()
this.executionToToolCallId.clear()
this.logger.debug("CommandStream", "Reset command stream state")
}
/**

2
apps/cli/src/acp/index.ts
View file

@ -16,7 +16,6 @@ export type {
IContentFormatter,
IExtensionClient,
IExtensionHost,
IUpdateBuffer,
IDeltaTracker,
IPromptStateMachine,
ICommandStreamManager,
@ -34,7 +33,6 @@ export { acpLog } from "./logger.js"
// Utilities
export { DeltaTracker } from "./delta-tracker.js"
export { UpdateBuffer, type UpdateBufferOptions } from "./update-buffer.js"
// Shared utility functions
export {

32
apps/cli/src/acp/interfaces.ts
View file

@ -193,30 +193,6 @@ export interface IExtensionHost {
sendToExtension(message: unknown): void
}
// =============================================================================
// Update Buffer Interface
// =============================================================================
/**
* Interface for update buffering.
*/
export interface IUpdateBuffer {
/**
* Queue an update for sending.
*/
queueUpdate(update: acp.SessionNotification["update"]): Promise<void>
/**
* Flush all pending buffered content.
*/
flush(): Promise<void>
/**
* Reset the buffer state.
*/
reset(): void
}
// =============================================================================
// Delta Tracker Interface
// =============================================================================
@ -304,6 +280,12 @@ export interface IPromptStateMachine {
*/
complete(success: boolean): acp.StopReason
/**
* Transition to completion with a specific stop reason.
* This allows direct control over the stop reason (e.g., for cancellation).
*/
transitionToComplete(stopReason: acp.StopReason): void
/**
* Cancel the current prompt.
*/
@ -371,8 +353,6 @@ export interface AcpSessionDependencies {
contentFormatter?: IContentFormatter
/** Delta tracker factory */
createDeltaTracker?: () => IDeltaTracker
/** Update buffer factory */
createUpdateBuffer?: (sendUpdate: (update: acp.SessionNotification["update"]) => Promise<void>) => IUpdateBuffer
/** Prompt state machine factory */
createPromptStateMachine?: () => IPromptStateMachine
}

15
apps/cli/src/acp/prompt-state.ts
View file

@ -127,12 +127,10 @@ export class PromptStateMachine {
*/
startPrompt(promptText: string): Promise<PromptCompletionResult> {
if (this.state !== "idle") {
this.logger.warn("PromptStateMachine", `Cannot start prompt in state: ${this.state}`)
// Cancel existing prompt first
this.cancel()
}
this.logger.debug("PromptStateMachine", "Transitioning: idle -> processing")
this.state = "processing"
this.abortController = new AbortController()
this.currentPromptText = promptText
@ -143,7 +141,6 @@ export class PromptStateMachine {
// Handle abort signal
this.abortController?.signal.addEventListener("abort", () => {
if (this.state === "processing") {
this.logger.debug("PromptStateMachine", "Abort signal received")
this.transitionToComplete("cancelled")
}
})
@ -169,11 +166,9 @@ export class PromptStateMachine {
*/
cancel(): void {
if (this.state !== "processing") {
this.logger.debug("PromptStateMachine", `Cancel ignored in state: ${this.state}`)
return
}
this.logger.debug("PromptStateMachine", "Cancelling prompt")
this.abortController?.abort()
// Note: The abort handler will call transitionToComplete
}
@ -184,8 +179,6 @@ export class PromptStateMachine {
* Should be called when starting a new prompt to ensure clean state.
*/
reset(): void {
this.logger.debug("PromptStateMachine", `Resetting from state: ${this.state}`)
// Clean up any pending resources
if (this.abortController) {
this.abortController.abort()
@ -198,20 +191,18 @@ export class PromptStateMachine {
}
// ===========================================================================
// Private Methods
// Public Methods (for direct control)
// ===========================================================================
/**
* Transition to completion and resolve the promise.
* This is public to allow direct control of the stop reason (e.g., for cancellation).
*/
private transitionToComplete(stopReason: acp.StopReason): void {
transitionToComplete(stopReason: acp.StopReason): void {
if (this.state !== "processing") {
this.logger.debug("PromptStateMachine", `Already completed, ignoring transition with reason: ${stopReason}`)
return
}
this.logger.debug("PromptStateMachine", `Transitioning: processing -> idle (reason: ${stopReason})`)
this.state = "idle"
// Resolve the promise

71
apps/cli/src/acp/session-event-handler.ts
View file

@ -10,7 +10,13 @@ import type { ClineMessage, ClineAsk, ClineSay, ExtensionMessage, ExtensionState
import type { WaitingForInputEvent, TaskCompletedEvent, CommandExecutionOutputEvent } from "@/agent/events.js"
import { translateToAcpUpdate, isPermissionAsk, isCompletionAsk } from "./translator.js"
import {
translateToAcpUpdate,
isPermissionAsk,
isCompletionAsk,
isTodoListMessage,
createPlanUpdateFromMessage,
} from "./translator.js"
import { isUserEcho } from "./utils/index.js"
import type {
IAcpLogger,
@ -122,6 +128,8 @@ export interface SessionEventHandlerDeps {
workspacePath: string
/** Initial mode ID */
initialModeId: string
/** Callback to check if cancellation is in progress */
isCancelling: () => boolean
}
/**
@ -163,6 +171,7 @@ export class SessionEventHandler {
private readonly respondWithText: (text: string) => void
private readonly sendToExtension: (message: unknown) => void
private readonly workspacePath: string
private readonly isCancelling: () => boolean
private taskCompletedCallback: TaskCompletedCallback | null = null
private modeChangedCallback: ModeChangedCallback | null = null
@ -199,6 +208,7 @@ export class SessionEventHandler {
this.sendToExtension = deps.sendToExtension
this.workspacePath = deps.workspacePath
this.currentModeId = deps.initialModeId
this.isCancelling = deps.isCancelling
}
// ===========================================================================
@ -301,10 +311,30 @@ export class SessionEventHandler {
* which message types should be delta-streamed and how.
*/
private handleMessage(message: ClineMessage): void {
this.logger.debug(
"SessionEventHandler",
`Message received: type=${message.type}, say=${message.say}, ask=${message.ask}, ts=${message.ts}, partial=${message.partial}`,
)
// Don't process messages if there's no active prompt
// NOTE: isCancelling guard REMOVED - we now show all content even during cancellation
// so the user can see exactly what was produced before the task paused
if (!this.promptState.isProcessing()) {
return
}
// === TEST LOGGING: Log messages that arrive during cancellation ===
if (this.isCancelling()) {
const msgType = message.type === "say" ? `say:${message.say}` : `ask:${message.ask}`
const partial = message.partial ? "PARTIAL" : "COMPLETE"
this.logger.info("EventHandler", `MSG DURING CANCEL (processing): ${msgType} ${partial} ts=${message.ts}`)
}
// Handle todo list updates - translate to ACP plan updates
// Detects both tool asks for updateTodoList and user_edit_todos say messages
if (isTodoListMessage(message)) {
const planUpdate = createPlanUpdateFromMessage(message)
if (planUpdate) {
this.sendUpdate(planUpdate)
}
// Don't return - let the message also be processed by other handlers
// (e.g., for permission requests that may follow)
}
// Handle streaming for tool ask messages (file creates/edits)
// These contain content that grows as the LLM generates it
@ -326,7 +356,6 @@ export class SessionEventHandler {
if (config) {
// Filter out user message echo
if (message.say === "text" && isUserEcho(message.text, this.promptState.getPromptText())) {
this.logger.debug("SessionEventHandler", `Skipping user echo (${message.text.length} chars)`)
return
}
@ -349,9 +378,6 @@ export class SessionEventHandler {
// For non-streaming message types, use the translator
const update = translateToAcpUpdate(message)
if (update) {
this.logger.notification("sessionUpdate", {
updateKind: (update as { sessionUpdate?: string }).sessionUpdate,
})
this.sendUpdate(update)
}
}
@ -366,18 +392,24 @@ export class SessionEventHandler {
private async handleWaitingForInput(event: WaitingForInputEvent): Promise<void> {
const { ask, message } = event
const askType = ask as ClineAsk
this.logger.debug("SessionEventHandler", `Waiting for input: ask=${askType}`)
// Don't auto-approve asks if there's no active prompt or if cancellation is in progress
if (!this.promptState.isProcessing() || this.isCancelling()) {
// === TEST LOGGING: Skipped ask due to cancellation ===
if (this.isCancelling()) {
this.logger.info("EventHandler", `ASK SKIPPED (cancelling): ask=${askType} ts=${message.ts}`)
}
return
}
// Handle permission-required asks
if (isPermissionAsk(askType)) {
this.logger.info("SessionEventHandler", `Permission request: ${askType}`)
this.handlePermissionRequest(message, askType)
return
}
// Handle completion asks
if (isCompletionAsk(askType)) {
this.logger.debug("SessionEventHandler", "Completion ask - handled by taskCompleted event")
// Completion is handled by taskCompleted event
return
}
@ -386,27 +418,23 @@ export class SessionEventHandler {
// In a more sophisticated implementation, these could be surfaced
// to the ACP client for user input
if (askType === "followup") {
this.logger.debug("SessionEventHandler", "Auto-responding to followup")
this.respondWithText("")
return
}
// Handle resume_task - auto-resume
if (askType === "resume_task") {
this.logger.debug("SessionEventHandler", "Auto-approving resume_task")
this.approveAction()
return
}
// Handle API failures - auto-retry for now
if (askType === "api_req_failed") {
this.logger.warn("SessionEventHandler", "API request failed, auto-retrying")
this.approveAction()
return
}
// Default: approve and continue
this.logger.debug("SessionEventHandler", `Auto-approving unknown ask type: ${askType}`)
this.approveAction()
}
@ -429,7 +457,6 @@ export class SessionEventHandler {
// Check if we've already processed this permission request
if (this.processedPermissions.has(permissionKey)) {
this.logger.debug("SessionEventHandler", `Skipping duplicate permission request: ${ask}`)
// Still need to approve the action to unblock the extension
this.approveAction()
return
@ -444,9 +471,6 @@ export class SessionEventHandler {
// Dispatch to the appropriate handler via the registry
const result = this.toolHandlerRegistry.handle(context)
this.logger.debug("SessionEventHandler", `Auto-approving tool: ask=${ask}`)
this.logger.debug("SessionEventHandler", `Sending tool_call update`)
// Send the initial in_progress update
this.sendUpdate(result.initialUpdate)
@ -458,7 +482,6 @@ export class SessionEventHandler {
// Send completion update if available (non-command tools)
if (result.completionUpdate) {
this.logger.debug("SessionEventHandler", `Sending tool_call_update (completed)`)
this.sendUpdate(result.completionUpdate)
}
@ -474,8 +497,6 @@ export class SessionEventHandler {
* Handle task completion.
*/
private handleTaskCompleted(event: TaskCompletedEvent): void {
this.logger.info("SessionEventHandler", `Task completed: success=${event.success}`)
if (this.taskCompletedCallback) {
this.taskCompletedCallback(event.success)
}
@ -491,7 +512,6 @@ export class SessionEventHandler {
private handleExtensionMessage(msg: ExtensionMessage): void {
// Handle "modes" message - list of available modes
if (msg.type === "modes" && msg.modes) {
this.logger.debug("SessionEventHandler", `Received modes: ${msg.modes.length} modes`)
this.availableModes = msg.modes.map((m) => ({
id: m.slug,
name: m.name,
@ -503,9 +523,7 @@ export class SessionEventHandler {
if (msg.type === "state" && msg.state) {
const state = msg.state as ExtensionState
if (state.mode && state.mode !== this.currentModeId) {
const previousMode = this.currentModeId
this.currentModeId = state.mode
this.logger.info("SessionEventHandler", `Mode changed: ${previousMode} -> ${this.currentModeId}`)
// Send mode update notification
this.sendUpdate({
@ -535,7 +553,6 @@ export class SessionEventHandler {
name: m.name,
description: undefined,
}))
this.logger.debug("SessionEventHandler", `Updated available modes: ${this.availableModes.length} modes`)
}
}

176
apps/cli/src/acp/session.ts
View file

@ -15,12 +15,12 @@ import {
} from "@agentclientprotocol/sdk"
import { type ExtensionHostOptions, ExtensionHost } from "@/agent/extension-host.js"
import { AgentLoopState } from "@/agent/agent-state.js"
import { DEFAULT_MODELS } from "./types.js"
import { extractPromptText, extractPromptImages } from "./translator.js"
import { acpLog } from "./logger.js"
import { DeltaTracker } from "./delta-tracker.js"
import { UpdateBuffer } from "./update-buffer.js"
import { PromptStateMachine } from "./prompt-state.js"
import { ToolHandlerRegistry } from "./tool-handler.js"
import { CommandStreamManager } from "./command-stream.js"
@ -30,7 +30,6 @@ import type {
IAcpSession,
IAcpLogger,
IDeltaTracker,
IUpdateBuffer,
IPromptStateMachine,
AcpSessionDependencies,
} from "./interfaces.js"
@ -70,9 +69,6 @@ export class AcpSession implements IAcpSession {
/** Delta tracker for streaming content - ensures only new text is sent */
private readonly deltaTracker: IDeltaTracker
/** Update buffer for batching session updates to reduce message frequency */
private readonly updateBuffer: IUpdateBuffer
/** Tool handler registry for polymorphic tool dispatch */
private readonly toolHandlerRegistry: ToolHandlerRegistry
@ -91,6 +87,9 @@ export class AcpSession implements IAcpSession {
/** Current model ID */
private currentModelId: string = DEFAULT_MODELS[0]!.modelId
/** Track if we're in the process of cancelling a task */
private isCancelling: boolean = false
private constructor(
private readonly sessionId: string,
private readonly extensionHost: ExtensionHost,
@ -106,23 +105,13 @@ export class AcpSession implements IAcpSession {
this.promptState = deps.createPromptStateMachine?.() ?? new PromptStateMachine({ logger: this.logger })
this.deltaTracker = deps.createDeltaTracker?.() ?? new DeltaTracker()
// Initialize update buffer with the actual send function.
// Uses defaults: 200 chars min buffer, 500ms delay.
// Wrap sendUpdateDirect to match the expected Promise<void> signature.
const sendDirectAdapter = async (update: SessionNotification["update"]): Promise<void> => {
await this.sendUpdateDirect(update)
// Result is logged internally; adapter converts to void for interface compatibility.
}
this.updateBuffer =
deps.createUpdateBuffer?.(sendDirectAdapter) ?? new UpdateBuffer(sendDirectAdapter, { logger: this.logger })
// Initialize tool handler registry.
this.toolHandlerRegistry = new ToolHandlerRegistry()
// Create send update callback for stream managers.
// Updates are sent directly to preserve chunk ordering.
const sendUpdate = (update: SessionNotification["update"]) => {
void this.sendUpdate(update)
void this.sendUpdateDirect(update)
}
// Initialize stream managers with injected logger.
@ -155,9 +144,48 @@ export class AcpSession implements IAcpSession {
this.extensionHost.sendToExtension(message as Parameters<typeof this.extensionHost.sendToExtension>[0]),
workspacePath,
initialModeId: initialMode,
isCancelling: () => this.isCancelling,
})
this.eventHandler.onTaskCompleted((success) => this.handleTaskCompleted(success))
// Listen for state changes to log and detect cancellation completion
this.extensionHost.client.on("stateChange", (event) => {
const prev = event.previousState
const curr = event.currentState
// Only log if something actually changed
const stateChanged =
prev.state !== curr.state ||
prev.isRunning !== curr.isRunning ||
prev.isStreaming !== curr.isStreaming ||
prev.currentAsk !== curr.currentAsk
if (stateChanged) {
this.logger.info(
"ExtensionClient",
`STATE: ${prev.state} → ${curr.state} (running=${curr.isRunning}, streaming=${curr.isStreaming}, ask=${curr.currentAsk || "none"})`,
)
}
// If we're cancelling and the extension transitions to NO_TASK or IDLE, complete the cancellation
// NO_TASK: messages were cleared
// IDLE: task stopped (e.g., completion_result, api_req_failed, or just stopped)
if (this.isCancelling) {
const newState = curr.state
const isTerminalState =
newState === AgentLoopState.NO_TASK ||
newState === AgentLoopState.IDLE ||
newState === AgentLoopState.RESUMABLE
// Also check if the agent is no longer running/streaming (it has stopped processing)
const hasStopped = !curr.isRunning && !curr.isStreaming
if (isTerminalState || hasStopped) {
this.handleCancellationComplete()
}
}
})
}
// ===========================================================================
@ -185,9 +213,6 @@ export class AcpSession implements IAcpSession {
options: AcpSessionOptions,
deps: AcpSessionDependencies = {},
): Promise<AcpSession> {
const logger = deps.logger ?? acpLog
logger.info("Session", `Creating session ${sessionId} in ${cwd}`)
// Create ExtensionHost with ACP-specific configuration.
const hostOptions: ExtensionHostOptions = {
mode: options.mode,
@ -203,10 +228,8 @@ export class AcpSession implements IAcpSession {
ephemeral: true,
}
logger.debug("Session", "Creating ExtensionHost", hostOptions)
const extensionHost = new ExtensionHost(hostOptions)
await extensionHost.activate()
logger.info("Session", `ExtensionHost activated for session ${sessionId}`)
const session = new AcpSession(sessionId, extensionHost, connection, cwd, options.mode, deps)
session.setupEventHandlers()
@ -231,19 +254,35 @@ export class AcpSession implements IAcpSession {
*/
private resetForNewPrompt(): void {
this.eventHandler.reset()
this.updateBuffer.reset()
this.isCancelling = false
}
/**
* Handle task completion.
*/
private handleTaskCompleted(success: boolean): void {
// Flush any buffered updates before completing.
void this.updateBuffer.flush().then(() => {
// Complete the prompt using the state machine.
const stopReason = this.promptState.complete(success)
this.logger.debug("Session", `Resolving prompt with stopReason: ${stopReason}`)
})
// If we're cancelling, override the stop reason to "cancelled"
if (this.isCancelling) {
this.handleCancellationComplete()
} else {
// Normal completion
this.promptState.complete(success)
}
}
/**
* Handle cancellation completion.
* Called when the extension has finished cancelling (either via taskCompleted or NO_TASK transition).
*/
private handleCancellationComplete(): void {
if (!this.isCancelling) {
return // Already handled
}
this.isCancelling = false
// Directly transition to complete with "cancelled" stop reason
this.promptState.transitionToComplete("cancelled")
}
// ===========================================================================
@ -254,7 +293,31 @@ export class AcpSession implements IAcpSession {
* Process a prompt request from the ACP client.
*/
async prompt(params: PromptRequest): Promise<PromptResponse> {
this.logger.info("Session", `Processing prompt for session ${this.sessionId}`)
// Extract text and images from prompt.
const text = extractPromptText(params.prompt)
const images = extractPromptImages(params.prompt)
// Check if we're in a resumable state (paused after cancel).
// If so, resume the existing conversation instead of starting fresh.
const currentState = this.extensionHost.client.getAgentState()
if (currentState.state === AgentLoopState.RESUMABLE && currentState.currentAsk === "resume_task") {
this.logger.info(
"Session",
`RESUME TASK: resuming paused task with user input (was ask=${currentState.currentAsk})`,
)
// Reset state for the resumed prompt (but don't cancel - task is already paused)
this.eventHandler.reset()
this.isCancelling = false
// Start tracking the prompt
const promise = this.promptState.startPrompt(text)
// Resume the task with the user's message as follow-up
this.extensionHost.client.respond(text, images.length > 0 ? images : undefined)
return promise
}
// Cancel any pending prompt.
this.cancel()
@ -262,20 +325,12 @@ export class AcpSession implements IAcpSession {
// Reset state for new prompt.
this.resetForNewPrompt()
// Extract text and images from prompt.
const text = extractPromptText(params.prompt)
const images = extractPromptImages(params.prompt)
this.logger.debug("Session", `Prompt text (${text.length} chars), images: ${images.length}`)
// Start the prompt using the state machine.
const promise = this.promptState.startPrompt(text)
if (images.length > 0) {
this.logger.debug("Session", "Starting task with images")
this.extensionHost.sendToExtension({ type: "newTask", text, images })
} else {
this.logger.debug("Session", "Starting task (text only)")
this.extensionHost.sendToExtension({ type: "newTask", text })
}
@ -287,10 +342,19 @@ export class AcpSession implements IAcpSession {
*/
cancel(): void {
if (this.promptState.isProcessing()) {
this.logger.info("Session", "Cancelling pending prompt")
this.promptState.cancel()
this.logger.info("Session", "Sending cancelTask to extension")
// === TEST LOGGING: Cancel triggered ===
const currentState = this.extensionHost.client.getAgentState()
this.logger.info(
"Session",
`CANCEL TASK: sending cancelTask (state=${currentState.state}, running=${currentState.isRunning}, streaming=${currentState.isStreaming}, ask=${currentState.currentAsk || "none"})`,
)
this.isCancelling = true
// Content continues flowing to the client during cancellation so users
// see what the LLM was generating when cancel was triggered.
this.extensionHost.sendToExtension({ type: "cancelTask" })
// We wait for the extension to send a taskCompleted event or transition to NO_TASK
// which will trigger handleTaskCompleted -> promptState.transitionToComplete("cancelled")
}
}
@ -299,7 +363,6 @@ export class AcpSession implements IAcpSession {
* The mode change is tracked by the event handler which listens to extension state updates.
*/
setMode(mode: string): void {
this.logger.info("Session", `Setting mode to: ${mode}`)
this.extensionHost.sendToExtension({ type: "updateSettings", updatedSettings: { mode } })
}
@ -308,7 +371,6 @@ export class AcpSession implements IAcpSession {
* This updates the provider settings to use the specified model.
*/
setModel(modelId: string): void {
this.logger.info("Session", `Setting model to: ${modelId}`)
this.currentModelId = modelId
// Map model ID to extension settings
@ -347,16 +409,12 @@ export class AcpSession implements IAcpSession {
* Dispose of the session and release resources.
*/
async dispose(): Promise<void> {
this.logger.info("Session", `Disposing session ${this.sessionId}`)
this.cancel()
// Clean up event handler listeners
this.eventHandler.cleanup()
// Flush any remaining buffered updates.
await this.updateBuffer.flush()
await this.extensionHost.dispose()
this.logger.info("Session", `Session ${this.sessionId} disposed`)
}
// ===========================================================================
@ -364,32 +422,14 @@ export class AcpSession implements IAcpSession {
// ===========================================================================
/**
* Send an update to the ACP client through the buffer.
* Text chunks are batched, other updates are sent immediately.
*
* @returns Result indicating success or failure.
*/
private async sendUpdate(update: SessionNotification["update"]): Promise<Result<void>> {
try {
await this.updateBuffer.queueUpdate(update)
return ok(undefined)
} catch (error) {
const errorMessage = error instanceof Error ? error.message : String(error)
this.logger.error("Session", `Failed to queue update: ${errorMessage}`)
return err(`Failed to queue update: ${errorMessage}`)
}
}
/**
* Send an update directly to the ACP client (bypasses buffer).
* Used by the UpdateBuffer to actually send batched updates.
* Send an update directly to the ACP client.
*
* @returns Result indicating success or failure with error details.
*/
private async sendUpdateDirect(update: SessionNotification["update"]): Promise<Result<void>> {
try {
// Log the full update being sent to ACP connection
this.logger.debug("Session", `ACP OUT: ${JSON.stringify({ sessionId: this.sessionId, update })}`)
// Log the update being sent to ACP connection (commented out - too noisy)
// this.logger.info("Session", `OUT: ${JSON.stringify(update)}`)
await this.connection.sessionUpdate({ sessionId: this.sessionId, update })
return ok(undefined)
} catch (error) {

15
apps/cli/src/acp/tool-content-stream.ts
View file

@ -95,15 +95,9 @@ export class ToolContentStreamManager {
// Only stream content for file write operations (uses tool registry)
if (!isFileWriteTool(toolName)) {
this.logger.debug("ToolContentStream", `Skipping content streaming for non-file tool: ${toolName}`)
return true // Handled (by skipping)
}
this.logger.debug(
"ToolContentStream",
`handleToolContentStreaming: tool=${toolName}, path=${toolPath}, partial=${isPartial}, contentLen=${content.length}`,
)
// Check if we have valid path and content to start streaming
// Path must have a file extension to be considered valid (uses shared utility)
const validPath = hasValidFilePath(toolPath)
@ -123,7 +117,6 @@ export class ToolContentStreamManager {
*/
reset(): void {
this.toolContentHeadersSent.clear()
this.logger.debug("ToolContentStream", "Reset tool content stream state")
}
/**
@ -170,7 +163,6 @@ export class ToolContentStreamManager {
// perceived latency during the gap while LLM generates file content.
if (hasValidPath && !this.toolContentHeadersSent.has(ts)) {
this.toolContentHeadersSent.add(ts)
this.logger.debug("ToolContentStream", `Sending tool content header for ${toolPath}`)
this.sendUpdate({
sessionUpdate: "agent_message_chunk",
content: { type: "text", text: `\n**Creating ${toolPath}**\n\`\`\`\n` },
@ -184,7 +176,6 @@ export class ToolContentStreamManager {
const delta = this.deltaTracker.getDelta(deltaKey, content)
if (delta) {
this.logger.debug("ToolContentStream", `Streaming tool content delta: ${delta.length} chars`)
this.sendUpdate({
sessionUpdate: "agent_message_chunk",
content: { type: "text", text: delta },
@ -196,7 +187,7 @@ export class ToolContentStreamManager {
/**
* Handle a complete (non-partial) tool message.
*/
private handleCompleteMessage(ts: number, toolPath: string, content: string): void {
private handleCompleteMessage(ts: number, _toolPath: string, _content: string): void {
// Message complete - finish streaming and clean up
if (this.toolContentHeadersSent.has(ts)) {
// Send closing code fence
@ -209,9 +200,5 @@ export class ToolContentStreamManager {
// Note: The actual tool_call notification will be sent via handleWaitingForInput
// when the waitingForInput event fires (which happens when partial becomes false)
this.logger.debug(
"ToolContentStream",
`Tool content streaming complete for ${toolPath}: ${content.length} chars`,
)
}
}

13
apps/cli/src/acp/translator.ts
View file

@ -9,6 +9,7 @@
* - translator/prompt-extractor.ts: Prompt content extraction
* - translator/tool-parser.ts: Tool information parsing
* - translator/message-translator.ts: Main message translation
* - translator/plan-translator.ts: TodoItem to ACP PlanEntry translation
*
* Import from this file or directly from translator/index.ts
*/
@ -40,4 +41,16 @@ export {
createPermissionOptions,
// Backward compatibility
mapToolKind,
// Plan translation (TodoItem to ACP PlanEntry)
todoItemToPlanEntry,
todoListToPlanUpdate,
parseTodoListFromMessage,
isTodoListMessage,
extractTodoListFromMessage,
createPlanUpdateFromMessage,
type PlanEntry,
type PlanEntryPriority,
type PlanEntryStatus,
type PlanUpdate,
type PriorityConfig,
} from "./translator/index.js"

16
apps/cli/src/acp/translator/index.ts
View file

@ -10,6 +10,7 @@
* - prompt-extractor: Prompt content extraction
* - tool-parser: Tool information parsing
* - message-translator: Main message translation
* - plan-translator: TodoItem to ACP PlanEntry translation
*/
// Diff parsing
@ -41,3 +42,18 @@ export {
// Re-export mapToolKind for backward compatibility
// (now uses mapToolToKind from tool-registry internally)
export { mapToolToKind as mapToolKind } from "../tool-registry.js"
// Plan translation (TodoItem to ACP PlanEntry)
export {
todoItemToPlanEntry,
todoListToPlanUpdate,
parseTodoListFromMessage,
isTodoListMessage,
extractTodoListFromMessage,
createPlanUpdateFromMessage,
type PlanEntry,
type PlanEntryPriority,
type PlanEntryStatus,
type PlanUpdate,
type PriorityConfig,
} from "./plan-translator.js"

260
apps/cli/src/acp/translator/plan-translator.ts Normal file
View file

@ -0,0 +1,260 @@
/**
* Plan Translator
*
* Translates between Roo CLI TodoItem format and ACP PlanEntry format.
* This enables the agent to communicate execution plans to ACP clients
* when using the update_todo_list tool.
*
* @see https://agentclientprotocol.com/protocol/agent-plan
*/
import type { TodoItem } from "@roo-code/types"
// =============================================================================
// Types
// =============================================================================
/**
* Priority levels for plan entries.
* Maps to ACP PlanEntryPriority.
*/
export type PlanEntryPriority = "high" | "medium" | "low"
/**
* Status levels for plan entries.
* Maps to ACP PlanEntryStatus (same as TodoStatus).
*/
export type PlanEntryStatus = "pending" | "in_progress" | "completed"
/**
* A single entry in the execution plan.
* Represents a task or goal that the agent intends to accomplish.
*/
export interface PlanEntry {
/** Human-readable description of what this task aims to accomplish */
content: string
/** The relative importance of this task */
priority: PlanEntryPriority
/** Current execution status of this task */
status: PlanEntryStatus
}
/**
* ACP Plan session update payload.
*/
export interface PlanUpdate {
sessionUpdate: "plan"
entries: PlanEntry[]
}
/**
* Configuration for priority assignment when converting todos to plan entries.
*/
export interface PriorityConfig {
/** Default priority for all items (default: "medium") */
defaultPriority: PlanEntryPriority
/** Assign high priority to in_progress items (default: true) */
prioritizeInProgress: boolean
/** Assign higher priority to earlier items in the list (default: false) */
prioritizeByOrder: boolean
/** Number of top items to mark as high priority when prioritizeByOrder is true */
highPriorityCount: number
}
/**
* Default priority configuration.
*/
const DEFAULT_PRIORITY_CONFIG: PriorityConfig = {
defaultPriority: "medium",
prioritizeInProgress: true,
prioritizeByOrder: false,
highPriorityCount: 3,
}
// =============================================================================
// Priority Determination
// =============================================================================
/**
* Determine the priority of a todo item based on configuration.
*
* @param item - The todo item
* @param index - Position in the list (0-based)
* @param total - Total number of items
* @param config - Priority configuration
* @returns The determined priority
*/
function determinePriority(item: TodoItem, index: number, total: number, config: PriorityConfig): PlanEntryPriority {
// In-progress items get high priority
if (config.prioritizeInProgress && item.status === "in_progress") {
return "high"
}
// Order-based priority
if (config.prioritizeByOrder && total > 0) {
if (index < config.highPriorityCount) {
return "high"
}
if (index < Math.floor(total / 2)) {
return "medium"
}
return "low"
}
return config.defaultPriority
}
// =============================================================================
// Translation Functions
// =============================================================================
/**
* Translate a single TodoItem to a PlanEntry.
*
* @param item - The todo item to translate
* @param index - Position in the list (0-based)
* @param total - Total number of items
* @param config - Priority configuration
* @returns The translated plan entry
*/
export function todoItemToPlanEntry(
item: TodoItem,
index: number = 0,
total: number = 1,
config: PriorityConfig = DEFAULT_PRIORITY_CONFIG,
): PlanEntry {
return {
content: item.content,
priority: determinePriority(item, index, total, config),
status: item.status,
}
}
/**
* Translate an array of TodoItems to a PlanUpdate.
*
* @param todos - Array of todo items
* @param config - Optional partial priority configuration
* @returns The plan update payload
*/
export function todoListToPlanUpdate(todos: TodoItem[], config?: Partial<PriorityConfig>): PlanUpdate {
const mergedConfig: PriorityConfig = { ...DEFAULT_PRIORITY_CONFIG, ...config }
const total = todos.length
return {
sessionUpdate: "plan",
entries: todos.map((item, index) => todoItemToPlanEntry(item, index, total, mergedConfig)),
}
}
// =============================================================================
// Message Detection and Parsing
// =============================================================================
/**
* Parsed todo list message structure.
*/
interface ParsedTodoMessage {
tool: "updateTodoList"
todos: TodoItem[]
}
/**
* Type guard to check if parsed JSON is a valid todo list message.
*/
function isParsedTodoMessage(obj: unknown): obj is ParsedTodoMessage {
if (!obj || typeof obj !== "object") return false
const record = obj as Record<string, unknown>
return record.tool === "updateTodoList" && Array.isArray(record.todos)
}
/**
* Parse todo list from a tool message text.
*
* @param text - The message text (JSON string)
* @returns Array of TodoItems or null if not a valid todo message
*/
export function parseTodoListFromMessage(text: string): TodoItem[] | null {
try {
const parsed: unknown = JSON.parse(text)
if (isParsedTodoMessage(parsed)) {
return parsed.todos
}
} catch {
// Not valid JSON - ignore
}
return null
}
/**
* Minimal message interface for detection.
*/
interface MessageLike {
type: string
ask?: string
say?: string
text?: string
}
/**
* Check if a message contains a todo list update.
*
* Detects two types of messages:
* 1. Tool ask messages with updateTodoList
* 2. user_edit_todos say messages (when user edits the todo list)
*
* @param message - The message to check
* @returns true if message contains a todo list update
*/
export function isTodoListMessage(message: MessageLike): boolean {
// Check for tool ask message with updateTodoList
if (message.type === "ask" && message.ask === "tool" && message.text) {
const todos = parseTodoListFromMessage(message.text)
return todos !== null
}
// Check for user_edit_todos say message
if (message.type === "say" && message.say === "user_edit_todos" && message.text) {
const todos = parseTodoListFromMessage(message.text)
return todos !== null
}
return false
}
/**
* Extract todo list from a message if present.
*
* @param message - The message to extract from
* @returns Array of TodoItems or null if not a todo message
*/
export function extractTodoListFromMessage(message: MessageLike): TodoItem[] | null {
if (!message.text) return null
if (message.type === "ask" && message.ask === "tool") {
return parseTodoListFromMessage(message.text)
}
if (message.type === "say" && message.say === "user_edit_todos") {
return parseTodoListFromMessage(message.text)
}
return null
}
/**
* Create a plan update from a message if it contains a todo list.
*
* Convenience function that combines detection, extraction, and translation.
*
* @param message - The message to process
* @param config - Optional priority configuration
* @returns PlanUpdate or null if message doesn't contain todos
*/
export function createPlanUpdateFromMessage(message: MessageLike, config?: Partial<PriorityConfig>): PlanUpdate | null {
const todos = extractTodoListFromMessage(message)
if (!todos || todos.length === 0) {
return null
}
return todoListToPlanUpdate(todos, config)
}

4
apps/cli/src/acp/translator/tool-parser.ts
View file

@ -158,6 +158,10 @@ export function generateToolTitle(toolName: string, filePath?: string): string {
// Browser actions
browser_action: "Browser action",
browserAction: "Browser action",
// Plan updates
updateTodoList: "Update plan",
update_todo_list: "Update plan",
}
return toolTitles[toolName] || (fileName ? `${toolName}: ${fileName}` : toolName)

220
apps/cli/src/acp/update-buffer.ts
View file

@ -1,220 +0,0 @@
/**
* ACP Update Buffer
*
* Intelligently buffers session updates to reduce message frequency.
* Text chunks are batched based on size and time thresholds, while
* tool calls and other updates are passed through immediately.
*/
import type * as acp from "@agentclientprotocol/sdk"
import type { IAcpLogger } from "./interfaces.js"
import { NullLogger } from "./interfaces.js"
// =============================================================================
// Types (exported)
// =============================================================================
export type { UpdateBufferOptions }
interface UpdateBufferOptions {
/** Minimum characters to buffer before flushing (default: 200) */
minBufferSize?: number
/** Maximum time in ms before flushing (default: 500) */
flushDelayMs?: number
/** Logger instance (optional, defaults to NullLogger) */
logger?: IAcpLogger
}
type TextChunkUpdate = {
sessionUpdate: "agent_message_chunk" | "agent_thought_chunk"
content: { type: "text"; text: string }
}
type SessionUpdate = acp.SessionNotification["update"]
// Type guard for text chunk updates
function isTextChunkUpdate(update: SessionUpdate): update is TextChunkUpdate {
const u = update as TextChunkUpdate
return (
(u.sessionUpdate === "agent_message_chunk" || u.sessionUpdate === "agent_thought_chunk") &&
u.content?.type === "text"
)
}
// =============================================================================
// UpdateBuffer Class
// =============================================================================
/**
* Buffers session updates to reduce the number of messages sent to the client.
*
* Text chunks (agent_message_chunk, agent_thought_chunk) are batched together
* and flushed when either:
* - The buffer size reaches minBufferSize
* - The flush delay timer expires
* - flush() is called manually
*
* Tool calls and other updates are passed through immediately.
*/
export class UpdateBuffer {
private readonly minBufferSize: number
private readonly flushDelayMs: number
private readonly logger: IAcpLogger
/** Buffered text for agent_message_chunk */
private messageBuffer = ""
/** Buffered text for agent_thought_chunk */
private thoughtBuffer = ""
/** Timer for delayed flush */
private flushTimer: ReturnType<typeof setTimeout> | null = null
/** Callback to send updates */
private readonly sendUpdate: (update: SessionUpdate) => Promise<void>
/** Track if we have pending buffered content */
private hasPendingContent = false
constructor(sendUpdate: (update: SessionUpdate) => Promise<void>, options: UpdateBufferOptions = {}) {
this.minBufferSize = options.minBufferSize ?? 200
this.flushDelayMs = options.flushDelayMs ?? 500
this.logger = options.logger ?? new NullLogger()
this.sendUpdate = sendUpdate
}
// ===========================================================================
// Public API
// ===========================================================================
/**
* Queue an update for sending.
*
* Text chunks are buffered and batched. Other updates are sent immediately.
*/
async queueUpdate(update: SessionUpdate): Promise<void> {
if (isTextChunkUpdate(update)) {
this.bufferTextChunk(update)
} else {
// Flush any pending text before sending non-text update
// This ensures correct ordering
await this.flush()
await this.sendUpdate(update)
}
}
/**
* Flush all pending buffered content.
*
* Should be called when the session ends or when immediate delivery is needed.
*/
async flush(): Promise<void> {
this.clearFlushTimer()
if (!this.hasPendingContent) {
return
}
this.logger.debug(
"UpdateBuffer",
`Flushing buffers: message=${this.messageBuffer.length}, thought=${this.thoughtBuffer.length}`,
)
// Send buffered message content
if (this.messageBuffer.length > 0) {
await this.sendUpdate({
sessionUpdate: "agent_message_chunk",
content: { type: "text", text: this.messageBuffer },
})
this.messageBuffer = ""
}
// Send buffered thought content
if (this.thoughtBuffer.length > 0) {
await this.sendUpdate({
sessionUpdate: "agent_thought_chunk",
content: { type: "text", text: this.thoughtBuffer },
})
this.thoughtBuffer = ""
}
this.hasPendingContent = false
}
/**
* Reset the buffer state.
*
* Should be called when starting a new prompt.
*/
reset(): void {
this.clearFlushTimer()
this.messageBuffer = ""
this.thoughtBuffer = ""
this.hasPendingContent = false
this.logger.debug("UpdateBuffer", "Buffer reset")
}
/**
* Get current buffer sizes for debugging/testing.
*/
getBufferSizes(): { message: number; thought: number } {
return {
message: this.messageBuffer.length,
thought: this.thoughtBuffer.length,
}
}
// ===========================================================================
// Private Methods
// ===========================================================================
/**
* Buffer a text chunk update.
*/
private bufferTextChunk(update: TextChunkUpdate): void {
const text = update.content.text
if (update.sessionUpdate === "agent_message_chunk") {
this.messageBuffer += text
} else {
this.thoughtBuffer += text
}
this.hasPendingContent = true
// Check if we should flush based on size
const totalSize = this.messageBuffer.length + this.thoughtBuffer.length
if (totalSize >= this.minBufferSize) {
this.logger.debug(
"UpdateBuffer",
`Size threshold reached (${totalSize} >= ${this.minBufferSize}), flushing`,
)
void this.flush()
return
}
// Schedule delayed flush if not already scheduled
this.scheduleFlush()
}
/**
* Schedule a delayed flush.
*/
private scheduleFlush(): void {
if (this.flushTimer !== null) {
return // Already scheduled
}
this.flushTimer = setTimeout(() => {
this.flushTimer = null
this.logger.debug("UpdateBuffer", "Flush timer expired")
void this.flush()
}, this.flushDelayMs)
}
/**
* Clear the flush timer.
*/
private clearFlushTimer(): void {
if (this.flushTimer !== null) {
clearTimeout(this.flushTimer)
this.flushTimer = null
}
}
}

8
apps/cli/src/agent/extension-client.ts
View file

@ -39,6 +39,7 @@ import {
type ModeChangedEvent,
} from "./events.js"
import { AgentLoopState, type AgentStateInfo } from "./agent-state.js"
import { testLog } from "./test-logger.js"
// =============================================================================
// Extension Client Configuration
@ -429,6 +430,13 @@ export class ExtensionClient {
* Use this to interrupt a task that is currently processing.
*/
cancelTask(): void {
// === TEST LOGGING: Cancel triggered ===
const currentState = this.store.getAgentState()
testLog.info(
"ExtensionClient",
`CANCEL TASK: sending cancelTask (state=${currentState.state}, running=${currentState.isRunning}, streaming=${currentState.isStreaming}, ask=${currentState.currentAsk || "none"})`,
)
const message: WebviewMessage = {
type: "cancelTask",
}

47
apps/cli/src/agent/extension-host.ts
View file

@ -37,6 +37,7 @@ import { ExtensionClient } from "./extension-client.js"
import { OutputManager } from "./output-manager.js"
import { PromptManager } from "./prompt-manager.js"
import { AskDispatcher } from "./ask-dispatcher.js"
import { testLog } from "./test-logger.js"
// Pre-configured logger for CLI message activity debugging.
const cliLogger = new DebugLogger("CLI")
@ -246,8 +247,33 @@ export class ExtensionHost extends EventEmitter implements ExtensionHostInterfac
* The client emits events, managers handle them.
*/
private setupClientEventHandlers(): void {
// === TEST LOGGING: State changes (matches ACP session.ts logging) ===
this.client.on("stateChange", (event) => {
const prev = event.previousState
const curr = event.currentState
// Only log if something actually changed
const stateChanged =
prev.state !== curr.state ||
prev.isRunning !== curr.isRunning ||
prev.isStreaming !== curr.isStreaming ||
prev.currentAsk !== curr.currentAsk
if (stateChanged) {
testLog.info(
"ExtensionClient",
`STATE: ${prev.state} → ${curr.state} (running=${curr.isRunning}, streaming=${curr.isStreaming}, ask=${curr.currentAsk || "none"})`,
)
}
})
// Handle new messages - delegate to OutputManager.
this.client.on("message", (msg: ClineMessage) => {
// === TEST LOGGING: New messages ===
const msgType = msg.type === "say" ? `say:${msg.say}` : `ask:${msg.ask}`
const partial = msg.partial ? "PARTIAL" : "COMPLETE"
testLog.info("ExtensionClient", `MSG NEW: ${msgType} ${partial} ts=${msg.ts}`)
this.logMessageDebug(msg, "new")
// DEBUG: Log all incoming messages with timestamp (only when -d flag is set)
if (this.options.debug) {
@ -261,6 +287,11 @@ export class ExtensionHost extends EventEmitter implements ExtensionHostInterfac
// Handle message updates - delegate to OutputManager.
this.client.on("messageUpdated", (msg: ClineMessage) => {
// === TEST LOGGING: Message updates ===
const msgType = msg.type === "say" ? `say:${msg.say}` : `ask:${msg.ask}`
const partial = msg.partial ? "PARTIAL" : "COMPLETE"
testLog.info("ExtensionClient", `MSG UPDATE: ${msgType} ${partial} ts=${msg.ts}`)
this.logMessageDebug(msg, "updated")
// DEBUG: Log all message updates with timestamp (only when -d flag is set)
if (this.options.debug) {
@ -274,11 +305,16 @@ export class ExtensionHost extends EventEmitter implements ExtensionHostInterfac
// Handle waiting for input - delegate to AskDispatcher.
this.client.on("waitingForInput", (event: WaitingForInputEvent) => {
// === TEST LOGGING: Waiting for input ===
testLog.info("ExtensionClient", `WAITING FOR INPUT: ask=${event.ask}`)
this.askDispatcher.handleAsk(event.message)
})
// Handle task completion.
this.client.on("taskCompleted", (event: TaskCompletedEvent) => {
// === TEST LOGGING: Task completed ===
testLog.info("ExtensionClient", `TASK COMPLETED: success=${event.success}`)
// Output completion message via OutputManager.
// Note: completion_result is an "ask" type, not a "say" type.
if (event.message && event.message.type === "ask" && event.message.ask === "completion_result") {
@ -455,6 +491,17 @@ export class ExtensionHost extends EventEmitter implements ExtensionHostInterfac
throw new Error("You cannot send messages to the extension before it is ready")
}
// === TEST LOGGING: Track outgoing messages to extension (especially cancelTask) ===
if (message.type === "cancelTask") {
const currentState = this.client.getAgentState()
testLog.info(
"ExtensionHost",
`SEND TO EXT: cancelTask (state=${currentState.state}, running=${currentState.isRunning}, streaming=${currentState.isStreaming}, ask=${currentState.currentAsk || "none"})`,
)
} else if (message.type === "askResponse") {
testLog.info("ExtensionHost", `SEND TO EXT: askResponse (response=${message.askResponse})`)
}
this.emit("webviewMessage", message)
}

15
apps/cli/src/agent/message-processor.ts
View file

@ -105,9 +105,6 @@ export class MessageProcessor {
* @param message - The raw message from the extension
*/
processMessage(message: ExtensionMessage): void {
// Debug logging for ALL messages to trace flow (always enabled for debugging)
console.error(`[MessageProcessor-DEBUG] processMessage: type=${message.type}`)
if (this.options.debug) {
debugLog("[MessageProcessor] Received message", { type: message.type })
}
@ -251,12 +248,6 @@ export class MessageProcessor {
const clineMessage = message.clineMessage
// Debug logging for messageUpdated
const msgType = clineMessage.type === "ask" ? `ask:${clineMessage.ask}` : `say:${clineMessage.say}`
console.error(
`[MessageProcessor-DEBUG] handleMessageUpdated: ${msgType}, ts=${clineMessage.ts}, partial=${clineMessage.partial}, textLen=${clineMessage.text?.length || 0}`,
)
const previousState = this.store.getAgentState()
// Update the message in the store
@ -431,12 +422,6 @@ export class MessageProcessor {
// A more sophisticated implementation would track seen message timestamps
const lastMessage = messages[messages.length - 1]
if (lastMessage) {
// Debug logging for emitted messages
const msgType = lastMessage.type === "ask" ? `ask:${lastMessage.ask}` : `say:${lastMessage.say}`
console.error(
`[MessageProcessor-DEBUG] emitNewMessageEvents (last of ${messages.length}): ${msgType}, ts=${lastMessage.ts}, partial=${lastMessage.partial}, textLen=${lastMessage.text?.length || 0}`,
)
// DEBUG: Log all emitted ask messages to trace partial handling
if (this.options.debug && lastMessage.type === "ask") {
debugLog("[MessageProcessor] EMIT message", {

115
apps/cli/src/agent/test-logger.ts Normal file
View file

@ -0,0 +1,115 @@
/**
* Test Logger for CLI/ACP Cancellation Debugging
*
* This writes logs to ~/.roo/cli-acp-test.log for comparing CLI
* behavior with ACP during cancellation testing.
*
* Format matches ACP logger for easy side-by-side comparison.
*/
import * as fs from "node:fs"
import * as path from "node:path"
import * as os from "node:os"
const LOG_DIR = path.join(os.homedir(), ".roo")
const LOG_FILE = path.join(LOG_DIR, "cli-acp-test.log")
let stream: fs.WriteStream | null = null
/**
* Ensure log file and directory exist.
*/
function ensureLogFile(): void {
try {
if (!fs.existsSync(LOG_DIR)) {
fs.mkdirSync(LOG_DIR, { recursive: true })
}
if (!stream) {
stream = fs.createWriteStream(LOG_FILE, { flags: "a" })
}
} catch {
// Silently fail
}
}
/**
* Format and write a log entry.
*/
function write(level: string, component: string, message: string, data?: unknown): void {
ensureLogFile()
if (!stream) return
const timestamp = new Date().toISOString()
let formatted = `[${timestamp}] [${level}] [${component}] ${message}`
if (data !== undefined) {
try {
const dataStr = JSON.stringify(data, null, 2)
formatted += `\n${dataStr}`
} catch {
formatted += ` [Data: unserializable]`
}
}
stream.write(formatted + "\n")
}
/**
* Test logger for CLI cancellation debugging.
*
* Usage:
* testLog.info("ExtensionClient", "STATE: idle → running (running=true, streaming=true, ask=none)")
* testLog.info("Session", "CANCEL: triggered")
*/
export const testLog = {
info(component: string, message: string, data?: unknown): void {
write("INFO", component, message, data)
},
debug(component: string, message: string, data?: unknown): void {
write("DEBUG", component, message, data)
},
warn(component: string, message: string, data?: unknown): void {
write("WARN", component, message, data)
},
error(component: string, message: string, data?: unknown): void {
write("ERROR", component, message, data)
},
/**
* Clear the log file (call at start of test session).
*/
clear(): void {
try {
if (stream) {
stream.end()
stream = null
}
fs.writeFileSync(LOG_FILE, "")
} catch {
// Silently fail
}
},
/**
* Get the log file path.
*/
getLogPath(): string {
return LOG_FILE
},
/**
* Close the logger.
*/
close(): void {
if (stream) {
stream.end()
stream = null
}
},
}
// Log startup
testLog.info("TestLogger", `CLI test logging initialized. Log file: ${LOG_FILE}`)