Skip to content

Commit 91dd0a9

Browse files
committed
feat(chat): replace onTasksChange/onActivateTools with reactor events
Hosts subscribe to custom.chat.tasks.changed and custom.chat.tools.activate on the agent stream; TUI repaints the task panel, exec activates tools. Fixes CL-7916
1 parent 50a6b16 commit 91dd0a9

15 files changed

Lines changed: 190 additions & 170 deletions

‎src/agent/director.test.ts‎

Lines changed: 0 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -135,7 +135,6 @@ describe("ChatDirector tool-only loop protection", () => {
135135

136136
test("nudges once at the family threshold, after pending tools execute", async () => {
137137
const director = createChatDirector("system", [], {
138-
onTasksChange: () => undefined,
139138
provider: providerlessPolicy,
140139
});
141140
const capabilities = makeCapabilities();
@@ -149,7 +148,6 @@ describe("ChatDirector tool-only loop protection", () => {
149148

150149
test("the nudge is one-shot — it does not repeat on the next tool-only turn", async () => {
151150
const director = createChatDirector("system", [], {
152-
onTasksChange: () => undefined,
153151
provider: providerlessPolicy,
154152
});
155153
const capabilities = makeCapabilities();
@@ -168,7 +166,6 @@ describe("ChatDirector tool-only loop protection", () => {
168166
// well past any prior hard-pause threshold without ever pausing.
169167
test("a long productive tool-only streak continues without pausing", async () => {
170168
const director = createChatDirector("system", [], {
171-
onTasksChange: () => undefined,
172169
provider: providerlessPolicy,
173170
});
174171
const capabilities = makeCapabilities();
@@ -210,7 +207,6 @@ describe("ChatDirector inference-error recovery (CL-6910)", () => {
210207
"does not re-issue inference for a %s error already exhausted by the harness",
211208
async (category) => {
212209
const director = createChatDirector("system", [], {
213-
onTasksChange: () => undefined,
214210
provider: providerlessPolicy,
215211
});
216212
const capabilities = makeCapabilities();
@@ -232,7 +228,6 @@ describe("ChatDirector inference-error recovery (CL-6910)", () => {
232228

233229
test("still recovers on internal-recovery abort, bounded by MAX_INFERENCE_RECOVERIES", async () => {
234230
const director = createChatDirector("system", [], {
235-
onTasksChange: () => undefined,
236231
provider: providerlessPolicy,
237232
});
238233
const capabilities = makeCapabilities();
@@ -262,7 +257,6 @@ describe("ChatDirector inference-error recovery (CL-6910)", () => {
262257

263258
test("an unrelated aborted error (not internal-recovery) is not recovered by the director", async () => {
264259
const director = createChatDirector("system", [], {
265-
onTasksChange: () => undefined,
266260
provider: providerlessPolicy,
267261
});
268262
const capabilities = makeCapabilities();
@@ -279,7 +273,6 @@ describe("ChatDirector inference-error recovery (CL-6910)", () => {
279273

280274
test("inference-recovery budget resets at the next turn boundary", async () => {
281275
const director = createChatDirector("system", [], {
282-
onTasksChange: () => undefined,
283276
provider: providerlessPolicy,
284277
});
285278
const capabilities = makeCapabilities();
@@ -323,7 +316,6 @@ describe("ChatDirector inference-error recovery (CL-6910)", () => {
323316
// bounded, not open-ended, and never reaches 9.
324317
test("worst case: director-owned recovery path issues at most 1 + MAX_INFERENCE_RECOVERIES infer calls", async () => {
325318
const director = createChatDirector("system", [], {
326-
onTasksChange: () => undefined,
327319
provider: providerlessPolicy,
328320
});
329321
const capabilities = makeCapabilities();
@@ -344,7 +336,6 @@ describe("ChatDirector inference-error recovery (CL-6910)", () => {
344336

345337
test("timeout category produces the timeout preamble, not the fatal fallback", async () => {
346338
const director = createChatDirector("system", [], {
347-
onTasksChange: () => undefined,
348339
provider: providerlessPolicy,
349340
});
350341
const capabilities = makeCapabilities();

‎src/agent/director.ts‎

Lines changed: 38 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@ import {
3030
applyManageTasks,
3131
hasActiveTasks,
3232
parseManageTasksArgs,
33+
TaskSchema,
3334
type Task,
3435
} from "./tasks.js";
3536
import { createCorbitsRetryPolicy } from "./retry-policy.js";
@@ -412,7 +413,7 @@ function isCodeFile(path: string): boolean {
412413
// Returns null when the call is not manage_tasks or its arguments don't
413414
// parse, so callers can distinguish "no valid manage_tasks call here" from
414415
// "a valid call that happened to be a no-op" — the latter still counts as an
415-
// update for onTasksChange purposes.
416+
// update for tasks-changed event purposes.
416417
function applyManageTasksToolCall(
417418
tasks: Task[],
418419
block: { name: string; arguments: unknown },
@@ -422,15 +423,27 @@ function applyManageTasksToolCall(
422423
return taskArgs !== null ? applyManageTasks(tasks, taskArgs) : null;
423424
}
424425

426+
// Reactor events the chat director emits in place of host closures. Hosts
427+
// (TUI, exec) subscribe on the agent stream: task-list changes replace the
428+
// former onTasksChange callback, tool activation replaces onActivateTools.
429+
// Neither namespace collides with the reactor's reserved prefixes
430+
// (inference., tool., reactor., fork.).
431+
export const CHAT_TASKS_CHANGED_EVENT = "custom.chat.tasks.changed";
432+
export const CHAT_TOOLS_ACTIVATE_EVENT = "custom.chat.tools.activate";
433+
export const ChatTasksChangedDataSchema = type({
434+
tasks: TaskSchema.array(),
435+
});
436+
export const ChatToolsActivateDataSchema = type({
437+
names: "string[]",
438+
});
439+
425440
export interface ChatDirectorOptions {
426441
taskClassifier?:
427442
| ((message: string, metadata: SessionMetadata) => Promise<TaskBoundary>)
428443
| undefined;
429-
onActivateTools?: ((names: string[]) => void) | undefined;
430444
inactivityTimeoutMs?: number | undefined;
431445
totalTimeoutMs?: number | undefined;
432446
workflowCoordinator?: WorkflowCoordinator | undefined;
433-
onTasksChange: (tasks: Task[]) => void;
434447
requestContinuation?: (() => void) | undefined;
435448
provider?: { providerName: string; model?: string } | undefined;
436449
/**
@@ -464,7 +477,6 @@ class ChatDirectorImpl extends DefaultDirector {
464477
>();
465478
private readonly lspTriggerCalls = new Set<string>();
466479
private readonly askOperatorCalls = new Set<string>();
467-
private readonly onActivateTools: ((names: string[]) => void) | undefined;
468480
private readonly taskClassifier:
469481
| ((message: string, metadata: SessionMetadata) => Promise<TaskBoundary>)
470482
| undefined;
@@ -481,7 +493,6 @@ class ChatDirectorImpl extends DefaultDirector {
481493
private lastInferenceTurnHadContent = false;
482494
private operatorJustResponded = false;
483495
private tasks: Task[] = [];
484-
private readonly onTasksChange: ((tasks: Task[]) => void) | undefined;
485496
private turnCount = 0;
486497
private currentTaskLabel: string | undefined;
487498
private lastTaskSummary: string | undefined;
@@ -500,6 +511,11 @@ class ChatDirectorImpl extends DefaultDirector {
500511
private toolOnlyStreak = 0;
501512
private toolOnlyNudgeFired = false;
502513
private pendingToolOnlyNudge = false;
514+
// Reactor events queued while scanning the current inbound event. Drained
515+
// in decide() and appended to whatever the turn returns, so task/tool
516+
// notifications ride along with every terminal action list (emit is
517+
// composable with all other actions).
518+
private pendingEmits: ReactorAction[] = [];
503519

504520
constructor(
505521
systemPrompt: string,
@@ -527,9 +543,7 @@ class ChatDirectorImpl extends DefaultDirector {
527543
this.inactivityTimeoutMs = options.inactivityTimeoutMs;
528544
this.totalTimeoutMs = options.totalTimeoutMs;
529545
this.taskClassifier = options.taskClassifier;
530-
this.onActivateTools = options.onActivateTools;
531546
this.workflowCoordinator = options.workflowCoordinator;
532-
this.onTasksChange = options.onTasksChange;
533547
this.compaction = createCompactionGovernor(
534548
options.requestContinuation,
535549
composedPrompt,
@@ -559,10 +573,11 @@ class ChatDirectorImpl extends DefaultDirector {
559573
// A resumed session's task list lives in the transcript, not in the freshly
560574
// constructed director. Without this the chrome panel would read an empty
561575
// list until the model happened to call manage_tasks again, disagreeing
562-
// with the task block already painted in the transcript.
576+
// with the task block already painted in the transcript. The host emits
577+
// the tasks-changed reactor event after calling this (the director cannot
578+
// emit outside decide()).
563579
restoreTasks(tasks: Task[]): void {
564580
this.tasks = [...tasks];
565-
this.onTasksChange?.(this.tasks);
566581
}
567582

568583
// The status bar's context meter falls back to this when a provider omits
@@ -671,7 +686,11 @@ class ChatDirectorImpl extends DefaultDirector {
671686
await this.decideInner(event, state, capabilities),
672687
capabilities,
673688
);
674-
return this.withCurrentTools(settled);
689+
const withTools = this.withCurrentTools(settled);
690+
if (this.pendingEmits.length === 0) return withTools;
691+
const emits = this.pendingEmits;
692+
this.pendingEmits = [];
693+
return [...(Array.isArray(withTools) ? withTools : [withTools]), ...emits];
675694
}
676695

677696
private async decideInner(
@@ -867,7 +886,11 @@ class ChatDirectorImpl extends DefaultDirector {
867886
const next = applyManageTasksToolCall(this.tasks, block);
868887
if (next !== null) {
869888
this.tasks = next;
870-
this.onTasksChange?.(this.tasks);
889+
this.pendingEmits.push(
890+
capabilities.emit(CHAT_TASKS_CHANGED_EVENT, {
891+
tasks: this.tasks,
892+
}),
893+
);
871894
}
872895
} else if (block.name === "read_file" || block.name === "edit_file") {
873896
const pathResult = PathArgSchema(block.arguments);
@@ -915,7 +938,10 @@ class ChatDirectorImpl extends DefaultDirector {
915938
this.lspTriggerCalls.has(event.result.callId)
916939
) {
917940
this.lspTriggerCalls.delete(event.result.callId);
918-
if (!event.result.isError) this.onActivateTools?.(["lsp"]);
941+
if (!event.result.isError)
942+
this.pendingEmits.push(
943+
capabilities.emit(CHAT_TOOLS_ACTIVATE_EVENT, { names: ["lsp"] }),
944+
);
919945
}
920946

921947
if (event.type === "tool.done") {

0 commit comments

Comments
 (0)