diff --git a/src/agent/chat-event-subscribers.test.ts b/src/agent/chat-event-subscribers.test.ts new file mode 100644 index 000000000..96aa88213 --- /dev/null +++ b/src/agent/chat-event-subscribers.test.ts @@ -0,0 +1,122 @@ +import { describe, expect, test } from "bun:test"; +import { + CHAT_TASKS_CHANGED_EVENT, + CHAT_TOOLS_ACTIVATE_EVENT, +} from "./director.js"; +import { handleChatDirectorEvent } from "./chat-event-subscribers.js"; +import type { Task } from "./tasks.js"; + +function makeLog() { + const calls: { message: string; fields?: Record }[] = []; + return { + calls, + log: (message: string, fields?: Record): void => { + calls.push(fields !== undefined ? { message, fields } : { message }); + }, + }; +} + +describe("handleChatDirectorEvent", () => { + test("dispatches a valid tasks-changed payload without logging", () => { + const seen: Task[][] = []; + const log = makeLog(); + const handled = handleChatDirectorEvent( + { + type: CHAT_TASKS_CHANGED_EVENT, + data: { tasks: [{ id: "t1", title: "work", status: "doing" }] }, + }, + { + onTasksChanged: (tasks) => seen.push(tasks), + onToolsActivate: () => { + throw new Error("unexpected tools-activate dispatch"); + }, + }, + log.log, + ); + expect(handled).toBe(true); + expect(seen).toEqual([[{ id: "t1", title: "work", status: "doing" }]]); + expect(log.calls).toEqual([]); + }); + + test("dispatches a valid tools-activate payload without logging", () => { + const seen: string[][] = []; + const log = makeLog(); + const handled = handleChatDirectorEvent( + { type: CHAT_TOOLS_ACTIVATE_EVENT, data: { names: ["lsp"] } }, + { + onTasksChanged: () => { + throw new Error("unexpected tasks-changed dispatch"); + }, + onToolsActivate: (names) => seen.push([...names]), + }, + log.log, + ); + expect(handled).toBe(true); + expect(seen).toEqual([["lsp"]]); + expect(log.calls).toEqual([]); + }); + + test("drops an invalid tasks payload with a debug log naming the failure", () => { + let dispatched = false; + const log = makeLog(); + const handled = handleChatDirectorEvent( + { type: CHAT_TASKS_CHANGED_EVENT, data: { tasks: "not-a-list" } }, + { + onTasksChanged: () => { + dispatched = true; + }, + onToolsActivate: () => { + dispatched = true; + }, + }, + log.log, + ); + expect(handled).toBe(true); + expect(dispatched).toBe(false); + expect(log.calls).toHaveLength(1); + expect(log.calls[0]?.message).toMatch(/tasks-changed/); + expect(typeof log.calls[0]?.fields?.["error"]).toBe("string"); + }); + + test("drops an invalid tools payload with a debug log naming the failure", () => { + let dispatched = false; + const log = makeLog(); + const handled = handleChatDirectorEvent( + { type: CHAT_TOOLS_ACTIVATE_EVENT, data: { names: [42] } }, + { + onTasksChanged: () => { + dispatched = true; + }, + onToolsActivate: () => { + dispatched = true; + }, + }, + log.log, + ); + expect(handled).toBe(true); + expect(dispatched).toBe(false); + expect(log.calls).toHaveLength(1); + expect(log.calls[0]?.message).toMatch(/tools-activate/); + expect(typeof log.calls[0]?.fields?.["error"]).toBe("string"); + }); + + test("ignores unrelated events without logging or dispatching", () => { + let dispatched = false; + const log = makeLog(); + const handled = handleChatDirectorEvent( + { type: "inference.done", data: {} }, + { + onTasksChanged: () => { + dispatched = true; + }, + onToolsActivate: () => { + dispatched = true; + }, + }, + log.log, + ); + expect(handled).toBe(false); + expect(dispatched).toBe(false); + expect(log.calls).toEqual([]); + }); +}); diff --git a/src/agent/chat-event-subscribers.ts b/src/agent/chat-event-subscribers.ts new file mode 100644 index 000000000..ffc8a98d0 --- /dev/null +++ b/src/agent/chat-event-subscribers.ts @@ -0,0 +1,62 @@ +/** + * Shared subscriber for the chat-director reactor events. The TUI and exec + * stream sinks both listen for the task-list and tool-activation events the + * chat director emits in place of the former host closures; the parse, + * validation, and invalid-payload handling live here so the two sinks cannot + * drift apart. + */ + +import { type } from "arktype"; +import { + CHAT_TASKS_CHANGED_EVENT, + CHAT_TOOLS_ACTIVATE_EVENT, + ChatTasksChangedDataSchema, + ChatToolsActivateDataSchema, +} from "./director.js"; +import type { Task } from "./tasks.js"; + +export interface ChatDirectorEventHandlers { + onTasksChanged: (tasks: Task[]) => void; + onToolsActivate: (names: string[]) => void; +} + +export type ChatDirectorEventDebugLog = ( + message: string, + fields?: Record, +) => void; + +/** + * Dispatch one stream event to the chat-director handlers. Returns true when + * the event is a chat-director event (valid or not) so sinks can fall through + * to their own handling otherwise. Invalid payloads are dropped after a + * debug-level log naming the failure — never silently. + */ +export function handleChatDirectorEvent( + event: { type: string; data: unknown }, + handlers: ChatDirectorEventHandlers, + logDebug: ChatDirectorEventDebugLog, +): boolean { + if (event.type === CHAT_TASKS_CHANGED_EVENT) { + const parsed = ChatTasksChangedDataSchema(event.data); + if (parsed instanceof type.errors) { + logDebug("chat tasks-changed event dropped invalid payload: {error}", { + error: parsed.summary, + }); + return true; + } + handlers.onTasksChanged(parsed.tasks); + return true; + } + if (event.type === CHAT_TOOLS_ACTIVATE_EVENT) { + const parsed = ChatToolsActivateDataSchema(event.data); + if (parsed instanceof type.errors) { + logDebug("chat tools-activate event dropped invalid payload: {error}", { + error: parsed.summary, + }); + return true; + } + handlers.onToolsActivate(parsed.names); + return true; + } + return false; +} diff --git a/src/agent/director.test.ts b/src/agent/director.test.ts index f634e1169..65b201b1d 100644 --- a/src/agent/director.test.ts +++ b/src/agent/director.test.ts @@ -5,7 +5,12 @@ import type { ReactorInboundEvent, ReactorState, } from "@intx/types/runtime"; -import { createChatDirector, toolSetDigest } from "./director.js"; +import { + CHAT_TASKS_CHANGED_EVENT, + createChatDirector, + toolSetDigest, +} from "./director.js"; +import type { WorkflowCoordinator } from "../workflows/coordinator.js"; const mockState: ReactorState = { turns: [] } as unknown as ReactorState; @@ -356,4 +361,70 @@ describe("ChatDirector inference-error recovery (CL-6910)", () => { "unrecoverable inference error", ); }); + + // A turn that throws after queueing task-change notifications must drop the + // queue instead of flushing it stale on the next turn. + test("a throwing turn drops queued task-change notifications", async () => { + const throwingCoordinator = { + isActive: () => true, + currentStepIsGate: () => true, + currentStepId: () => null, + directive: () => { + throw new Error("tool-listing exploded"); + }, + handleToolDone: () => false, + } as unknown as WorkflowCoordinator; + const director = createChatDirector("system", [], { + workflowCoordinator: throwingCoordinator, + }); + const capabilities = makeCapabilities(); + + const manageTasksTurn = { + type: "inference.done", + turn: { + role: "assistant", + model: "test", + timestamp: 0, + content: [ + { + type: "tool_call", + id: "manage-tasks", + name: "manage_tasks", + arguments: { + action: "create", + tasks: [{ id: "t1", title: "work", status: "doing" }], + }, + }, + ], + }, + usage: { input: 0, output: 0 }, + source: "test", + } as unknown as ReactorInboundEvent; + + await expect( + director.decide(manageTasksTurn, mockState, capabilities), + ).rejects.toThrow("tool-listing exploded"); + + director.setWorkflowCoordinator(undefined); + const textTurn = { + type: "inference.done", + turn: { + role: "assistant", + model: "test", + timestamp: 0, + content: [{ type: "text", text: "all set" }], + }, + usage: { input: 0, output: 0 }, + source: "test", + } as unknown as ReactorInboundEvent; + const actions = actionsArray( + await director.decide(textTurn, mockState, capabilities), + ); + const stale = actions.filter( + (a) => + a.type === "emit" && + (a as { eventType?: string }).eventType === CHAT_TASKS_CHANGED_EVENT, + ); + expect(stale).toEqual([]); + }); }); diff --git a/src/agent/director.ts b/src/agent/director.ts index e0a50d8b0..7f3a5d1a3 100644 --- a/src/agent/director.ts +++ b/src/agent/director.ts @@ -682,15 +682,26 @@ class ChatDirectorImpl extends DefaultDirector { state: ReactorState, capabilities: ReactorCapabilities, ): Promise { - const settled = ensureCycleSettlesWithReply( - await this.decideInner(event, state, capabilities), - capabilities, - ); - const withTools = this.withCurrentTools(settled); - if (this.pendingEmits.length === 0) return withTools; - const emits = this.pendingEmits; - this.pendingEmits = []; - return [...(Array.isArray(withTools) ? withTools : [withTools]), ...emits]; + try { + const settled = ensureCycleSettlesWithReply( + await this.decideInner(event, state, capabilities), + capabilities, + ); + const withTools = this.withCurrentTools(settled); + if (this.pendingEmits.length === 0) return withTools; + const emits = this.pendingEmits; + this.pendingEmits = []; + return [ + ...(Array.isArray(withTools) ? withTools : [withTools]), + ...emits, + ]; + } catch (err) { + // A failed turn must not leak its queued task/tool notifications into + // the next turn — drop them so the next turn starts clean instead of + // flushing stale updates. + this.pendingEmits = []; + throw err; + } } private async decideInner( diff --git a/src/exec/runner.ts b/src/exec/runner.ts index a8621b05a..1196a1594 100644 --- a/src/exec/runner.ts +++ b/src/exec/runner.ts @@ -21,14 +21,8 @@ import { import { formatDirectorSystemPrompt } from "../agent/directors/identity.js"; import { DIRECTOR_REGISTRY } from "../agent/directors/registry.js"; import type { DirectorId, DirectorPackage } from "../agent/directors/types.js"; -import { - CHAT_TASKS_CHANGED_EVENT, - CHAT_TOOLS_ACTIVATE_EVENT, - ChatTasksChangedDataSchema, - ChatToolsActivateDataSchema, - submitOutputDefinition, -} from "../agent/director.js"; -import { type } from "arktype"; +import { submitOutputDefinition } from "../agent/director.js"; +import { handleChatDirectorEvent } from "../agent/chat-event-subscribers.js"; import { shellDefinition, updatePlanDefinition, @@ -976,18 +970,18 @@ export async function runExec(config: Config): Promise { // stdout output today (unlike the TUI's chrome zone) — debug logging // is the closest match to how this mode already surfaces other // in-session state changes. - if (event.type === CHAT_TASKS_CHANGED_EVENT) { - const parsed = ChatTasksChangedDataSchema(event.data); - if (!(parsed instanceof type.errors)) { - logger.debug("tasks updated: {tasks}", { - tasks: parsed.tasks.map((t) => `${t.status}:${t.title}`).join(", "), - }); - } - } else if (event.type === CHAT_TOOLS_ACTIVATE_EVENT) { - const parsed = ChatToolsActivateDataSchema(event.data); - if (!(parsed instanceof type.errors)) - activatedToolNames.activate(parsed.names); - } + handleChatDirectorEvent( + event, + { + onTasksChanged: (tasks) => { + logger.debug("tasks updated: {tasks}", { + tasks: tasks.map((t) => `${t.status}:${t.title}`).join(", "), + }); + }, + onToolsActivate: (names) => activatedToolNames.activate(names), + }, + (message, fields) => logger.debug(message, fields), + ); if (event.type === "inference.start" || event.type === "inference.done") { providerFailureObserved = false; providerError = undefined; diff --git a/src/tui/runner/exit.ts b/src/tui/runner/exit.ts index f93fa5dc4..8bbb833b4 100644 --- a/src/tui/runner/exit.ts +++ b/src/tui/runner/exit.ts @@ -58,13 +58,7 @@ import { type SnapshotKind, type SnapshotStatus, } from "./state.js"; -import { type } from "arktype"; -import { - CHAT_TASKS_CHANGED_EVENT, - CHAT_TOOLS_ACTIVATE_EVENT, - ChatTasksChangedDataSchema, - ChatToolsActivateDataSchema, -} from "../../agent/director.js"; +import { handleChatDirectorEvent } from "../../agent/chat-event-subscribers.js"; const tuiLogger = getLogger([LOG_NAMESPACE_ROOT, "tui"]); @@ -309,15 +303,14 @@ export async function createRunLifecycle( // Chat-director reactor events (replacing the former onTasksChange / // onActivateTools closures): task-list changes repaint the chrome panel, // tool activation opens the call gate for the named tools. - if (event.type === CHAT_TASKS_CHANGED_EVENT) { - const parsed = ChatTasksChangedDataSchema(event.data); - if (!(parsed instanceof type.errors)) - services.emitter.emit("tasks", parsed.tasks); - } else if (event.type === CHAT_TOOLS_ACTIVATE_EVENT) { - const parsed = ChatToolsActivateDataSchema(event.data); - if (!(parsed instanceof type.errors)) - services.activatedToolNames.activate(parsed.names); - } + handleChatDirectorEvent( + event, + { + onTasksChanged: (tasks) => services.emitter.emit("tasks", tasks), + onToolsActivate: (names) => services.activatedToolNames.activate(names), + }, + (message, fields) => tuiLogger.debug(message, fields), + ); services.runSink.sink(eventForSink); services.cycleRecorder.handleEvent(event); if (onTurnBoundary(event)) {