diff --git a/apps/server/scripts/acp-mock-agent.ts b/apps/server/scripts/acp-mock-agent.ts index c923fa4010216baeb0f17eebe5fee3446acfec1d..e4fd848ab6e1b8c420c1d734510ce63f3dab3240 100644 --- a/apps/server/scripts/acp-mock-agent.ts +++ b/apps/server/scripts/acp-mock-agent.ts @@ -30,6 +30,8 @@ const emitXAiAskUserQuestionThenHang = const emitContentThenHang = process.env.T3_ACP_EMIT_CONTENT_THEN_HANG === "1"; const emitPlanThenHang = process.env.T3_ACP_EMIT_PLAN_THEN_HANG === "1"; const emitActiveToolThenHang = process.env.T3_ACP_EMIT_ACTIVE_TOOL_THEN_HANG === "1"; +const emitGrokMonitorPostTurnPoll = process.env.T3_ACP_EMIT_GROK_MONITOR_POST_TURN_POLL === "1"; +const emitGrokBackgroundTaskStarted = process.env.T3_ACP_EMIT_GROK_BACKGROUND_TASK_STARTED === "1"; const emitForeignSessionUpdates = process.env.T3_ACP_EMIT_FOREIGN_SESSION_UPDATES === "1"; const waitForResumeRelease = process.env.T3_ACP_WAIT_FOR_RESUME_RELEASE === "1"; const completeFirstPromptOnCancel = process.env.T3_ACP_COMPLETE_FIRST_PROMPT_ON_CANCEL === "1"; @@ -835,6 +837,108 @@ const program = Effect.gen(function* () { return yield* Effect.never; } + if (emitGrokMonitorPostTurnPoll) { + const monitorCallId = "call-monitor-1"; + const pollCallId = "call-monitor-poll-1"; + const taskId = "01a05f41-5107-7550-821e-79e8d1cd7687"; + const description = "Watch count-sheet Typst unit until done"; + writeJsonRpcNotification("session/update", { + sessionId: requestedSessionId, + update: { + sessionUpdate: "tool_call", + toolCallId: monitorCallId, + title: "monitor", + kind: "other", + status: "pending", + rawInput: { description }, + _meta: { + "x.ai/tool": { version: 1, name: "monitor", kind: "task", namespace: "grok_build" }, + }, + }, + }); + writeJsonRpcNotification("session/update", { + sessionId: requestedSessionId, + update: { + sessionUpdate: "tool_call_update", + toolCallId: monitorCallId, + status: "completed", + rawInput: { description }, + rawOutput: { + type: "Monitor", + taskId, + timeoutMs: 36_000_000, + }, + }, + }); + writeJsonRpcNotification("_x.ai/session/prompt_complete", { + sessionId: requestedSessionId, + promptId: promptIdFromRequestMeta(request) ?? "mock-xai-prompt-1", + stopReason: "end_turn", + agentResult: null, + }); + yield* Effect.sleep("120 millis"); + writeJsonRpcNotification("session/update", { + sessionId: requestedSessionId, + update: { + sessionUpdate: "tool_call", + toolCallId: pollCallId, + title: "get_command_or_subagent_output", + kind: "other", + status: "completed", + rawInput: { variant: "TaskOutput", task_ids: [taskId], timeout_ms: 0 }, + rawOutput: { + type: "TaskOutput", + Result: { + task_id: taskId, + command: `[monitor] ${description}`, + status: "completed", + exit_code: 0, + output: "Monitor finished.", + }, + }, + }, + }); + return yield* Effect.never; + } + + if (emitGrokBackgroundTaskStarted) { + const toolCallId = "call-fb9d0000-0000-0000-0000-000000000026"; + const command = "sleep 40; echo done-a"; + writeJsonRpcNotification("session/update", { + sessionId: requestedSessionId, + update: { + sessionUpdate: "tool_call", + toolCallId, + title: "run_terminal_command", + kind: "execute", + status: "in_progress", + rawInput: { command }, + }, + }); + writeJsonRpcNotification("session/update", { + sessionId: requestedSessionId, + update: { + sessionUpdate: "tool_call_update", + toolCallId, + status: "completed", + rawOutput: { + type: "BackgroundTaskStarted", + task_id: toolCallId, + task_type: "bash", + status: "running", + command, + }, + }, + }); + writeJsonRpcNotification("_x.ai/session/prompt_complete", { + sessionId: requestedSessionId, + promptId: promptIdFromRequestMeta(request) ?? "mock-xai-prompt-1", + stopReason: "end_turn", + agentResult: null, + }); + return yield* Effect.never; + } + if (emitInterleavedAssistantToolCalls) { const toolCallId = "tool-call-1"; diff --git a/apps/server/src/provider/Layers/GrokAdapter.test.ts b/apps/server/src/provider/Layers/GrokAdapter.test.ts index 9efb28d806289657d6bf9a0d808ba0ec5718d6de..545a6a094e4e5a116d4987b574089cceccea3a5a 100644 --- a/apps/server/src/provider/Layers/GrokAdapter.test.ts +++ b/apps/server/src/provider/Layers/GrokAdapter.test.ts @@ -24,8 +24,10 @@ import { TurnId, type ProviderRuntimeEvent, } from "@t3tools/contracts"; +import { HostProcessPlatform } from "@t3tools/shared/hostProcess"; import { ServerConfig } from "../../config.ts"; +import { execScriptSource, writeFakeCli } from "../../testUtils/fakeCli.ts"; import { grokPromptSettlementBelongsToContext, isGrokEnterPlanModeToolCall, @@ -33,8 +35,7 @@ import { nextGrokPlanModeActive, selectGrokPermissionOptionId, } from "./GrokAdapter.ts"; -import { execScriptSource, writeFakeCli } from "../../testUtils/fakeCli.ts"; -import { HostProcessPlatform } from "@t3tools/shared/hostProcess"; + const decodeGrokSettings = Schema.decodeSync(GrokSettings); const __dirname = NodePath.dirname(NodeURL.fileURLToPath(import.meta.url)); @@ -232,6 +233,65 @@ it.layer(grokAdapterTestLayer)("GrokAdapterLive", (it) => { }), ); + for (const taskType of ["monitor", "shell"] as const) { + it.effect(`emits the ${taskType} background lifecycle`, () => + Effect.gen(function* () { + const threadId = ThreadId.make(`grok-background-${taskType}`); + const wrapperPath = yield* Effect.promise(() => + makeMockGrokWrapper({ + [taskType === "monitor" + ? "T3_ACP_EMIT_GROK_MONITOR_POST_TURN_POLL" + : "T3_ACP_EMIT_GROK_BACKGROUND_TASK_STARTED"]: "1", + }), + ); + const adapter = yield* makeTestAdapter(wrapperPath); + const events: ProviderRuntimeEvent[] = []; + const finished = yield* Deferred.make(); + const eventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) => + Effect.gen(function* () { + events.push(event); + if (event.type === (taskType === "monitor" ? "task.completed" : "turn.completed")) { + yield* Deferred.succeed(finished, undefined); + } + }), + ).pipe(Effect.forkChild); + yield* adapter.startSession({ threadId, cwd: process.cwd(), runtimeMode: "full-access" }); + yield* adapter.sendTurn({ threadId, input: "watch the unit" }); + yield* Deferred.await(finished).pipe(Effect.timeout("3 seconds")); + + const started = events.find((event) => event.type === "task.started"); + const taskId = + taskType === "monitor" + ? "01a05f41-5107-7550-821e-79e8d1cd7687" + : "call-fb9d0000-0000-0000-0000-000000000026"; + assert.equal(started?.payload.taskType, taskType); + assert.equal(started?.payload.taskId, taskId); + if (taskType === "monitor") { + const completed = events.find((event) => event.type === "task.completed"); + const turnEnd = events.findIndex((event) => event.type === "turn.completed"); + assert.equal(completed?.payload.status, "completed"); + assert.equal(completed?.payload.taskId, taskId); + assert.equal(completed?.turnId, undefined); + assert.isAtLeast(turnEnd, 0); + assert.isAbove( + events.findIndex((event) => event.type === "task.completed"), + turnEnd, + ); + assert.deepEqual( + events + .slice(turnEnd + 1) + .filter((event) => event.type === "item.updated" || event.type === "item.completed"), + [], + ); + } else { + assert.equal(started?.payload.description, "sleep 40; echo done-a"); + } + yield* Fiber.interrupt(eventsFiber); + yield* adapter.stopSession(threadId); + }).pipe(TestClock.withLive), + ); + } + it.effect("sends runtime context with the current model without changing saved prompts", () => Effect.gen(function* () { const threadId = ThreadId.make("grok-runtime-context"); diff --git a/apps/server/src/provider/Layers/GrokAdapter.ts b/apps/server/src/provider/Layers/GrokAdapter.ts index 395d7e546f9a181fb0f06aac6a67f020cf5bacf6..01ec3d118a3eb828e59a04552be0745fad8c1f29 100644 --- a/apps/server/src/provider/Layers/GrokAdapter.ts +++ b/apps/server/src/provider/Layers/GrokAdapter.ts @@ -68,6 +68,10 @@ import { normalizeGrokReasoningEffort, resolveGrokAcpBaseModelId, } from "../acp/GrokAcpSupport.ts"; +import { + buildGrokBackgroundTaskEvents, + type GrokBackgroundTaskRecord, +} from "../acp/XAiBackgroundTasks.ts"; import { extractGrokPlanMarkdownFromToolCallData, extractXAiAskUserQuestions, @@ -171,6 +175,8 @@ interface GrokSessionContext { currentModelId: string | undefined; currentReasoningEffort: string | undefined; stopped: boolean; + /** Live monitor/shell identities and their originating turns. */ + readonly backgroundTasks: Map; } function settlePendingApprovalsAsCancelled( @@ -1309,6 +1315,7 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte ? normalizeGrokReasoningEffort(requestedStartReasoningEffort) : currentStartReasoningEffort, stopped: false, + backgroundTasks: new Map(), }; const nf = yield* Stream.runDrain( @@ -1331,6 +1338,24 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte } const notificationTurnId = resolveNotificationTurnId(ctx); + if (event._tag === "ToolCallUpdated" && !ctx.stopped) { + for (const taskEvent of buildGrokBackgroundTaskEvents({ + tasks: ctx.backgroundTasks, + toolCallId: event.toolCall.toolCallId, + rawInput: event.toolCall.data.rawInput, + rawOutput: event.toolCall.data.rawOutput, + toolCallStatus: event.toolCall.status, + turnId: notificationTurnId, + })) { + yield* offerRuntimeEvent({ + ...taskEvent, + ...(yield* makeEventStamp()), + provider: PROVIDER, + threadId: ctx.threadId, + }); + } + } + if ( notificationTurnId === undefined || ctx.interruptedTurnIds.has(notificationTurnId) diff --git a/apps/server/src/provider/acp/XAiBackgroundTasks.test.ts b/apps/server/src/provider/acp/XAiBackgroundTasks.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..93227b260481a58a7b23a8c80135d2551ea661b6 --- /dev/null +++ b/apps/server/src/provider/acp/XAiBackgroundTasks.test.ts @@ -0,0 +1,202 @@ +import { describe, expect, it } from "vite-plus/test"; +import { TurnId } from "@t3tools/contracts"; + +import { + buildGrokBackgroundTaskEvents, + type GrokBackgroundTaskRecord, +} from "./XAiBackgroundTasks.ts"; + +const turnId = TurnId.make("turn-1"); +const monitor = { type: "Monitor", taskId: "monitor-1", timeoutMs: 60_000 }; +const shell = { type: "BackgroundTaskStarted", task_id: "shell-1", command: "sleep 40" }; + +function mapper() { + const tasks = new Map(); + const update = ( + rawOutput: unknown, + overrides: Partial[0]> = {}, + ) => + buildGrokBackgroundTaskEvents({ + tasks, + toolCallId: "call-1", + rawInput: { description: "Watch" }, + rawOutput, + toolCallStatus: "completed", + turnId, + ...overrides, + }); + return { tasks, update }; +} + +describe("Grok background tasks", () => { + it.each([ + [monitor, "monitor-1", "monitor", "Watch"], + [shell, "shell-1", "shell", "sleep 40"], + ] as const)("starts and deduplicates %j", (output, id, taskType, description) => { + const { tasks, update } = mapper(); + expect(update(output)).toEqual([ + { + type: "task.started", + turnId, + payload: { taskId: id, taskType, description, title: description, toolUseId: "call-1" }, + }, + ]); + expect(update(output)).toEqual([]); + expect(tasks.size).toBe(1); + }); + + it("accepts automatic backgrounding before the tool becomes terminal", () => { + const { update } = mapper(); + expect(update(shell, { toolCallStatus: "inProgress" })[0]?.type).toBe("task.started"); + expect(update(monitor, { toolCallStatus: "inProgress" })).toEqual([]); + expect(update(monitor, { toolCallStatus: "failed" })).toEqual([]); + }); + + it.each([ + ["running", null, "task.progress", undefined], + ["pending", null, "task.progress", undefined], + ["completed", 0, "task.completed", "completed"], + ["success", 0, "task.completed", "completed"], + ["succeeded", 0, "task.completed", "completed"], + ["failed", 1, "task.completed", "failed"], + ["error", 1, "task.completed", "failed"], + ["stopped", 137, "task.completed", "stopped"], + ["killed", 137, "task.completed", "stopped"], + ["cancelled", 137, "task.completed", "stopped"], + [undefined, 0, "task.completed", "completed"], + [undefined, 1, "task.completed", "failed"], + ])("maps poll status %s / exit %s", (status, exit_code, type, expectedStatus) => { + const { tasks, update } = mapper(); + update(monitor); + const events = update({ + type: "TaskOutput", + Result: { + task_id: "monitor-1", + command: "[monitor:Watch]", + status, + exit_code, + output: "\n result\nmore", + }, + }); + expect(events).toEqual([ + { + type, + turnId, + payload: { + taskId: "monitor-1", + taskType: "monitor", + description: "Watch", + title: "Watch", + toolUseId: "call-1", + summary: "result", + ...(expectedStatus ? { status: expectedStatus } : {}), + }, + }, + ]); + expect(tasks.size).toBe(type === "task.progress" ? 1 : 0); + }); + + it.each([undefined, TurnId.make("turn-2")])( + "does not attribute old tasks to a later turn: %s", + (laterTurnId) => { + const { tasks, update } = mapper(); + update(shell); + const events = update( + { + type: "TaskOutput", + Result: { task_id: "shell-1", command: "sleep 40", status: "completed" }, + }, + { turnId: laterTurnId }, + ); + expect(events).toHaveLength(1); + expect(events[0]?.turnId).toBeUndefined(); + expect(tasks.size).toBe(0); + }, + ); + + it("starts unknown poll tasks before progress or completion and ignores subagents", () => { + const { tasks, update } = mapper(); + const events = update({ + type: "TaskOutput", + MultiResult: { + results: [ + { task_id: "shell-1", command: "sleep 40", status: "running" }, + { task_id: "monitor-1", command: "[monitor] Watch", status: "completed" }, + { task_id: "agent-1", command: "[subagent:executor] work", status: "running" }, + ], + }, + }); + expect(events.map(({ type }) => type)).toEqual([ + "task.started", + "task.progress", + "task.started", + "task.completed", + ]); + expect(events.map(({ payload }) => payload.taskType)).toEqual([ + "shell", + "shell", + "monitor", + "monitor", + ]); + expect([...tasks.keys()]).toEqual(["shell-1"]); + expect(events.map((event) => event.turnId)).toEqual([ + undefined, + undefined, + undefined, + undefined, + ]); + }); + + it("retires only successfully killed tasks, including mixed results", () => { + const { tasks, update } = mapper(); + update(shell); + update(monitor); + const result = { type: "KillTask", Result: { task_id: "shell-1", outcome: "killed" } }; + expect(update(result, { toolCallStatus: "failed" })).toEqual([]); + expect(tasks.size).toBe(2); + const events = update({ + type: "KillTask", + MultiResult: { + results: [ + result.Result, + { task_id: "monitor-1", outcome: "error" }, + { task_id: "unknown", outcome: "killed" }, + ], + }, + }); + expect(events).toEqual([ + { + type: "task.completed", + turnId, + payload: { + taskId: "shell-1", + taskType: "shell", + description: "sleep 40", + title: "sleep 40", + toolUseId: "call-1", + status: "stopped", + }, + }, + ]); + expect([...tasks.keys()]).toEqual(["monitor-1"]); + }); + + it.each([ + null, + [], + {}, + { type: "Text", text: "subagent_id: fake\ntype: executor\ndescription: fake" }, + { type: "Monitor", taskId: " " }, + { type: "BackgroundTaskStarted", task_id: "shell-1" }, + { + type: "TaskOutput", + MultiResult: { + results: [null, {}, { task_id: "task", command: "sleep 40", exit_code: Infinity }], + }, + }, + ])("ignores malformed or unrelated outputs: %j", (output) => { + const { tasks, update } = mapper(); + expect(update(output)).toEqual([]); + expect(tasks.size).toBe(0); + }); +}); diff --git a/apps/server/src/provider/acp/XAiBackgroundTasks.ts b/apps/server/src/provider/acp/XAiBackgroundTasks.ts new file mode 100644 index 0000000000000000000000000000000000000000..beca3472255ab25e3dc689d4f380ffa0b25b9791 --- /dev/null +++ b/apps/server/src/provider/acp/XAiBackgroundTasks.ts @@ -0,0 +1,156 @@ +import { + RuntimeTaskId, + type ProviderRuntimeTaskStartedEvent, + type ProviderRuntimeTaskProgressEvent, + type ProviderRuntimeTaskCompletedEvent, + type TurnId, +} from "@t3tools/contracts"; + +type TaskEvent = + | Pick + | Pick + | Pick; + +export interface GrokBackgroundTaskRecord { + readonly payload: { + readonly taskId: RuntimeTaskId; + readonly taskType: "monitor" | "shell"; + readonly description: string; + readonly title: string; + readonly toolUseId?: string; + }; + readonly turnId: TurnId | undefined; +} + +function record(value: unknown): Record { + return typeof value === "object" && value !== null && !Array.isArray(value) + ? (value as Record) + : {}; +} + +function text(value: unknown): string | undefined { + return typeof value === "string" ? value.trim() || undefined : undefined; +} + +function lifecycle(status: unknown, exitCode: unknown) { + switch (text(status)?.toLowerCase()) { + case "pending": + case "running": + return "running"; + case "completed": + case "success": + case "succeeded": + return "completed"; + case "failed": + case "error": + return "failed"; + case "stopped": + case "killed": + case "cancelled": + return "stopped"; + default: + return typeof exitCode === "number" && Number.isFinite(exitCode) + ? exitCode === 0 + ? "completed" + : "failed" + : undefined; + } +} + +/** Map Grok's discriminated tool results, including notifications after the turn ends. */ +export function buildGrokBackgroundTaskEvents(input: { + readonly tasks: Map; + readonly toolCallId: string; + readonly rawInput: unknown; + readonly rawOutput: unknown; + readonly toolCallStatus: string | undefined; + readonly turnId?: TurnId | undefined; +}): TaskEvent[] { + const { tasks, toolCallId, toolCallStatus, turnId } = input; + const output = record(input.rawOutput); + const events: TaskEvent[] = []; + if ( + toolCallStatus !== "completed" && + toolCallStatus !== "failed" && + output.type !== "BackgroundTaskStarted" + ) { + return events; + } + const attribution = (task: GrokBackgroundTaskRecord) => + task.turnId !== undefined && task.turnId === turnId ? { turnId } : {}; + const start = ( + id: string, + taskType: "monitor" | "shell", + description: string, + toolUseId?: string, + ) => { + const known = tasks.get(id); + if (known) return known; + const task: GrokBackgroundTaskRecord = { + payload: { + taskId: RuntimeTaskId.make(id), + taskType, + description, + title: description, + ...(toolUseId ? { toolUseId } : {}), + }, + // Polls can rediscover older tasks without establishing their originating turn. + turnId: toolUseId ? turnId : undefined, + }; + tasks.set(id, task); + events.push({ type: "task.started", payload: task.payload, ...attribution(task) }); + return task; + }; + const complete = ( + task: GrokBackgroundTaskRecord, + status: "completed" | "failed" | "stopped", + summary?: string, + ) => { + tasks.delete(task.payload.taskId); + events.push({ + type: "task.completed", + payload: { ...task.payload, status, ...(summary ? { summary } : {}) }, + ...attribution(task), + }); + }; + + if (output.type === "Monitor" && toolCallStatus === "completed") { + const id = text(output.taskId); + if (id) start(id, "monitor", text(record(input.rawInput).description) ?? "Monitor", toolCallId); + } else if (output.type === "BackgroundTaskStarted") { + const id = text(output.task_id) ?? text(output.taskId); + const command = text(output.command); + if (id && command) start(id, "shell", command.split("\n")[0]!.slice(0, 200), toolCallId); + } else if (output.type === "TaskOutput" || output.type === "KillTask") { + const results = record(output.MultiResult).results; + for (const value of Array.isArray(results) ? results : [output.Result]) { + const result = record(value); + const id = text(result.task_id); + if (!id) continue; + if (output.type === "KillTask") { + const task = tasks.get(id); + if (task && toolCallStatus === "completed" && result.outcome === "killed") + complete(task, "stopped"); + continue; + } + const command = text(result.command); + const status = lifecycle(result.status, result.exit_code); + if (!command || !status || command.startsWith("[subagent:")) continue; + const task = start(id, /^\[monitor[:\]]/.test(command) ? "monitor" : "shell", command); + const summary = text(result.output) + ?.split("\n") + .find((line) => line.trim()) + ?.trim(); + if (status === "running") { + events.push({ + type: "task.progress", + payload: { ...task.payload, ...(summary ? { summary } : {}) }, + ...attribution(task), + }); + } else { + complete(task, status, summary); + } + } + } + return events; +}