authorgravatar for 152941963+Svyk@users.noreply.github.comSvyk <152941963+Svyk@users.noreply.github.com> 2026-09-12 21:29:23-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-13 01:29:23-03:00
log3138f5716098a331f9a7d4cfc1bcd07118967a83
treef33abec8ddc4e60fe6c4f0b3d60bdf48de20caf6
parent0c5771d60a8ef2db34dfbbc142f7524badc829e0
signature Signed by PGP key B5690EEEBB952194

fix(grok): emit task lifecycle for monitors and background shells (#9139)

Co-authored-by: maria-rcks <254055478+maria-rcks@users.noreply.github.com>

5 files changed, 549 insertions(+), 2 deletions(-)

apps/server/scripts/acp-mock-agent.ts+104
......@@ -30,6 +30,8 @@ const emitXAiAskUserQuestionThenHang =
3030const emitContentThenHang = process.env.T3_ACP_EMIT_CONTENT_THEN_HANG === "1";
3131const emitPlanThenHang = process.env.T3_ACP_EMIT_PLAN_THEN_HANG === "1";
3232const emitActiveToolThenHang = process.env.T3_ACP_EMIT_ACTIVE_TOOL_THEN_HANG === "1";
33const emitGrokMonitorPostTurnPoll = process.env.T3_ACP_EMIT_GROK_MONITOR_POST_TURN_POLL === "1";
34const emitGrokBackgroundTaskStarted = process.env.T3_ACP_EMIT_GROK_BACKGROUND_TASK_STARTED === "1";
3335const emitForeignSessionUpdates = process.env.T3_ACP_EMIT_FOREIGN_SESSION_UPDATES === "1";
3436const waitForResumeRelease = process.env.T3_ACP_WAIT_FOR_RESUME_RELEASE === "1";
3537const completeFirstPromptOnCancel = process.env.T3_ACP_COMPLETE_FIRST_PROMPT_ON_CANCEL === "1";
......@@ -835,6 +837,108 @@ const program = Effect.gen(function* () {
835837 return yield* Effect.never;
836838 }
837839
840 if (emitGrokMonitorPostTurnPoll) {
841 const monitorCallId = "call-monitor-1";
842 const pollCallId = "call-monitor-poll-1";
843 const taskId = "01a05f41-5107-7550-821e-79e8d1cd7687";
844 const description = "Watch count-sheet Typst unit until done";
845 writeJsonRpcNotification("session/update", {
846 sessionId: requestedSessionId,
847 update: {
848 sessionUpdate: "tool_call",
849 toolCallId: monitorCallId,
850 title: "monitor",
851 kind: "other",
852 status: "pending",
853 rawInput: { description },
854 _meta: {
855 "x.ai/tool": { version: 1, name: "monitor", kind: "task", namespace: "grok_build" },
856 },
857 },
858 });
859 writeJsonRpcNotification("session/update", {
860 sessionId: requestedSessionId,
861 update: {
862 sessionUpdate: "tool_call_update",
863 toolCallId: monitorCallId,
864 status: "completed",
865 rawInput: { description },
866 rawOutput: {
867 type: "Monitor",
868 taskId,
869 timeoutMs: 36_000_000,
870 },
871 },
872 });
873 writeJsonRpcNotification("_x.ai/session/prompt_complete", {
874 sessionId: requestedSessionId,
875 promptId: promptIdFromRequestMeta(request) ?? "mock-xai-prompt-1",
876 stopReason: "end_turn",
877 agentResult: null,
878 });
879 yield* Effect.sleep("120 millis");
880 writeJsonRpcNotification("session/update", {
881 sessionId: requestedSessionId,
882 update: {
883 sessionUpdate: "tool_call",
884 toolCallId: pollCallId,
885 title: "get_command_or_subagent_output",
886 kind: "other",
887 status: "completed",
888 rawInput: { variant: "TaskOutput", task_ids: [taskId], timeout_ms: 0 },
889 rawOutput: {
890 type: "TaskOutput",
891 Result: {
892 task_id: taskId,
893 command: `[monitor] ${description}`,
894 status: "completed",
895 exit_code: 0,
896 output: "Monitor finished.",
897 },
898 },
899 },
900 });
901 return yield* Effect.never;
902 }
903
904 if (emitGrokBackgroundTaskStarted) {
905 const toolCallId = "call-fb9d0000-0000-0000-0000-000000000026";
906 const command = "sleep 40; echo done-a";
907 writeJsonRpcNotification("session/update", {
908 sessionId: requestedSessionId,
909 update: {
910 sessionUpdate: "tool_call",
911 toolCallId,
912 title: "run_terminal_command",
913 kind: "execute",
914 status: "in_progress",
915 rawInput: { command },
916 },
917 });
918 writeJsonRpcNotification("session/update", {
919 sessionId: requestedSessionId,
920 update: {
921 sessionUpdate: "tool_call_update",
922 toolCallId,
923 status: "completed",
924 rawOutput: {
925 type: "BackgroundTaskStarted",
926 task_id: toolCallId,
927 task_type: "bash",
928 status: "running",
929 command,
930 },
931 },
932 });
933 writeJsonRpcNotification("_x.ai/session/prompt_complete", {
934 sessionId: requestedSessionId,
935 promptId: promptIdFromRequestMeta(request) ?? "mock-xai-prompt-1",
936 stopReason: "end_turn",
937 agentResult: null,
938 });
939 return yield* Effect.never;
940 }
941
838942 if (emitInterleavedAssistantToolCalls) {
839943 const toolCallId = "tool-call-1";
840944
apps/server/src/provider/Layers/GrokAdapter.test.ts+62-2
......@@ -24,8 +24,10 @@ import {
2424 TurnId,
2525 type ProviderRuntimeEvent,
2626} from "@t3tools/contracts";
27import { HostProcessPlatform } from "@t3tools/shared/hostProcess";
2728
2829import { ServerConfig } from "../../config.ts";
30import { execScriptSource, writeFakeCli } from "../../testUtils/fakeCli.ts";
2931import {
3032 grokPromptSettlementBelongsToContext,
3133 isGrokEnterPlanModeToolCall,
......@@ -33,8 +35,7 @@ import {
3335 nextGrokPlanModeActive,
3436 selectGrokPermissionOptionId,
3537} from "./GrokAdapter.ts";
36import { execScriptSource, writeFakeCli } from "../../testUtils/fakeCli.ts";
37import { HostProcessPlatform } from "@t3tools/shared/hostProcess";
38
3839const decodeGrokSettings = Schema.decodeSync(GrokSettings);
3940
4041const __dirname = NodePath.dirname(NodeURL.fileURLToPath(import.meta.url));
......@@ -232,6 +233,65 @@ it.layer(grokAdapterTestLayer)("GrokAdapterLive", (it) => {
232233 }),
233234 );
234235
236 for (const taskType of ["monitor", "shell"] as const) {
237 it.effect(`emits the ${taskType} background lifecycle`, () =>
238 Effect.gen(function* () {
239 const threadId = ThreadId.make(`grok-background-${taskType}`);
240 const wrapperPath = yield* Effect.promise(() =>
241 makeMockGrokWrapper({
242 [taskType === "monitor"
243 ? "T3_ACP_EMIT_GROK_MONITOR_POST_TURN_POLL"
244 : "T3_ACP_EMIT_GROK_BACKGROUND_TASK_STARTED"]: "1",
245 }),
246 );
247 const adapter = yield* makeTestAdapter(wrapperPath);
248 const events: ProviderRuntimeEvent[] = [];
249 const finished = yield* Deferred.make<void>();
250 const eventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) =>
251 Effect.gen(function* () {
252 events.push(event);
253 if (event.type === (taskType === "monitor" ? "task.completed" : "turn.completed")) {
254 yield* Deferred.succeed(finished, undefined);
255 }
256 }),
257 ).pipe(Effect.forkChild);
258 yield* adapter.startSession({ threadId, cwd: process.cwd(), runtimeMode: "full-access" });
259 yield* adapter.sendTurn({ threadId, input: "watch the unit" });
260 yield* Deferred.await(finished).pipe(Effect.timeout("3 seconds"));
261
262 const started = events.find((event) => event.type === "task.started");
263 const taskId =
264 taskType === "monitor"
265 ? "01a05f41-5107-7550-821e-79e8d1cd7687"
266 : "call-fb9d0000-0000-0000-0000-000000000026";
267 assert.equal(started?.payload.taskType, taskType);
268 assert.equal(started?.payload.taskId, taskId);
269 if (taskType === "monitor") {
270 const completed = events.find((event) => event.type === "task.completed");
271 const turnEnd = events.findIndex((event) => event.type === "turn.completed");
272 assert.equal(completed?.payload.status, "completed");
273 assert.equal(completed?.payload.taskId, taskId);
274 assert.equal(completed?.turnId, undefined);
275 assert.isAtLeast(turnEnd, 0);
276 assert.isAbove(
277 events.findIndex((event) => event.type === "task.completed"),
278 turnEnd,
279 );
280 assert.deepEqual(
281 events
282 .slice(turnEnd + 1)
283 .filter((event) => event.type === "item.updated" || event.type === "item.completed"),
284 [],
285 );
286 } else {
287 assert.equal(started?.payload.description, "sleep 40; echo done-a");
288 }
289 yield* Fiber.interrupt(eventsFiber);
290 yield* adapter.stopSession(threadId);
291 }).pipe(TestClock.withLive),
292 );
293 }
294
235295 it.effect("sends runtime context with the current model without changing saved prompts", () =>
236296 Effect.gen(function* () {
237297 const threadId = ThreadId.make("grok-runtime-context");
apps/server/src/provider/Layers/GrokAdapter.ts+25
......@@ -68,6 +68,10 @@ import {
6868 normalizeGrokReasoningEffort,
6969 resolveGrokAcpBaseModelId,
7070} from "../acp/GrokAcpSupport.ts";
71import {
72 buildGrokBackgroundTaskEvents,
73 type GrokBackgroundTaskRecord,
74} from "../acp/XAiBackgroundTasks.ts";
7175import {
7276 extractGrokPlanMarkdownFromToolCallData,
7377 extractXAiAskUserQuestions,
......@@ -171,6 +175,8 @@ interface GrokSessionContext {
171175 currentModelId: string | undefined;
172176 currentReasoningEffort: string | undefined;
173177 stopped: boolean;
178 /** Live monitor/shell identities and their originating turns. */
179 readonly backgroundTasks: Map<string, GrokBackgroundTaskRecord>;
174180}
175181
176182function settlePendingApprovalsAsCancelled(
......@@ -1309,6 +1315,7 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte
13091315 ? normalizeGrokReasoningEffort(requestedStartReasoningEffort)
13101316 : currentStartReasoningEffort,
13111317 stopped: false,
1318 backgroundTasks: new Map(),
13121319 };
13131320
13141321 const nf = yield* Stream.runDrain(
......@@ -1331,6 +1338,24 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte
13311338 }
13321339
13331340 const notificationTurnId = resolveNotificationTurnId(ctx);
1341 if (event._tag === "ToolCallUpdated" && !ctx.stopped) {
1342 for (const taskEvent of buildGrokBackgroundTaskEvents({
1343 tasks: ctx.backgroundTasks,
1344 toolCallId: event.toolCall.toolCallId,
1345 rawInput: event.toolCall.data.rawInput,
1346 rawOutput: event.toolCall.data.rawOutput,
1347 toolCallStatus: event.toolCall.status,
1348 turnId: notificationTurnId,
1349 })) {
1350 yield* offerRuntimeEvent({
1351 ...taskEvent,
1352 ...(yield* makeEventStamp()),
1353 provider: PROVIDER,
1354 threadId: ctx.threadId,
1355 });
1356 }
1357 }
1358
13341359 if (
13351360 notificationTurnId === undefined ||
13361361 ctx.interruptedTurnIds.has(notificationTurnId)
apps/server/src/provider/acp/XAiBackgroundTasks.test.ts created+202
......@@ -0,0 +1,202 @@
1import { describe, expect, it } from "vite-plus/test";
2import { TurnId } from "@t3tools/contracts";
3
4import {
5 buildGrokBackgroundTaskEvents,
6 type GrokBackgroundTaskRecord,
7} from "./XAiBackgroundTasks.ts";
8
9const turnId = TurnId.make("turn-1");
10const monitor = { type: "Monitor", taskId: "monitor-1", timeoutMs: 60_000 };
11const shell = { type: "BackgroundTaskStarted", task_id: "shell-1", command: "sleep 40" };
12
13function mapper() {
14 const tasks = new Map<string, GrokBackgroundTaskRecord>();
15 const update = (
16 rawOutput: unknown,
17 overrides: Partial<Parameters<typeof buildGrokBackgroundTaskEvents>[0]> = {},
18 ) =>
19 buildGrokBackgroundTaskEvents({
20 tasks,
21 toolCallId: "call-1",
22 rawInput: { description: "Watch" },
23 rawOutput,
24 toolCallStatus: "completed",
25 turnId,
26 ...overrides,
27 });
28 return { tasks, update };
29}
30
31describe("Grok background tasks", () => {
32 it.each([
33 [monitor, "monitor-1", "monitor", "Watch"],
34 [shell, "shell-1", "shell", "sleep 40"],
35 ] as const)("starts and deduplicates %j", (output, id, taskType, description) => {
36 const { tasks, update } = mapper();
37 expect(update(output)).toEqual([
38 {
39 type: "task.started",
40 turnId,
41 payload: { taskId: id, taskType, description, title: description, toolUseId: "call-1" },
42 },
43 ]);
44 expect(update(output)).toEqual([]);
45 expect(tasks.size).toBe(1);
46 });
47
48 it("accepts automatic backgrounding before the tool becomes terminal", () => {
49 const { update } = mapper();
50 expect(update(shell, { toolCallStatus: "inProgress" })[0]?.type).toBe("task.started");
51 expect(update(monitor, { toolCallStatus: "inProgress" })).toEqual([]);
52 expect(update(monitor, { toolCallStatus: "failed" })).toEqual([]);
53 });
54
55 it.each([
56 ["running", null, "task.progress", undefined],
57 ["pending", null, "task.progress", undefined],
58 ["completed", 0, "task.completed", "completed"],
59 ["success", 0, "task.completed", "completed"],
60 ["succeeded", 0, "task.completed", "completed"],
61 ["failed", 1, "task.completed", "failed"],
62 ["error", 1, "task.completed", "failed"],
63 ["stopped", 137, "task.completed", "stopped"],
64 ["killed", 137, "task.completed", "stopped"],
65 ["cancelled", 137, "task.completed", "stopped"],
66 [undefined, 0, "task.completed", "completed"],
67 [undefined, 1, "task.completed", "failed"],
68 ])("maps poll status %s / exit %s", (status, exit_code, type, expectedStatus) => {
69 const { tasks, update } = mapper();
70 update(monitor);
71 const events = update({
72 type: "TaskOutput",
73 Result: {
74 task_id: "monitor-1",
75 command: "[monitor:Watch]",
76 status,
77 exit_code,
78 output: "\n result\nmore",
79 },
80 });
81 expect(events).toEqual([
82 {
83 type,
84 turnId,
85 payload: {
86 taskId: "monitor-1",
87 taskType: "monitor",
88 description: "Watch",
89 title: "Watch",
90 toolUseId: "call-1",
91 summary: "result",
92 ...(expectedStatus ? { status: expectedStatus } : {}),
93 },
94 },
95 ]);
96 expect(tasks.size).toBe(type === "task.progress" ? 1 : 0);
97 });
98
99 it.each([undefined, TurnId.make("turn-2")])(
100 "does not attribute old tasks to a later turn: %s",
101 (laterTurnId) => {
102 const { tasks, update } = mapper();
103 update(shell);
104 const events = update(
105 {
106 type: "TaskOutput",
107 Result: { task_id: "shell-1", command: "sleep 40", status: "completed" },
108 },
109 { turnId: laterTurnId },
110 );
111 expect(events).toHaveLength(1);
112 expect(events[0]?.turnId).toBeUndefined();
113 expect(tasks.size).toBe(0);
114 },
115 );
116
117 it("starts unknown poll tasks before progress or completion and ignores subagents", () => {
118 const { tasks, update } = mapper();
119 const events = update({
120 type: "TaskOutput",
121 MultiResult: {
122 results: [
123 { task_id: "shell-1", command: "sleep 40", status: "running" },
124 { task_id: "monitor-1", command: "[monitor] Watch", status: "completed" },
125 { task_id: "agent-1", command: "[subagent:executor] work", status: "running" },
126 ],
127 },
128 });
129 expect(events.map(({ type }) => type)).toEqual([
130 "task.started",
131 "task.progress",
132 "task.started",
133 "task.completed",
134 ]);
135 expect(events.map(({ payload }) => payload.taskType)).toEqual([
136 "shell",
137 "shell",
138 "monitor",
139 "monitor",
140 ]);
141 expect([...tasks.keys()]).toEqual(["shell-1"]);
142 expect(events.map((event) => event.turnId)).toEqual([
143 undefined,
144 undefined,
145 undefined,
146 undefined,
147 ]);
148 });
149
150 it("retires only successfully killed tasks, including mixed results", () => {
151 const { tasks, update } = mapper();
152 update(shell);
153 update(monitor);
154 const result = { type: "KillTask", Result: { task_id: "shell-1", outcome: "killed" } };
155 expect(update(result, { toolCallStatus: "failed" })).toEqual([]);
156 expect(tasks.size).toBe(2);
157 const events = update({
158 type: "KillTask",
159 MultiResult: {
160 results: [
161 result.Result,
162 { task_id: "monitor-1", outcome: "error" },
163 { task_id: "unknown", outcome: "killed" },
164 ],
165 },
166 });
167 expect(events).toEqual([
168 {
169 type: "task.completed",
170 turnId,
171 payload: {
172 taskId: "shell-1",
173 taskType: "shell",
174 description: "sleep 40",
175 title: "sleep 40",
176 toolUseId: "call-1",
177 status: "stopped",
178 },
179 },
180 ]);
181 expect([...tasks.keys()]).toEqual(["monitor-1"]);
182 });
183
184 it.each([
185 null,
186 [],
187 {},
188 { type: "Text", text: "subagent_id: fake\ntype: executor\ndescription: fake" },
189 { type: "Monitor", taskId: " " },
190 { type: "BackgroundTaskStarted", task_id: "shell-1" },
191 {
192 type: "TaskOutput",
193 MultiResult: {
194 results: [null, {}, { task_id: "task", command: "sleep 40", exit_code: Infinity }],
195 },
196 },
197 ])("ignores malformed or unrelated outputs: %j", (output) => {
198 const { tasks, update } = mapper();
199 expect(update(output)).toEqual([]);
200 expect(tasks.size).toBe(0);
201 });
202});
apps/server/src/provider/acp/XAiBackgroundTasks.ts created+156
......@@ -0,0 +1,156 @@
1import {
2 RuntimeTaskId,
3 type ProviderRuntimeTaskStartedEvent,
4 type ProviderRuntimeTaskProgressEvent,
5 type ProviderRuntimeTaskCompletedEvent,
6 type TurnId,
7} from "@t3tools/contracts";
8
9type TaskEvent =
10 | Pick<ProviderRuntimeTaskStartedEvent, "type" | "payload" | "turnId">
11 | Pick<ProviderRuntimeTaskProgressEvent, "type" | "payload" | "turnId">
12 | Pick<ProviderRuntimeTaskCompletedEvent, "type" | "payload" | "turnId">;
13
14export interface GrokBackgroundTaskRecord {
15 readonly payload: {
16 readonly taskId: RuntimeTaskId;
17 readonly taskType: "monitor" | "shell";
18 readonly description: string;
19 readonly title: string;
20 readonly toolUseId?: string;
21 };
22 readonly turnId: TurnId | undefined;
23}
24
25function record(value: unknown): Record<string, unknown> {
26 return typeof value === "object" && value !== null && !Array.isArray(value)
27 ? (value as Record<string, unknown>)
28 : {};
29}
30
31function text(value: unknown): string | undefined {
32 return typeof value === "string" ? value.trim() || undefined : undefined;
33}
34
35function lifecycle(status: unknown, exitCode: unknown) {
36 switch (text(status)?.toLowerCase()) {
37 case "pending":
38 case "running":
39 return "running";
40 case "completed":
41 case "success":
42 case "succeeded":
43 return "completed";
44 case "failed":
45 case "error":
46 return "failed";
47 case "stopped":
48 case "killed":
49 case "cancelled":
50 return "stopped";
51 default:
52 return typeof exitCode === "number" && Number.isFinite(exitCode)
53 ? exitCode === 0
54 ? "completed"
55 : "failed"
56 : undefined;
57 }
58}
59
60/** Map Grok's discriminated tool results, including notifications after the turn ends. */
61export function buildGrokBackgroundTaskEvents(input: {
62 readonly tasks: Map<string, GrokBackgroundTaskRecord>;
63 readonly toolCallId: string;
64 readonly rawInput: unknown;
65 readonly rawOutput: unknown;
66 readonly toolCallStatus: string | undefined;
67 readonly turnId?: TurnId | undefined;
68}): TaskEvent[] {
69 const { tasks, toolCallId, toolCallStatus, turnId } = input;
70 const output = record(input.rawOutput);
71 const events: TaskEvent[] = [];
72 if (
73 toolCallStatus !== "completed" &&
74 toolCallStatus !== "failed" &&
75 output.type !== "BackgroundTaskStarted"
76 ) {
77 return events;
78 }
79 const attribution = (task: GrokBackgroundTaskRecord) =>
80 task.turnId !== undefined && task.turnId === turnId ? { turnId } : {};
81 const start = (
82 id: string,
83 taskType: "monitor" | "shell",
84 description: string,
85 toolUseId?: string,
86 ) => {
87 const known = tasks.get(id);
88 if (known) return known;
89 const task: GrokBackgroundTaskRecord = {
90 payload: {
91 taskId: RuntimeTaskId.make(id),
92 taskType,
93 description,
94 title: description,
95 ...(toolUseId ? { toolUseId } : {}),
96 },
97 // Polls can rediscover older tasks without establishing their originating turn.
98 turnId: toolUseId ? turnId : undefined,
99 };
100 tasks.set(id, task);
101 events.push({ type: "task.started", payload: task.payload, ...attribution(task) });
102 return task;
103 };
104 const complete = (
105 task: GrokBackgroundTaskRecord,
106 status: "completed" | "failed" | "stopped",
107 summary?: string,
108 ) => {
109 tasks.delete(task.payload.taskId);
110 events.push({
111 type: "task.completed",
112 payload: { ...task.payload, status, ...(summary ? { summary } : {}) },
113 ...attribution(task),
114 });
115 };
116
117 if (output.type === "Monitor" && toolCallStatus === "completed") {
118 const id = text(output.taskId);
119 if (id) start(id, "monitor", text(record(input.rawInput).description) ?? "Monitor", toolCallId);
120 } else if (output.type === "BackgroundTaskStarted") {
121 const id = text(output.task_id) ?? text(output.taskId);
122 const command = text(output.command);
123 if (id && command) start(id, "shell", command.split("\n")[0]!.slice(0, 200), toolCallId);
124 } else if (output.type === "TaskOutput" || output.type === "KillTask") {
125 const results = record(output.MultiResult).results;
126 for (const value of Array.isArray(results) ? results : [output.Result]) {
127 const result = record(value);
128 const id = text(result.task_id);
129 if (!id) continue;
130 if (output.type === "KillTask") {
131 const task = tasks.get(id);
132 if (task && toolCallStatus === "completed" && result.outcome === "killed")
133 complete(task, "stopped");
134 continue;
135 }
136 const command = text(result.command);
137 const status = lifecycle(result.status, result.exit_code);
138 if (!command || !status || command.startsWith("[subagent:")) continue;
139 const task = start(id, /^\[monitor[:\]]/.test(command) ? "monitor" : "shell", command);
140 const summary = text(result.output)
141 ?.split("\n")
142 .find((line) => line.trim())
143 ?.trim();
144 if (status === "running") {
145 events.push({
146 type: "task.progress",
147 payload: { ...task.payload, ...(summary ? { summary } : {}) },
148 ...attribution(task),
149 });
150 } else {
151 complete(task, status, summary);
152 }
153 }
154 }
155 return events;
156}