refactor(chat): share reactor event-subscriber helper across sinks

Extract the duplicated tasks-changed/tools-activate subscriber blocks into handleChatDirectorEvent, log invalid payloads at debug level instead of dropping them silently, and clear queued task-change notifications when a turn throws.

Author: TheGreatAxios

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<string, unknown> }[] = [];
+  return {
+    calls,
+    log: (message: string, fields?: Record<string, unknown>): 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([]);
+  });
+});

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<string, unknown>,
+) => 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;
+}

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([]);
+  });
 });

src/agent/director.ts

@@ -682,15 +682,26 @@ class ChatDirectorImpl extends DefaultDirector {
     state: ReactorState,
     capabilities: ReactorCapabilities,
   ): Promise<ReactorAction | ReactorAction[]> {
-    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(

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<ExecResult> {
       // 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;

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)) {