diff --git a/apps/server/scripts/acp-mock-agent.ts b/apps/server/scripts/acp-mock-agent.ts index 114c6224bb54..4298ac72ad7d 100644 --- a/apps/server/scripts/acp-mock-agent.ts +++ b/apps/server/scripts/acp-mock-agent.ts @@ -19,6 +19,7 @@ const emitInterleavedAssistantToolCalls = const emitGenericToolPlaceholders = process.env.T3_ACP_EMIT_GENERIC_TOOL_PLACEHOLDERS === "1"; const emitAskQuestion = process.env.T3_ACP_EMIT_ASK_QUESTION === "1"; const emitXAiAskUserQuestion = process.env.T3_ACP_EMIT_XAI_ASK_USER_QUESTION === "1"; +const emitXAiSubagent = process.env.T3_ACP_EMIT_XAI_SUBAGENT === "1"; const emitXAiPromptCompleteThenHang = process.env.T3_ACP_EMIT_XAI_PROMPT_COMPLETE_THEN_HANG === "1"; const emitForeignSessionUpdates = process.env.T3_ACP_EMIT_FOREIGN_SESSION_UPDATES === "1"; const hangPromptForever = process.env.T3_ACP_HANG_PROMPT_FOREVER === "1"; @@ -832,6 +833,59 @@ const program = Effect.gen(function* () { return { stopReason: "end_turn" }; } + if (emitXAiSubagent) { + yield* agent.client.sessionUpdate({ + sessionId: requestedSessionId, + update: { + sessionUpdate: "tool_call", + toolCallId: "spawn-subagent-tool-1", + title: "spawn_subagent", + kind: "other", + status: "pending", + rawInput: { + description: "Map current Cursor ACP", + prompt: "Research the Cursor adapter.", + }, + }, + }); + writeJsonRpcNotification("_x.ai/session/update", { + sessionId: requestedSessionId, + update: { + sessionUpdate: "subagent_spawned", + subagent_id: "mock-subagent-1", + parent_session_id: requestedSessionId, + child_session_id: "mock-subagent-1", + subagent_type: "explore", + description: "Map current Cursor ACP", + role: "explore", + model: "grok-4.6", + capability_mode: "read-only", + }, + }); + writeJsonRpcNotification("_x.ai/session/update", { + sessionId: requestedSessionId, + update: { + sessionUpdate: "subagent_finished", + subagent_id: "mock-subagent-1", + child_session_id: "mock-subagent-1", + status: "completed", + tool_calls: 12, + turns: 1, + duration_ms: 1500, + tokens_used: 4096, + output: "Cursor still uses ACP. Grok should keep the shared stack.", + }, + }); + yield* agent.client.sessionUpdate({ + sessionId: requestedSessionId, + update: { + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: "spawned an explorer" }, + }, + }); + return { stopReason: "end_turn" }; + } + if (emitForeignSessionUpdates) { yield* agent.client.sessionUpdate({ sessionId: requestedSessionId, diff --git a/apps/server/src/provider/Layers/GrokAdapter.test.ts b/apps/server/src/provider/Layers/GrokAdapter.test.ts index 9ccb83a0c94c..1cf70ea5af9c 100644 --- a/apps/server/src/provider/Layers/GrokAdapter.test.ts +++ b/apps/server/src/provider/Layers/GrokAdapter.test.ts @@ -1313,4 +1313,66 @@ it.layer(grokAdapterTestLayer)("GrokAdapterLive", (it) => { yield* adapter.stopSession(threadId); }), ); + + it.effect("maps xAI subagent spawn and finish onto the Agents task events", () => + Effect.gen(function* () { + const threadId = ThreadId.make("grok-subagent-roster"); + const wrapperPath = yield* Effect.promise(() => + makeMockGrokWrapper({ T3_ACP_EMIT_XAI_SUBAGENT: "1" }), + ); + const adapter = yield* makeTestAdapter(wrapperPath); + const started = + yield* Deferred.make>(); + const completed = + yield* Deferred.make>(); + const toolTitles = yield* Ref.make>([]); + const eventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) => { + if (String(event.threadId) !== String(threadId)) { + return Effect.void; + } + if (event.type === "task.started") { + return Deferred.succeed(started, event).pipe(Effect.ignore); + } + if (event.type === "task.completed") { + return Deferred.succeed(completed, event).pipe(Effect.ignore); + } + if ( + (event.type === "item.updated" || event.type === "item.completed") && + typeof event.payload.title === "string" + ) { + return Ref.update(toolTitles, (titles) => [...titles, event.payload.title ?? ""]); + } + return Effect.void; + }).pipe(Effect.forkChild); + + yield* adapter.startSession({ + threadId, + provider: ProviderDriverKind.make("grok"), + cwd: process.cwd(), + runtimeMode: "full-access", + modelSelection: { instanceId: ProviderInstanceId.make("grok"), model: "grok-build" }, + }); + yield* adapter.sendTurn({ threadId, input: "use subagents", attachments: [] }); + + const startedEvent = yield* Deferred.await(started); + const completedEvent = yield* Deferred.await(completed); + assert.equal(String(startedEvent.payload.taskId), "mock-subagent-1"); + assert.equal(startedEvent.payload.taskType, "local_agent"); + assert.equal(startedEvent.payload.role, "explore"); + assert.equal(startedEvent.payload.title, "Map current Cursor ACP"); + assert.equal(startedEvent.payload.model, "grok-4.6"); + assert.equal(completedEvent.payload.status, "completed"); + assert.equal(completedEvent.payload.typedUsage?.totalTokens, 4096); + assert.equal(completedEvent.payload.typedUsage?.toolUses, 12); + assert.equal( + completedEvent.payload.summary, + "Cursor still uses ACP. Grok should keep the shared stack.", + ); + const titles = yield* Ref.get(toolTitles); + assert.isFalse(titles.includes("spawn_subagent")); + + yield* Fiber.interrupt(eventsFiber); + yield* adapter.stopSession(threadId); + }), + ); }); diff --git a/apps/server/src/provider/Layers/GrokAdapter.ts b/apps/server/src/provider/Layers/GrokAdapter.ts index a4d99f942f87..4176f4b14c42 100644 --- a/apps/server/src/provider/Layers/GrokAdapter.ts +++ b/apps/server/src/provider/Layers/GrokAdapter.ts @@ -9,6 +9,7 @@ import { ProviderDriverKind, ProviderInstanceId, RuntimeRequestId, + RuntimeTaskId, type ThreadId, TurnId, } from "@t3tools/contracts"; @@ -68,14 +69,20 @@ import { import { extractGrokTokenUsage, extractXAiAskUserQuestions, + GROK_SUBAGENT_TASK_TYPE, grokPromptCountForTurns, grokRewindTargetForTurnCount, + grokSubagentCompletedStatus, + grokSubagentResultSummary, + isGrokSpawnSubagentToolTitle, makeXAiAskUserQuestionCancelledResponse, makeXAiAskUserQuestionResponse, parseGrokRewindExecute, parseGrokRewindPoints, + parseXAiKnownSubagentUpdate, promptResponseHasMissingXAiStopReason, XAiAskUserQuestionRequest, + XAiSessionUpdateNotification, } from "../acp/XAiAcpExtension.ts"; import { type GrokAdapterShape } from "../Services/GrokAdapter.ts"; import { type EventNdjsonLogger, makeEventNdjsonLogger } from "./EventNdjsonLogger.ts"; @@ -132,6 +139,7 @@ interface GrokSessionContext { reasoningEffortMenus: Map>; maxTokensByModel: Map; maxTokens: number | undefined; + spawnSubagentToolIds: Set; stopped: boolean; } @@ -239,6 +247,99 @@ export function grokPromptSettlementBelongsToContext(input: { ); } +const emitGrokSubagentSessionUpdate = (input: { + readonly threadId: ThreadId; + readonly method: string; + readonly params: XAiSessionUpdateNotification; + readonly sessions: ReadonlyMap; + readonly offerRuntimeEvent: (event: ProviderRuntimeEvent) => Effect.Effect; + readonly makeEventStamp: () => Effect.Effect<{ eventId: EventId; createdAt: string }>; + readonly logNative: (threadId: ThreadId, method: string, payload: unknown) => Effect.Effect; +}) => + Effect.gen(function* () { + const update = parseXAiKnownSubagentUpdate(input.params.update); + if (update === undefined) { + return; + } + yield* input.logNative(input.threadId, input.method, input.params); + const turnId = resolveSessionCallbackTurnId(input.sessions, input.threadId); + const stamp = yield* input.makeEventStamp(); + + if (update.sessionUpdate === "subagent_spawned") { + const role = update.role?.trim() || update.subagent_type?.trim() || undefined; + const title = update.description?.trim() || undefined; + yield* input.offerRuntimeEvent({ + type: "task.started", + ...stamp, + provider: PROVIDER, + threadId: input.threadId, + turnId, + payload: { + taskId: RuntimeTaskId.make(update.subagent_id), + taskType: GROK_SUBAGENT_TASK_TYPE, + ...(update.description?.trim() ? { description: update.description.trim() } : {}), + ...(title ? { title } : {}), + ...(role ? { role } : {}), + ...(update.model?.trim() ? { model: update.model.trim() } : {}), + }, + raw: { + source: "acp.grok.extension", + method: input.method, + payload: input.params, + }, + }); + return; + } + + const summary = grokSubagentResultSummary(update.output); + const totalTokens = + typeof update.tokens_used === "number" && + Number.isFinite(update.tokens_used) && + update.tokens_used >= 0 + ? Math.trunc(update.tokens_used) + : undefined; + const toolUses = + typeof update.tool_calls === "number" && + Number.isFinite(update.tool_calls) && + update.tool_calls >= 0 + ? Math.trunc(update.tool_calls) + : undefined; + const durationMs = + typeof update.duration_ms === "number" && + Number.isFinite(update.duration_ms) && + update.duration_ms >= 0 + ? Math.trunc(update.duration_ms) + : undefined; + const typedUsage = + totalTokens !== undefined || toolUses !== undefined || durationMs !== undefined + ? { + totalTokens: totalTokens ?? 0, + ...(toolUses !== undefined ? { toolUses } : {}), + ...(durationMs !== undefined ? { durationMs } : {}), + } + : undefined; + + yield* input.offerRuntimeEvent({ + type: "task.completed", + ...stamp, + provider: PROVIDER, + threadId: input.threadId, + turnId, + payload: { + taskId: RuntimeTaskId.make(update.subagent_id), + status: grokSubagentCompletedStatus(update.status), + taskType: GROK_SUBAGENT_TASK_TYPE, + ...(summary ? { summary } : {}), + ...(typedUsage ? { typedUsage } : {}), + }, + raw: { + source: "acp.grok.extension", + method: input.method, + payload: input.params, + }, + }); + }); + export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapterLiveOptions) { return Effect.gen(function* () { const boundInstanceId = options?.instanceId ?? ProviderInstanceId.make("grok"); @@ -683,6 +784,46 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte ), { discard: true }, ); + yield* Effect.forEach( + ["_x.ai/session/update", "x.ai/session/update"] as const, + (method) => + Effect.all( + [ + acp.handleExtNotification(method, XAiSessionUpdateNotification, (params) => + mapAcpCallbackFailure( + emitGrokSubagentSessionUpdate({ + threadId: input.threadId, + method, + params, + sessions, + offerRuntimeEvent: (event) => + offerRuntimeEvent(event).pipe(Effect.asVoid), + makeEventStamp: () => makeEventStamp().pipe(Effect.orDie), + logNative: (threadId, loggedMethod, payload) => + logNative(threadId, loggedMethod, payload).pipe(Effect.orDie), + }), + ), + ), + acp.handleExtRequest(method, XAiSessionUpdateNotification, (params) => + mapAcpCallbackFailure( + emitGrokSubagentSessionUpdate({ + threadId: input.threadId, + method, + params, + sessions, + offerRuntimeEvent: (event) => + offerRuntimeEvent(event).pipe(Effect.asVoid), + makeEventStamp: () => makeEventStamp().pipe(Effect.orDie), + logNative: (threadId, loggedMethod, payload) => + logNative(threadId, loggedMethod, payload).pipe(Effect.orDie), + }).pipe(Effect.as({})), + ), + ), + ], + { discard: true }, + ), + { discard: true }, + ); yield* acp.handleRequestPermission((params) => mapAcpCallbackFailure( Effect.gen(function* () { @@ -817,6 +958,7 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte maxTokens: (boundModelId ? maxTokensByModel.get(boundModelId) : undefined) ?? currentGrokMaxTokensFromSessionSetup(started.sessionSetupResult), + spawnSubagentToolIds: new Set(), stopped: false, }; @@ -884,6 +1026,13 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte ); return; case "ToolCallUpdated": + if ( + isGrokSpawnSubagentToolTitle(event.toolCall.title) || + ctx.spawnSubagentToolIds.has(event.toolCall.toolCallId) + ) { + ctx.spawnSubagentToolIds.add(event.toolCall.toolCallId); + return; + } yield* offerRuntimeEvent( makeAcpToolCallEvent({ stamp, diff --git a/apps/server/src/provider/acp/XAiAcpExtension.test.ts b/apps/server/src/provider/acp/XAiAcpExtension.test.ts index 5e976cd8ae32..d68efc7d0bed 100644 --- a/apps/server/src/provider/acp/XAiAcpExtension.test.ts +++ b/apps/server/src/provider/acp/XAiAcpExtension.test.ts @@ -13,6 +13,9 @@ import { extractXAiAskUserQuestions, grokPromptCountForTurns, grokRewindTargetForTurnCount, + grokSubagentCompletedStatus, + grokSubagentResultSummary, + isGrokSpawnSubagentToolTitle, makeXAiAskUserQuestionCancelledResponse, makeXAiAskUserQuestionResponse, makeXAiPromptCompletionRuntime, @@ -350,6 +353,15 @@ describe("Grok rewind and usage helpers", () => { expect(grokPromptCountForTurns([{ items: [1] }, { items: [2, 3] }], 1)).toBe(2); }); + it("maps Grok subagent finish status and trims the result", () => { + expect(grokSubagentCompletedStatus("completed")).toBe("completed"); + expect(grokSubagentCompletedStatus("failed")).toBe("failed"); + expect(grokSubagentCompletedStatus("cancelled")).toBe("stopped"); + expect(isGrokSpawnSubagentToolTitle("spawn_subagent")).toBe(true); + expect(isGrokSpawnSubagentToolTitle("read_file")).toBe(false); + expect(grokSubagentResultSummary(` ${"a".repeat(2001)} `)?.endsWith("...")).toBe(true); + }); + it("reads Grok token usage from prompt _meta", () => { expect( extractGrokTokenUsage({ diff --git a/apps/server/src/provider/acp/XAiAcpExtension.ts b/apps/server/src/provider/acp/XAiAcpExtension.ts index 2b480da5e06e..4357bf97d3c6 100644 --- a/apps/server/src/provider/acp/XAiAcpExtension.ts +++ b/apps/server/src/provider/acp/XAiAcpExtension.ts @@ -5,6 +5,7 @@ import type { } from "@t3tools/contracts"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; +import * as Option from "effect/Option"; import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; import type * as EffectAcpSchema from "effect-acp/schema"; @@ -60,6 +61,81 @@ export const XAiAskUserQuestionRequest = Schema.Union([ XAiWrappedAskUserQuestionParams, ]); +const XAiSubagentSpawnedUpdate = Schema.Struct({ + sessionUpdate: Schema.Literal("subagent_spawned"), + subagent_id: Schema.String, + child_session_id: Schema.optional(Schema.String), + parent_session_id: Schema.optional(Schema.String), + parent_prompt_id: Schema.optional(Schema.String), + subagent_type: Schema.optional(Schema.String), + description: Schema.optional(Schema.String), + role: Schema.optional(Schema.String), + model: Schema.optional(Schema.String), + capability_mode: Schema.optional(Schema.String), +}); + +const XAiSubagentFinishedUpdate = Schema.Struct({ + sessionUpdate: Schema.Literal("subagent_finished"), + subagent_id: Schema.String, + child_session_id: Schema.optional(Schema.String), + status: Schema.optional(Schema.String), + tool_calls: Schema.optional(Schema.Number), + turns: Schema.optional(Schema.Number), + duration_ms: Schema.optional(Schema.Number), + tokens_used: Schema.optional(Schema.Number), + output: Schema.optional(Schema.String), +}); + +const XAiKnownSubagentUpdate = Schema.Union([XAiSubagentSpawnedUpdate, XAiSubagentFinishedUpdate]); +const decodeXAiKnownSubagentUpdate = Schema.decodeUnknownOption(XAiKnownSubagentUpdate); + +export const XAiSessionUpdateNotification = Schema.Struct({ + sessionId: Schema.String, + update: Schema.Unknown, +}); +export type XAiSessionUpdateNotification = typeof XAiSessionUpdateNotification.Type; +export type XAiKnownSubagentUpdate = typeof XAiKnownSubagentUpdate.Type; + +export function parseXAiKnownSubagentUpdate(update: unknown): XAiKnownSubagentUpdate | undefined { + const decoded = decodeXAiKnownSubagentUpdate(update); + return Option.isSome(decoded) ? decoded.value : undefined; +} + +export const GROK_SUBAGENT_TASK_TYPE = "local_agent"; +const GROK_SUBAGENT_SUMMARY_LIMIT = 2000; + +export function isGrokSpawnSubagentToolTitle(title: string | undefined): boolean { + return title?.trim() === "spawn_subagent"; +} + +export function grokSubagentResultSummary(output: string | undefined): string | undefined { + const text = output?.trim(); + if (!text) { + return undefined; + } + if (text.length <= GROK_SUBAGENT_SUMMARY_LIMIT) { + return text; + } + return `${text.slice(0, GROK_SUBAGENT_SUMMARY_LIMIT - 3).trimEnd()}...`; +} + +export function grokSubagentCompletedStatus( + status: string | undefined, +): "completed" | "failed" | "stopped" { + switch (status?.trim().toLowerCase()) { + case "failed": + case "error": + return "failed"; + case "cancelled": + case "canceled": + case "interrupted": + case "stopped": + return "stopped"; + default: + return "completed"; + } +} + type XAiAskUserQuestionRequestParams = typeof XAiAskUserQuestionParams.Type; type XAiAskUserQuestionRequest = typeof XAiAskUserQuestionRequest.Type;