diff --git a/apps/web/src/web-host-router.ts b/apps/web/src/web-host-router.ts index d562ba4cd1..04e35ea7b8 100644 --- a/apps/web/src/web-host-router.ts +++ b/apps/web/src/web-host-router.ts @@ -134,6 +134,13 @@ const agentStubRouter = router({ }), // Called by resetSessionService() on logout/project switch. resetAll: publicProcedure.mutation(() => undefined), + // Unreachable in practice: the UI gates "/btw" on sessionSupportsSideQuestion, + // which is false for cloud sessions. Present so SessionTrpc stays satisfied. + sideQuestion: publicProcedure + .input(z.object({ sessionId: z.string(), question: z.string() })) + .mutation(() => { + throw new Error("Side questions require a local session"); + }), // Model/mode/effort options for the task-input preview + cloud run creation // (a cloud run requires a model). Real: fetched from the CORS-open PostHog LLM // gateway, same logic the desktop main process runs (see web-agent-config.ts). diff --git a/packages/agent/src/acp-extensions.ts b/packages/agent/src/acp-extensions.ts index c700c32bb7..8753ae8566 100644 --- a/packages/agent/src/acp-extensions.ts +++ b/packages/agent/src/acp-extensions.ts @@ -118,6 +118,13 @@ export const POSTHOG_METHODS = { * completed so the caller can safely send the next prompt. */ REFRESH_SESSION: "_posthog/refresh_session", + + /** + * One-shot side question ("/btw"): forks the live session's transcript into + * a single-turn, tool-less query and returns `{ answer }` without touching + * the main conversation. Payload: `{ question: string }`. + */ + SIDE_QUESTION: "_posthog/side_question", } as const; type PosthogNotification = diff --git a/packages/agent/src/adapters/claude/claude-agent.side-question.test.ts b/packages/agent/src/adapters/claude/claude-agent.side-question.test.ts new file mode 100644 index 0000000000..7dcc03783a --- /dev/null +++ b/packages/agent/src/adapters/claude/claude-agent.side-question.test.ts @@ -0,0 +1,263 @@ +import { + type AgentSideConnection, + RequestError, +} from "@agentclientprotocol/sdk"; +import { beforeEach, describe, expect, it, vi } from "vitest"; +import { POSTHOG_METHODS } from "../../acp-extensions"; +import { Pushable } from "../../utils/streams"; + +type SdkMessage = + | { type: "assistant"; message: { content: unknown[] } } + | { type: "result"; subtype: string }; + +/** Messages the next one-shot query() will yield when iterated. */ +let nextQueryMessages: SdkMessage[] = []; +/** When set, the next query's iterator blocks on this promise before yielding. */ +let nextQueryGate: Promise | null = null; + +const lastQueryCall: { + prompt?: unknown; + options?: Record; +} = {}; + +vi.mock("@anthropic-ai/claude-agent-sdk", () => ({ + query: vi.fn( + (params: { prompt: unknown; options: Record }) => { + lastQueryCall.prompt = params.prompt; + lastQueryCall.options = params.options; + const messages = nextQueryMessages; + const gate = nextQueryGate; + return { + close: vi.fn(), + [Symbol.asyncIterator]: async function* () { + if (gate) await gate; + yield* messages; + }, + }; + }, + ), +})); + +// Import after the mocks so ClaudeAcpAgent resolves the mocked SDK +const { ClaudeAcpAgent } = await import("./claude-agent"); +type Agent = InstanceType; + +function makeAgent(): Agent { + const client = { + sessionUpdate: vi.fn().mockResolvedValue(undefined), + extNotification: vi.fn().mockResolvedValue(undefined), + } as unknown as AgentSideConnection; + return new ClaudeAcpAgent(client); +} + +function installFakeSession( + agent: Agent, + sessionId: string, + overrides: Partial<{ modelId: string }> = {}, +) { + const abortController = new AbortController(); + const input = new Pushable(); + const oldQuery = { close: vi.fn() }; + const hooks = { PreToolUse: [] }; + + const session = { + query: oldQuery, + queryOptions: { + sessionId, + cwd: "/tmp/repo", + model: "claude-sonnet-4-6", + mcpServers: { + posthog: { type: "http", url: "https://example" }, + }, + abortController, + hooks, + }, + input, + modelId: overrides.modelId, + cwd: "/tmp/repo", + } as unknown as Parameters[0]; + + (agent as unknown as { session: unknown }).session = session; + (agent as unknown as { sessionId: string }).sessionId = sessionId; + + return { session, oldQuery, input, abortController }; +} + +function assistantText(text: string): SdkMessage { + return { type: "assistant", message: { content: [{ type: "text", text }] } }; +} + +describe("ClaudeAcpAgent.extMethod side_question", () => { + beforeEach(() => { + nextQueryMessages = []; + nextQueryGate = null; + lastQueryCall.prompt = undefined; + lastQueryCall.options = undefined; + }); + + it.each<[string, Record]>([ + ["missing question", {}], + ["non-string question", { question: 42 }], + ["blank question", { question: " " }], + ])("rejects %s with an invalid-params error", async (_label, params) => { + const agent = makeAgent(); + installFakeSession(agent, "s-1"); + + await expect( + agent.extMethod(POSTHOG_METHODS.SIDE_QUESTION, params), + ).rejects.toThrow(/non-empty question/); + }); + + it("forks a one-shot tool-less query and returns the answer", async () => { + const agent = makeAgent(); + const { session, oldQuery, input, abortController } = installFakeSession( + agent, + "s-2", + ); + nextQueryMessages = [ + assistantText("The function "), + assistantText("parses JSONL."), + { type: "result", subtype: "success" }, + ]; + + const result = await agent.extMethod(POSTHOG_METHODS.SIDE_QUESTION, { + question: "what does it do?", + }); + + expect(result).toEqual({ answer: "The function parses JSONL." }); + + // The prompt is the wrapped question, sent as a self-terminating string. + expect(lastQueryCall.prompt).toContain(""); + expect(lastQueryCall.prompt).toContain("what does it do?"); + + // One-shot fork off the live transcript, toolset removed. + const options = lastQueryCall.options ?? {}; + expect(options.resume).toBe("s-2"); + expect(options.forkSession).toBe(true); + expect(options.maxTurns).toBe(1); + expect(options.tools).toEqual([]); + expect(options.allowedTools).toEqual([]); + expect(options.mcpServers).toEqual({}); + expect(options.sessionId).toBeUndefined(); + expect(options.hooks).toBeUndefined(); + // Fresh controller: aborting the one-shot must not poison the session. + expect(options.abortController).not.toBe(abortController); + + // canUseTool hard-denies anything that slips past the empty toolset. + const canUseTool = options.canUseTool as () => Promise<{ + behavior: string; + }>; + await expect(canUseTool()).resolves.toMatchObject({ behavior: "deny" }); + + // The live session is untouched. + const live = session as { + query: unknown; + input: unknown; + queryOptions: { sessionId: string }; + }; + expect(live.query).toBe(oldQuery); + expect(live.input).toBe(input); + expect(live.queryOptions.sessionId).toBe("s-2"); + expect((agent as unknown as { sessionId: string }).sessionId).toBe("s-2"); + }); + + it("answers on the live session model, not the creation-time option", async () => { + const agent = makeAgent(); + installFakeSession(agent, "s-3", { modelId: "claude-opus-4-8" }); + nextQueryMessages = [ + assistantText("yes"), + { type: "result", subtype: "success" }, + ]; + + await agent.extMethod(POSTHOG_METHODS.SIDE_QUESTION, { question: "hm?" }); + + expect(lastQueryCall.options?.model).toContain("opus"); + }); + + it("surfaces an error result subtype as a RequestError", async () => { + const agent = makeAgent(); + installFakeSession(agent, "s-4"); + nextQueryMessages = [{ type: "result", subtype: "error_max_turns" }]; + + const promise = agent.extMethod(POSTHOG_METHODS.SIDE_QUESTION, { + question: "hm?", + }); + await expect(promise).rejects.toBeInstanceOf(RequestError); + await expect(promise).rejects.toThrow(/error_max_turns/); + }); + + it("rejects when the fork produces no answer text", async () => { + const agent = makeAgent(); + installFakeSession(agent, "s-5"); + nextQueryMessages = [{ type: "result", subtype: "success" }]; + + await expect( + agent.extMethod(POSTHOG_METHODS.SIDE_QUESTION, { question: "hm?" }), + ).rejects.toThrow(/no answer/); + }); + + it("allows only one side question at a time", async () => { + const agent = makeAgent(); + installFakeSession(agent, "s-6"); + + let release: () => void = () => {}; + nextQueryGate = new Promise((resolve) => { + release = resolve; + }); + nextQueryMessages = [ + assistantText("first"), + { type: "result", subtype: "success" }, + ]; + + const first = agent.extMethod(POSTHOG_METHODS.SIDE_QUESTION, { + question: "first?", + }); + await expect( + agent.extMethod(POSTHOG_METHODS.SIDE_QUESTION, { question: "second?" }), + ).rejects.toThrow(/already in progress/); + + release(); + await expect(first).resolves.toEqual({ answer: "first" }); + + // The guard clears once the first completes. + nextQueryGate = null; + nextQueryMessages = [ + assistantText("third"), + { type: "result", subtype: "success" }, + ]; + await expect( + agent.extMethod(POSTHOG_METHODS.SIDE_QUESTION, { question: "third?" }), + ).resolves.toEqual({ answer: "third" }); + }); + + it("clears the in-flight guard after a failed side question", async () => { + const agent = makeAgent(); + installFakeSession(agent, "s-8"); + + nextQueryMessages = [{ type: "result", subtype: "error_max_turns" }]; + await expect( + agent.extMethod(POSTHOG_METHODS.SIDE_QUESTION, { question: "first?" }), + ).rejects.toBeInstanceOf(RequestError); + + nextQueryMessages = [ + assistantText("second"), + { type: "result", subtype: "success" }, + ]; + await expect( + agent.extMethod(POSTHOG_METHODS.SIDE_QUESTION, { question: "second?" }), + ).resolves.toEqual({ answer: "second" }); + }); + + it("wraps SDK failures (e.g. nothing to resume) in a clear RequestError", async () => { + const agent = makeAgent(); + installFakeSession(agent, "s-7"); + nextQueryGate = Promise.reject(new Error("No conversation found")); + nextQueryGate.catch(() => {}); + + const promise = agent.extMethod(POSTHOG_METHODS.SIDE_QUESTION, { + question: "too early?", + }); + await expect(promise).rejects.toBeInstanceOf(RequestError); + await expect(promise).rejects.toThrow(/No conversation found/); + }); +}); diff --git a/packages/agent/src/adapters/claude/claude-agent.ts b/packages/agent/src/adapters/claude/claude-agent.ts index 14974dd3c6..fea6f47951 100644 --- a/packages/agent/src/adapters/claude/claude-agent.ts +++ b/packages/agent/src/adapters/claude/claude-agent.ts @@ -133,6 +133,11 @@ import { type ProcessSpawnedInfo, } from "./session/options"; import { SettingsManager } from "./session/settings"; +import { + buildSideQuestionPrompt, + collectSideQuestionAnswer, + SIDE_QUESTION_TIMEOUT_MS, +} from "./side-question"; import { CODE_EXECUTION_MODES, type CodeExecutionMode, @@ -267,6 +272,8 @@ export class ClaudeAcpAgent extends BaseAcpAgent { private options?: ClaudeAcpAgentOptions; private enrichment?: Enrichment; private enrichedReadCache: EnrichedReadCache = new Map(); + /** One side question at a time; bounds concurrent forks off the transcript. */ + private sideQuestionInFlight = false; constructor(client: AgentSideConnection, options?: ClaudeAcpAgentOptions) { super(client); @@ -317,6 +324,7 @@ export class ClaudeAcpAgent extends BaseAcpAgent { posthog: { resumeSession: true, steering: "native", + sideQuestion: true, }, claudeCode: { promptQueueing: true, @@ -1487,10 +1495,24 @@ export class ClaudeAcpAgent extends BaseAcpAgent { method: string, params: Record, ): Promise> { - if (!isMethod(method, POSTHOG_METHODS.REFRESH_SESSION)) { - throw RequestError.methodNotFound(method); + if (isMethod(method, POSTHOG_METHODS.SIDE_QUESTION)) { + if (typeof params.question !== "string" || !params.question.trim()) { + throw new RequestError( + -32602, + "side_question requires a non-empty question", + ); + } + return await this.answerSideQuestion(params.question); + } + if (isMethod(method, POSTHOG_METHODS.REFRESH_SESSION)) { + return await this.handleRefreshSession(params); } + throw RequestError.methodNotFound(method); + } + private async handleRefreshSession( + params: Record, + ): Promise> { // Trust boundary: refresh is only safe when the caller is trusted infra // (e.g. the sandbox agent-server). Do not route this method from // untrusted clients — parseMcpServers does no URL/command validation. @@ -1515,6 +1537,89 @@ export class ClaudeAcpAgent extends BaseAcpAgent { return { refreshed: true }; } + /** + * Answers a "/btw" side question by forking the live session's transcript + * into a one-shot, tool-less, single-turn query. Strictly non-mutating: + * the fork gets its own SDK session id and AbortController, and nothing on + * `this.session` is touched, so the main conversation (including an + * in-flight turn) never sees the exchange. + */ + private async answerSideQuestion( + question: string, + ): Promise<{ answer: string }> { + if (this.sideQuestionInFlight) { + throw new RequestError(-32002, "A side question is already in progress"); + } + this.sideQuestionInFlight = true; + + const abortController = new AbortController(); + try { + // Drop `sessionId` (identity comes from `resume`) and `hooks` (they + // close over live-session caches and task state). + const { + sessionId: _sessionId, + hooks: _hooks, + ...rest + } = this.session.queryOptions; + const options: Options = { + ...rest, + resume: this.sessionId, + forkSession: true, + maxTurns: 1, + // Belt and braces: remove the toolset entirely and deny anything + // that slips through; the prompt wrapper also says "no tools". + tools: [], + allowedTools: [], + canUseTool: async () => ({ + behavior: "deny", + message: "Tools are unavailable while answering a side question.", + interrupt: false, + }), + // Never reuse in-process MCP instances ("Already connected to a + // transport"); the fork has no tools, so it needs no servers. + mcpServers: {}, + abortController, + // `rest.model` is the creation-time value; the user may have + // switched models since, so answer on the live session model. + ...(this.session.modelId && { + model: toSdkModelId(this.session.modelId), + }), + }; + + const oneShot = query({ + prompt: buildSideQuestionPrompt(question), + options, + }); + + const answer = await withTimeout( + collectSideQuestionAnswer(oneShot), + SIDE_QUESTION_TIMEOUT_MS, + ); + + if (answer.result === "timeout") { + throw new RequestError( + -32603, + `Side question timed out after ${SIDE_QUESTION_TIMEOUT_MS}ms`, + ); + } + if (!answer.value) { + throw new RequestError(-32603, "Side question produced no answer"); + } + return { answer: answer.value }; + } catch (error) { + if (error instanceof RequestError) throw error; + // A brand-new session has no transcript on disk yet, so `resume` has + // nothing to fork; surface that case clearly. + throw new RequestError( + -32603, + `Side question failed: ${error instanceof Error ? error.message : String(error)}`, + ); + } finally { + abortController.abort(); + this.sideQuestionInFlight = false; + } + } + private async refreshSession( mcpServers: Record, ): Promise { diff --git a/packages/agent/src/adapters/claude/side-question.test.ts b/packages/agent/src/adapters/claude/side-question.test.ts new file mode 100644 index 0000000000..73f375a302 --- /dev/null +++ b/packages/agent/src/adapters/claude/side-question.test.ts @@ -0,0 +1,15 @@ +import { describe, expect, it } from "vitest"; +import { buildSideQuestionPrompt } from "./side-question"; + +describe("buildSideQuestionPrompt", () => { + it("wraps the question in the side-question constraints", () => { + const prompt = buildSideQuestionPrompt("what does this function do?"); + + expect(prompt).toMatch( + /^[\s\S]*<\/system-reminder>\n\nwhat does this function do\?$/, + ); + expect(prompt).toContain("no tools available"); + expect(prompt).toContain("one-off response"); + expect(prompt).toContain("Never promise actions"); + }); +}); diff --git a/packages/agent/src/adapters/claude/side-question.ts b/packages/agent/src/adapters/claude/side-question.ts new file mode 100644 index 0000000000..2b9cad7593 --- /dev/null +++ b/packages/agent/src/adapters/claude/side-question.ts @@ -0,0 +1,58 @@ +/** + * Helpers for the one-shot "/btw" side question. Session-independent only — + * the forked query mechanics live in ClaudeAcpAgent.answerSideQuestion. + */ + +import { RequestError } from "@agentclientprotocol/sdk"; +import type { SDKMessage } from "@anthropic-ai/claude-agent-sdk"; + +export const SIDE_QUESTION_TIMEOUT_MS = 120_000; + +/** + * Drains the one-shot fork's message stream and returns the assistant's + * answer text. Throws when the turn ends in an error result. + */ +export async function collectSideQuestionAnswer( + messages: AsyncIterable, +): Promise { + const chunks: string[] = []; + for await (const message of messages) { + if (message.type === "assistant") { + for (const block of message.message.content) { + if (block.type === "text") { + chunks.push(block.text); + } + } + } else if (message.type === "result") { + if (message.subtype !== "success") { + throw new RequestError( + -32603, + `Side question failed: ${message.subtype}`, + ); + } + break; + } + } + return chunks.join("").trim(); +} + +/** + * Wraps the user's question in the constraints the fork must obey: no tools, + * single one-off response, answer only from what the conversation already + * contains. Mirrors the prompt shape Claude Code uses for its /btw command. + */ +export function buildSideQuestionPrompt(question: string): string { + return [ + "", + "The user is asking a quick side question. This is a one-off response", + "outside the main conversation; nothing you say here will be visible in", + "later turns. You have no tools available: you cannot read files, run", + "commands, search, or take any action. Answer concisely using only the", + "knowledge already present in the conversation. Never promise actions", + '(do not say things like "Let me try..." or offer to investigate);', + "if you do not know the answer, say so plainly.", + "", + "", + question, + ].join("\n"); +} diff --git a/packages/agent/src/index.ts b/packages/agent/src/index.ts index dd4e608178..4acb307033 100644 --- a/packages/agent/src/index.ts +++ b/packages/agent/src/index.ts @@ -1,4 +1,8 @@ -export { isNotification, POSTHOG_NOTIFICATIONS } from "./acp-extensions"; +export { + isNotification, + POSTHOG_METHODS, + POSTHOG_NOTIFICATIONS, +} from "./acp-extensions"; export { getMcpToolMetadata, isMcpToolReadOnly, diff --git a/packages/agent/src/utils/common.ts b/packages/agent/src/utils/common.ts index 867f6b2f61..36646bc454 100644 --- a/packages/agent/src/utils/common.ts +++ b/packages/agent/src/utils/common.ts @@ -9,14 +9,19 @@ export async function withTimeout( operation: Promise, timeoutMs: number, ): Promise<{ result: "success"; value: T } | { result: "timeout" }> { - const timeoutPromise = new Promise<{ result: "timeout" }>((resolve) => - setTimeout(() => resolve({ result: "timeout" }), timeoutMs), - ); + let timer: ReturnType | undefined; + const timeoutPromise = new Promise<{ result: "timeout" }>((resolve) => { + timer = setTimeout(() => resolve({ result: "timeout" }), timeoutMs); + }); const operationPromise = operation.then((value) => ({ result: "success" as const, value, })); - return Promise.race([operationPromise, timeoutPromise]); + try { + return await Promise.race([operationPromise, timeoutPromise]); + } finally { + clearTimeout(timer); + } } /** diff --git a/packages/core/src/sessions/sessionService.ts b/packages/core/src/sessions/sessionService.ts index 38a1ede091..3fa316cd09 100644 --- a/packages/core/src/sessions/sessionService.ts +++ b/packages/core/src/sessions/sessionService.ts @@ -35,6 +35,7 @@ import { type StoredLogEntry, sendableQueuePrefixLength, sessionSupportsNativeSteer, + sessionSupportsSideQuestion, type TaskRunStatus, } from "@posthog/shared"; import { ANALYTICS_EVENTS } from "@posthog/shared/analytics-events"; @@ -179,6 +180,7 @@ export interface SessionTrpc { reconnect: TrpcMutation; cancel: TrpcMutation; prompt: TrpcMutation; + sideQuestion: TrpcMutation; cancelPrompt: TrpcMutation; cancelPermission: TrpcMutation; respondToPermission: TrpcMutation; @@ -1499,9 +1501,14 @@ export function isPermissionRequestAlreadySurfaced( ); } -/** The steering capability on a loosely-typed agent start/reconnect result. */ -function readSteering(result: unknown): string | undefined { - return (result as { steering?: string } | undefined)?.steering; +/** The negotiated capabilities on a loosely-typed agent start/reconnect result. */ +function readCapabilities(result: unknown): { + steering?: string; + sideQuestion?: boolean; +} { + const { steering, sideQuestion } = + (result as { steering?: string; sideQuestion?: boolean } | undefined) ?? {}; + return { steering, sideQuestion }; } function classifyTurnEventKind( @@ -2061,7 +2068,7 @@ export class SessionService { this.d.store.updateSession(taskRunId, { status: "connected", configOptions, - steering: readSteering(result), + ...readCapabilities(result), }); // Persist the merged config options @@ -2405,7 +2412,7 @@ export class SessionService { | SessionConfigOption[] | undefined; session.configOptions = configOptions; - session.steering = readSteering(result); + Object.assign(session, readCapabilities(result)); // Persist the config options if (configOptions) { @@ -3570,6 +3577,35 @@ export class SessionService { }); } + /** + * Ask a one-shot "/btw" side question: a single-turn, tool-less query forked + * off the live transcript. The exchange never enters the conversation, so + * this bypasses the queue/steer machinery entirely and is valid both + * mid-turn and idle. + */ + async askSideQuestion(taskId: string, question: string): Promise { + if (!this.d.getIsOnline()) { + throw new Error( + "No internet connection. Please check your connection and try again.", + ); + } + + const session = this.d.store.getSessionByTaskId(taskId); + if (!session) throw new Error("No active session for task"); + if (!sessionSupportsSideQuestion(session)) { + throw new Error("Side questions aren't supported for this session yet."); + } + if (session.status !== "connected") { + throw new Error("Session is not ready. Try again once it's connected."); + } + + const result = (await this.d.trpc.agent.sideQuestion.mutate({ + sessionId: session.taskRunId, + question, + })) as { answer: string }; + return result.answer; + } + /** * Send the next queued message as its own prompt. * Called internally when a turn completes and there are queued messages. diff --git a/packages/core/src/sessions/sessionServiceSideQuestion.test.ts b/packages/core/src/sessions/sessionServiceSideQuestion.test.ts new file mode 100644 index 0000000000..8c1ffbf5ab --- /dev/null +++ b/packages/core/src/sessions/sessionServiceSideQuestion.test.ts @@ -0,0 +1,99 @@ +import type { AgentSession } from "@posthog/shared"; +import { describe, expect, it, vi } from "vitest"; +import { SessionService, type SessionServiceDeps } from "./sessionService"; + +function makeSession(overrides: Partial = {}): AgentSession { + return { + taskRunId: "run-1", + taskId: "task-1", + taskTitle: "Test task", + channel: "", + events: [], + startedAt: 1, + status: "connected", + isPromptPending: false, + isCompacting: false, + promptStartedAt: null, + pendingPermissions: new Map(), + pausedDurationMs: 0, + messageQueue: [], + optimisticItems: [], + adapter: "claude", + ...overrides, + } as AgentSession; +} + +function createHarness(session: AgentSession | undefined, online = true) { + const sideQuestionMutate = vi.fn().mockResolvedValue({ answer: "42" }); + const deps = { + store: { + getSessionByTaskId: () => session, + }, + log: { info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() }, + getIsOnline: () => online, + trpc: { + agent: { + sideQuestion: { mutate: sideQuestionMutate }, + onSessionIdleKilled: { + subscribe: () => ({ unsubscribe: vi.fn() }), + }, + }, + }, + } as unknown as SessionServiceDeps; + + return { service: new SessionService(deps), sideQuestionMutate }; +} + +describe("SessionService.askSideQuestion", () => { + it("sends the question keyed by taskRunId and returns the answer", async () => { + const { service, sideQuestionMutate } = createHarness(makeSession()); + + await expect(service.askSideQuestion("task-1", "why?")).resolves.toBe("42"); + expect(sideQuestionMutate).toHaveBeenCalledWith({ + sessionId: "run-1", + question: "why?", + }); + }); + + it("rejects while offline", async () => { + const { service, sideQuestionMutate } = createHarness(makeSession(), false); + + await expect(service.askSideQuestion("task-1", "why?")).rejects.toThrow( + /No internet connection/, + ); + expect(sideQuestionMutate).not.toHaveBeenCalled(); + }); + + it("rejects when no session exists for the task", async () => { + const { service } = createHarness(undefined); + + await expect(service.askSideQuestion("task-1", "why?")).rejects.toThrow( + /No active session/, + ); + }); + + it.each<[string, Partial]>([ + ["cloud session", { isCloud: true, sideQuestion: true }], + ["codex without the capability", { adapter: "codex" }], + ])("rejects a %s as unsupported", async (_label, overrides) => { + const { service, sideQuestionMutate } = createHarness( + makeSession(overrides), + ); + + await expect(service.askSideQuestion("task-1", "why?")).rejects.toThrow( + /aren't supported/, + ); + expect(sideQuestionMutate).not.toHaveBeenCalled(); + }); + + it("rejects when the session is not connected", async () => { + const { service, sideQuestionMutate } = createHarness( + makeSession({ status: "connecting" }), + ); + + await expect(service.askSideQuestion("task-1", "why?")).rejects.toThrow( + /not ready/, + ); + expect(sideQuestionMutate).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/host-router/src/routers/agent.router.ts b/packages/host-router/src/routers/agent.router.ts index f7a7a43f52..95725f139c 100644 --- a/packages/host-router/src/routers/agent.router.ts +++ b/packages/host-router/src/routers/agent.router.ts @@ -24,6 +24,8 @@ import { rtkStatusOutput, sessionResponseSchema, setConfigOptionInput, + sideQuestionInput, + sideQuestionOutput, startSessionInput, subscribeSessionInput, } from "@posthog/workspace-server/services/agent/schemas"; @@ -51,6 +53,15 @@ export const agentRouter = router({ }), ), + sideQuestion: publicProcedure + .input(sideQuestionInput) + .output(sideQuestionOutput) + .mutation(({ ctx, input }) => + ctx.container + .get(AGENT_SERVICE) + .sideQuestion(input.sessionId, input.question), + ), + cancel: publicProcedure .input(cancelSessionInput) .mutation(({ ctx, input }) => diff --git a/packages/shared/src/index.ts b/packages/shared/src/index.ts index ff03602e7e..bd028cbeaa 100644 --- a/packages/shared/src/index.ts +++ b/packages/shared/src/index.ts @@ -258,6 +258,7 @@ export { type SessionStatus, sendableQueuePrefixLength, sessionSupportsNativeSteer, + sessionSupportsSideQuestion, } from "./sessions"; export type { SignalReportOrderingField, diff --git a/packages/shared/src/sessions.test.ts b/packages/shared/src/sessions.test.ts index 8706578afe..04fc7a35d3 100644 --- a/packages/shared/src/sessions.test.ts +++ b/packages/shared/src/sessions.test.ts @@ -4,6 +4,7 @@ import { type AgentSession, resolveBypassRevertMode, sessionSupportsNativeSteer, + sessionSupportsSideQuestion, } from "./sessions"; function modeOption( @@ -115,3 +116,40 @@ describe("sessionSupportsNativeSteer", () => { expect(sessionSupportsNativeSteer(session)).toBe(expected); }); }); + +describe("sessionSupportsSideQuestion", () => { + type Case = Pick; + + it.each<[string, Case, boolean]>([ + [ + "claude advertises the capability", + { isCloud: false, sideQuestion: true, adapter: "claude" }, + true, + ], + // Fallback: pre-capability start paths leave sideQuestion unset; local + // claude is assumed capable, matching the steer fallback. + [ + "claude with no capability (fallback)", + { isCloud: false, sideQuestion: undefined, adapter: "claude" }, + true, + ], + [ + "codex with no capability (no fallback)", + { isCloud: false, sideQuestion: undefined, adapter: "codex" }, + false, + ], + [ + "claude explicitly without the capability", + { isCloud: false, sideQuestion: false, adapter: "claude" }, + false, + ], + // The fork runs against the local transcript, so cloud never qualifies. + [ + "cloud claude with the capability", + { isCloud: true, sideQuestion: true, adapter: "claude" }, + false, + ], + ])("%s", (_label, session, expected) => { + expect(sessionSupportsSideQuestion(session)).toBe(expected); + }); +}); diff --git a/packages/shared/src/sessions.ts b/packages/shared/src/sessions.ts index ca64dcb01e..8488b398be 100644 --- a/packages/shared/src/sessions.ts +++ b/packages/shared/src/sessions.ts @@ -79,6 +79,12 @@ export interface AgentSession { * means the host must cancel + resend. Drives the steer-vs-resend decision. */ steering?: string; + /** + * Adapter's negotiated side-question capability (`_meta.posthog.sideQuestion` + * from initialize). True means the adapter can answer a one-shot "/btw" + * question forked off the live transcript without touching the conversation. + */ + sideQuestion?: boolean; pendingPermissions: Map; pausedDurationMs: number; messageQueue: QueuedMessage[]; @@ -260,3 +266,20 @@ export function sessionSupportsNativeSteer( if (session.isCloud) return false; return session.steering == null && session.adapter === "claude"; } + +/** + * Whether the session can answer a one-shot "/btw" side question (a + * single-turn, tool-less query forked off the live transcript). Decided by the + * adapter's negotiated `sideQuestion` capability; cloud sessions can't — the + * fork runs against the local transcript. + * + * Fallback: if `sideQuestion` is unset (a session started before capability + * plumbing), local Claude is assumed capable, matching the steer fallback. + */ +export function sessionSupportsSideQuestion( + session: Pick, +): boolean { + if (session.isCloud) return false; + if (session.sideQuestion === true) return true; + return session.sideQuestion == null && session.adapter === "claude"; +} diff --git a/packages/ui/src/features/message-editor/commands.test.ts b/packages/ui/src/features/message-editor/commands.test.ts index 420de06c2f..18cc6edd58 100644 --- a/packages/ui/src/features/message-editor/commands.test.ts +++ b/packages/ui/src/features/message-editor/commands.test.ts @@ -1,7 +1,15 @@ -import { describe, expect, it } from "vitest"; -import { rewriteLocalSkillCommandPrompt } from "./commands"; +import { describe, expect, it, vi } from "vitest"; +import { + rewriteLocalSkillCommandPrompt, + tryExecuteCodeCommand, +} from "./commands"; import type { EditorAvailableCommand } from "./types"; +const toastError = vi.hoisted(() => vi.fn()); +vi.mock("@posthog/ui/primitives/toast", () => ({ + toast: { error: toastError, success: vi.fn() }, +})); + const commands: EditorAvailableCommand[] = [ { name: "local-test-skill", @@ -38,3 +46,52 @@ describe("message editor commands", () => { ).toBe(null); }); }); + +describe("/btw command", () => { + const baseContext = { + taskId: "task-1", + repoPath: "/tmp/repo", + session: null, + taskRun: null, + }; + + it("routes the question to askSideQuestion and reports handled", async () => { + const askSideQuestion = vi.fn(); + const handled = await tryExecuteCodeCommand("/btw what changed here?", { + ...baseContext, + askSideQuestion, + }); + expect(handled).toBe(true); + expect(askSideQuestion).toHaveBeenCalledWith("what changed here?"); + }); + + it("toasts on an empty question without calling askSideQuestion", async () => { + toastError.mockClear(); + const askSideQuestion = vi.fn(); + const handled = await tryExecuteCodeCommand("/btw", { + ...baseContext, + askSideQuestion, + }); + expect(handled).toBe(true); + expect(askSideQuestion).not.toHaveBeenCalled(); + expect(toastError).toHaveBeenCalledWith("Add a question after /btw"); + }); + + it("toasts when the session doesn't support side questions", async () => { + toastError.mockClear(); + const handled = await tryExecuteCodeCommand("/btw is this supported?", { + ...baseContext, + askSideQuestion: undefined, + }); + expect(handled).toBe(true); + expect(toastError).toHaveBeenCalledWith( + "Side questions aren't supported for this session yet.", + ); + }); + + it("leaves non-command text unhandled", async () => { + expect(await tryExecuteCodeCommand("btw not a command", baseContext)).toBe( + false, + ); + }); +}); diff --git a/packages/ui/src/features/message-editor/commands.ts b/packages/ui/src/features/message-editor/commands.ts index 8fb6956755..531942ed82 100644 --- a/packages/ui/src/features/message-editor/commands.ts +++ b/packages/ui/src/features/message-editor/commands.ts @@ -26,6 +26,8 @@ interface CommandContext { events: unknown[]; } | null; taskRun: { id?: string; log_url?: string } | null; + /** Fires a "/btw" side question. Absent when the session doesn't support them. */ + askSideQuestion?: (question: string) => void; } export interface CodeCommandInsertContext { @@ -104,8 +106,28 @@ const addDirCommand: CodeCommand = { }, }; +const btwCommand: CodeCommand = { + name: "btw", + description: + "Ask a quick side question without interrupting the conversation", + input: { hint: "your question" }, + execute(args, ctx) { + const question = args?.trim(); + if (!question) { + toast.error("Add a question after /btw"); + return; + } + if (!ctx.askSideQuestion) { + toast.error("Side questions aren't supported for this session yet."); + return; + } + ctx.askSideQuestion(question); + }, +}; + const commands: CodeCommand[] = [ addDirCommand, + btwCommand, makeFeedbackCommand("good", "good", "Positive"), makeFeedbackCommand("bad", "bad", "Negative"), makeFeedbackCommand("feedback", "general", "General"), diff --git a/packages/ui/src/features/sessions/components/SessionView.tsx b/packages/ui/src/features/sessions/components/SessionView.tsx index 163f19fbc4..8ad422d49c 100644 --- a/packages/ui/src/features/sessions/components/SessionView.tsx +++ b/packages/ui/src/features/sessions/components/SessionView.tsx @@ -30,6 +30,7 @@ import { QueuedMessagesDock } from "@posthog/ui/features/sessions/components/Que import { ReasoningLevelSelector } from "@posthog/ui/features/sessions/components/ReasoningLevelSelector"; import { RawLogsView } from "@posthog/ui/features/sessions/components/raw-logs/RawLogsView"; import { SessionResourcesBar } from "@posthog/ui/features/sessions/components/SessionResourcesBar"; +import { SideQuestionCard } from "@posthog/ui/features/sessions/components/SideQuestionCard"; import { SteerQueueToggle } from "@posthog/ui/features/sessions/components/SteerQueueToggle"; import { ThreadView } from "@posthog/ui/features/sessions/components/ThreadView"; import { CHAT_CONTENT_MAX_WIDTH } from "@posthog/ui/features/sessions/constants"; @@ -335,6 +336,7 @@ export function SessionView({ (s) => !!s?.editingQueuedId, ); const cancelQueuedEdit = useCancelQueuedMessageEdit(taskId); + const activeTaskRunId = useSessionSelector(taskId, (s) => s?.taskRunId); const [isDraggingFile, setIsDraggingFile] = useState(false); const editorRef = useRef(null); @@ -671,6 +673,12 @@ export function SessionView({ }`} > + {taskId && ( + + )} {taskId && } { + beforeEach(() => { + useSideQuestionStore.setState({ byTaskId: {} }); + }); + + it("renders nothing when there is no entry for the task", () => { + const { container } = render( + + + , + ); + expect(container.firstElementChild).toBeEmptyDOMElement(); + }); + + it("shows a spinner while the question is pending", () => { + useSideQuestionStore.setState({ + byTaskId: { + "task-1": { + id: "q-1", + question: "what changed?", + taskRunId: "run-1", + status: "pending", + }, + }, + }); + + render( + + + , + ); + + expect(screen.getByText("what changed?")).toBeInTheDocument(); + expect(screen.getByText("Answering…")).toBeInTheDocument(); + }); + + it("renders the answer once done", () => { + useSideQuestionStore.setState({ + byTaskId: { + "task-1": { + id: "q-1", + question: "what changed?", + taskRunId: "run-1", + status: "done", + answer: "The function parses JSONL.", + }, + }, + }); + + render( + + + , + ); + + expect(screen.getByText("The function parses JSONL.")).toBeInTheDocument(); + }); + + it("renders the error message on failure", () => { + useSideQuestionStore.setState({ + byTaskId: { + "task-1": { + id: "q-1", + question: "what changed?", + taskRunId: "run-1", + status: "error", + error: "Side question timed out", + }, + }, + }); + + render( + + + , + ); + + expect(screen.getByText("Side question timed out")).toBeInTheDocument(); + }); + + it("renders nothing when the entry belongs to a prior run", () => { + useSideQuestionStore.setState({ + byTaskId: { + "task-1": { + id: "q-1", + question: "what changed?", + taskRunId: "run-1", + status: "done", + answer: "The function parses JSONL.", + }, + }, + }); + + const { container } = render( + + + , + ); + + expect(container.firstElementChild).toBeEmptyDOMElement(); + }); + + it("dismisses the entry when the dismiss button is clicked", () => { + useSideQuestionStore.setState({ + byTaskId: { + "task-1": { + id: "q-1", + question: "what changed?", + taskRunId: "run-1", + status: "pending", + }, + }, + }); + + render( + + + , + ); + + fireEvent.click(screen.getByLabelText("Dismiss side question")); + + expect(useSideQuestionStore.getState().byTaskId["task-1"]).toBeUndefined(); + }); +}); diff --git a/packages/ui/src/features/sessions/components/SideQuestionCard.tsx b/packages/ui/src/features/sessions/components/SideQuestionCard.tsx new file mode 100644 index 0000000000..5d23867966 --- /dev/null +++ b/packages/ui/src/features/sessions/components/SideQuestionCard.tsx @@ -0,0 +1,63 @@ +import { ChatCircleDots, Spinner, X } from "@phosphor-icons/react"; +import { useSideQuestionStore } from "@posthog/ui/features/sessions/sideQuestionStore"; +import { Box, Flex, IconButton, Text, Tooltip } from "@radix-ui/themes"; +import { MarkdownRenderer } from "../../editor/components/MarkdownRenderer"; + +interface SideQuestionCardProps { + taskId: string; + /** The task's current run. An entry asked against a prior run is hidden. */ + taskRunId?: string; +} + +/** + * Ephemeral "/btw" side-question card pinned above the composer. The exchange + * lives only in view state — it is never part of the session transcript — so + * dismissing it leaves no trace. + */ +export function SideQuestionCard({ taskId, taskRunId }: SideQuestionCardProps) { + const entry = useSideQuestionStore((s) => s.byTaskId[taskId]); + const dismiss = useSideQuestionStore((s) => s.dismiss); + + if (!entry || entry.taskRunId !== taskRunId) return null; + + return ( + + + + + {entry.question} + + + dismiss(taskId)} + > + + + + + + {entry.status === "pending" && ( + + + Answering… + + )} + {entry.status === "done" && ( + + + + )} + {entry.status === "error" && ( + {entry.error} + )} + + + ); +} diff --git a/packages/ui/src/features/sessions/hooks/useSessionCallbacks.test.tsx b/packages/ui/src/features/sessions/hooks/useSessionCallbacks.test.tsx index 394156e47d..8fce8360c6 100644 --- a/packages/ui/src/features/sessions/hooks/useSessionCallbacks.test.tsx +++ b/packages/ui/src/features/sessions/hooks/useSessionCallbacks.test.tsx @@ -1,3 +1,4 @@ +import type { AgentSession } from "@posthog/shared"; import type { Task } from "@posthog/shared/domain-types"; import { renderHook } from "@testing-library/react"; import { beforeEach, describe, expect, it, vi } from "vitest"; @@ -46,9 +47,13 @@ vi.mock("@posthog/ui/features/sessions/hooks/useMessagingMode", () => ({ useMessagingMode: () => "queue", })); -// No code command / skill rewrite; the raw text is used as the prompt. +// No code command / skill rewrite by default; the raw text is used as the +// prompt. Kept as a spy so tests can inspect the CommandContext it received. +const tryExecuteCodeCommand = vi.hoisted(() => + vi.fn(async (_text: string, _ctx: Record) => false), +); vi.mock("@posthog/ui/features/message-editor/commands", () => ({ - tryExecuteCodeCommand: async () => false, + tryExecuteCodeCommand, rewriteLocalSkillCommandPrompt: () => null, resolveLocalSkillPrompt: async (text: string) => text, })); @@ -93,6 +98,27 @@ function renderCallbacks() { ); } +function makeSession(overrides: Partial): AgentSession { + return { + taskRunId: "run-1", + taskId: TASK, + taskTitle: "Test", + channel: "agent-event:run-1", + events: [], + startedAt: 0, + status: "connected", + isPromptPending: false, + isCompacting: false, + promptStartedAt: null, + pendingPermissions: new Map(), + pausedDurationMs: 0, + messageQueue: [], + optimisticItems: [], + adapter: "claude", + ...overrides, + } as AgentSession; +} + describe("useSessionCallbacks.handleSendPrompt while editing a queued message", () => { beforeEach(() => { vi.clearAllMocks(); @@ -224,3 +250,43 @@ describe("useSessionCallbacks.handleCancelPrompt", () => { }); }); }); + +describe("useSessionCallbacks.handleSendPrompt askSideQuestion gating", () => { + beforeEach(() => { + vi.clearAllMocks(); + sessionState.editingQueuedId = undefined; + sessionState.messageQueue = []; + }); + + it("passes a callable askSideQuestion for a session that supports it", async () => { + const { result } = renderHook(() => + useSessionCallbacks({ + taskId: TASK, + task, + session: makeSession({ isCloud: false, sideQuestion: true }), + repoPath: "/repo", + }), + ); + await result.current.handleSendPrompt("/btw what changed?"); + + const ctx = tryExecuteCodeCommand.mock.calls.at(-1)?.[1]; + expect(ctx).toBeDefined(); + expect(ctx?.askSideQuestion).toBeInstanceOf(Function); + }); + + it("passes undefined askSideQuestion for a session that does not support it", async () => { + const { result } = renderHook(() => + useSessionCallbacks({ + taskId: TASK, + task, + session: makeSession({ isCloud: true, sideQuestion: true }), + repoPath: "/repo", + }), + ); + await result.current.handleSendPrompt("/btw what changed?"); + + const ctx = tryExecuteCodeCommand.mock.calls.at(-1)?.[1]; + expect(ctx).toBeDefined(); + expect(ctx?.askSideQuestion).toBeUndefined(); + }); +}); diff --git a/packages/ui/src/features/sessions/hooks/useSessionCallbacks.ts b/packages/ui/src/features/sessions/hooks/useSessionCallbacks.ts index 188d470ce2..b2f65c2c56 100644 --- a/packages/ui/src/features/sessions/hooks/useSessionCallbacks.ts +++ b/packages/ui/src/features/sessions/hooks/useSessionCallbacks.ts @@ -13,6 +13,7 @@ import { } from "@posthog/core/sessions/sessionService"; import { useService } from "@posthog/di/react"; import { useHostTRPCClient } from "@posthog/host-router/react"; +import { sessionSupportsSideQuestion } from "@posthog/shared"; import type { Task } from "@posthog/shared/domain-types"; import { resolveLocalSkillPrompt, @@ -25,6 +26,7 @@ import { type AgentSession, sessionStoreSetters, } from "@posthog/ui/features/sessions/sessionStore"; +import { fireSideQuestion } from "@posthog/ui/features/sessions/sideQuestionStore"; import { useTaskViewed } from "@posthog/ui/features/sidebar/useTaskViewed"; import { SHELL_CLIENT, @@ -76,6 +78,16 @@ export function useSessionCallbacks({ } : null, taskRun: task.latest_run ?? null, + askSideQuestion: + currentSession && sessionSupportsSideQuestion(currentSession) + ? (question) => + fireSideQuestion( + sessionService, + taskId, + currentSession.taskRunId, + question, + ) + : undefined, }); if (handled) return; diff --git a/packages/ui/src/features/sessions/sideQuestionStore.test.ts b/packages/ui/src/features/sessions/sideQuestionStore.test.ts new file mode 100644 index 0000000000..f9da26dd35 --- /dev/null +++ b/packages/ui/src/features/sessions/sideQuestionStore.test.ts @@ -0,0 +1,40 @@ +import { beforeEach, describe, expect, it } from "vitest"; +import { useSideQuestionStore } from "./sideQuestionStore"; + +describe("sideQuestionStore stale-answer guard", () => { + beforeEach(() => { + useSideQuestionStore.setState({ byTaskId: {} }); + }); + + it("does not repopulate a dismissed entry with a late answer", () => { + const { ask, dismiss, resolve } = useSideQuestionStore.getState(); + + const id = ask("task-1", "run-1", "what changed?"); + dismiss("task-1"); + resolve("task-1", "run-1", id, "the late answer"); + + expect(useSideQuestionStore.getState().byTaskId["task-1"]).toBeUndefined(); + }); + + it("does not let a stale answer clobber a re-asked question", () => { + const { ask, resolve } = useSideQuestionStore.getState(); + + const firstId = ask("task-1", "run-1", "first question?"); + const secondId = ask("task-1", "run-1", "second question?"); + resolve("task-1", "run-1", firstId, "answer to the first question"); + + const entry = useSideQuestionStore.getState().byTaskId["task-1"]; + expect(entry?.id).toBe(secondId); + expect(entry?.status).toBe("pending"); + }); + + it("does not settle an answer against a run the task has since moved on from", () => { + const { ask, resolve } = useSideQuestionStore.getState(); + + const id = ask("task-1", "run-1", "what changed?"); + resolve("task-1", "run-2", id, "answer from the old run"); + + const entry = useSideQuestionStore.getState().byTaskId["task-1"]; + expect(entry?.status).toBe("pending"); + }); +}); diff --git a/packages/ui/src/features/sessions/sideQuestionStore.ts b/packages/ui/src/features/sessions/sideQuestionStore.ts new file mode 100644 index 0000000000..a308e9763e --- /dev/null +++ b/packages/ui/src/features/sessions/sideQuestionStore.ts @@ -0,0 +1,115 @@ +import type { SessionService } from "@posthog/core/sessions/sessionService"; +import { getErrorMessage } from "@posthog/shared"; +import { logger } from "@posthog/ui/shell/logger"; +import { create } from "zustand"; + +const log = logger.scope("side-question"); + +export type SideQuestionEntry = { + id: string; + question: string; + /** The run this question was asked against — a card must not outlive it. */ + taskRunId: string; +} & ( + | { status: "pending" } + | { status: "done"; answer: string } + | { status: "error"; error: string } +); + +interface SideQuestionState { + /** Latest side question per task. Ephemeral: never persisted, never part of session history. */ + byTaskId: Record; + ask: (taskId: string, taskRunId: string, question: string) => string; + resolve: ( + taskId: string, + taskRunId: string, + id: string, + answer: string, + ) => void; + fail: (taskId: string, taskRunId: string, id: string, error: string) => void; + dismiss: (taskId: string) => void; +} + +/** + * Settles the entry only if it is still the one the caller created for the + * same run — a re-asked or dismissed question, or a run that reconnected or + * restarted in the meantime, must not receive a stale answer. + */ +function settle( + state: SideQuestionState, + taskId: string, + taskRunId: string, + id: string, + outcome: + | { status: "done"; answer: string } + | { status: "error"; error: string }, +): Partial | SideQuestionState { + const entry = state.byTaskId[taskId]; + if (entry?.id !== id || entry.taskRunId !== taskRunId) return state; + return { + byTaskId: { + ...state.byTaskId, + [taskId]: { + id: entry.id, + question: entry.question, + taskRunId, + ...outcome, + }, + }, + }; +} + +export const useSideQuestionStore = create()((set) => ({ + byTaskId: {}, + ask: (taskId, taskRunId, question) => { + const id = crypto.randomUUID(); + set((state) => ({ + byTaskId: { + ...state.byTaskId, + [taskId]: { id, question, taskRunId, status: "pending" }, + }, + })); + return id; + }, + resolve: (taskId, taskRunId, id, answer) => + set((state) => + settle(state, taskId, taskRunId, id, { status: "done", answer }), + ), + fail: (taskId, taskRunId, id, error) => + set((state) => + settle(state, taskId, taskRunId, id, { status: "error", error }), + ), + dismiss: (taskId) => + set((state) => { + const { [taskId]: _dismissed, ...rest } = state.byTaskId; + return { byTaskId: rest }; + }), +})); + +/** + * Fire-and-forget "/btw" saga: record the pending entry, send the question, + * and settle the entry with the answer or error. Fire-and-forget so the + * composer clears immediately; the side question runs beside the + * conversation, mid-turn or idle. + */ +export function fireSideQuestion( + sessionService: Pick, + taskId: string, + taskRunId: string, + question: string, +): void { + const { ask, resolve, fail } = useSideQuestionStore.getState(); + const id = ask(taskId, taskRunId, question); + sessionService + .askSideQuestion(taskId, question) + .then((answer) => resolve(taskId, taskRunId, id, answer)) + .catch((error) => { + fail( + taskId, + taskRunId, + id, + getErrorMessage(error) || "Side question failed", + ); + log.error("Side question failed", error); + }); +} diff --git a/packages/workspace-server/src/services/agent/agent.test.ts b/packages/workspace-server/src/services/agent/agent.test.ts index 293ac3854e..e3e436ecad 100644 --- a/packages/workspace-server/src/services/agent/agent.test.ts +++ b/packages/workspace-server/src/services/agent/agent.test.ts @@ -97,6 +97,10 @@ vi.mock("@agentclientprotocol/sdk", () => ({ vi.mock("@posthog/agent", () => ({ isMcpToolReadOnly: vi.fn(() => false), + POSTHOG_METHODS: { + SIDE_QUESTION: "_posthog/side_question", + REFRESH_SESSION: "_posthog/refresh_session", + }, })); vi.mock("@posthog/agent/posthog-api", () => ({ @@ -561,6 +565,7 @@ describe("AgentService", () => { config: {}, promptPending: false, inFlightMcpToolCalls: new Map(), + pendingSideQuestions: 0, mcpToolApprovals: {}, toolInstallations: {}, ...overrides, @@ -699,6 +704,31 @@ describe("AgentService", () => { ); }); + it("reschedules when pendingSideQuestions is non-zero at timeout", () => { + injectSession(service, "run-1", { pendingSideQuestions: 1 }); + service.recordActivity("run-1"); + + vi.advanceTimersByTime(15 * 60 * 1000); + + expect(service.emit).not.toHaveBeenCalledWith( + "session-idle-killed", + expect.anything(), + ); + expect(getIdleTimeouts(service).has("run-1")).toBe(true); + }); + + it("kills session when pendingSideQuestions is zero", () => { + injectSession(service, "run-1", { pendingSideQuestions: 0 }); + service.recordActivity("run-1"); + + vi.advanceTimersByTime(15 * 60 * 1000); + + expect(service.emit).toHaveBeenCalledWith( + "session-idle-killed", + expect.objectContaining({ taskRunId: "run-1" }), + ); + }); + it("checkIdleDeadlines does not kill non-expired sessions", () => { injectSession(service, "run-1"); service.recordActivity("run-1"); @@ -717,6 +747,123 @@ describe("AgentService", () => { }); }); + describe("sideQuestion", () => { + function injectSession( + svc: AgentService, + taskRunId: string, + overrides: Record = {}, + ) { + const sessions = (svc as unknown as { sessions: Map }) + .sessions; + sessions.set(taskRunId, { + taskRunId, + taskId: `task-for-${taskRunId}`, + repoPath: "/mock/repo", + agent: { cleanup: vi.fn().mockResolvedValue(undefined) }, + clientSideConnection: { extMethod: vi.fn() }, + channel: `ch-${taskRunId}`, + createdAt: Date.now(), + lastActivityAt: 0, + config: { sessionId: "agent-session-1" }, + promptPending: false, + inFlightMcpToolCalls: new Map(), + pendingSideQuestions: 0, + mcpToolApprovals: {}, + toolInstallations: {}, + ...overrides, + }); + } + + it("throws when no session exists for the given id", async () => { + await expect(service.sideQuestion("unknown-run", "why?")).rejects.toThrow( + "Session not found: unknown-run", + ); + }); + + it("forwards the question to the adapter and returns the parsed answer", async () => { + const extMethod = vi.fn().mockResolvedValue({ answer: "42" }); + injectSession(service, "run-1", { + clientSideConnection: { extMethod }, + }); + + const result = await service.sideQuestion("run-1", "why?"); + + expect(result).toEqual({ answer: "42" }); + expect(extMethod).toHaveBeenCalledWith("_posthog/side_question", { + sessionId: "agent-session-1", + question: "why?", + }); + }); + + it("records activity on the session", async () => { + const extMethod = vi.fn().mockResolvedValue({ answer: "42" }); + injectSession(service, "run-1", { + clientSideConnection: { extMethod }, + lastActivityAt: 0, + }); + const recordActivitySpy = vi.spyOn(service, "recordActivity"); + + await service.sideQuestion("run-1", "why?"); + + expect(recordActivitySpy).toHaveBeenCalledWith("run-1"); + const sessions = ( + service as unknown as { + sessions: Map; + } + ).sessions; + expect(sessions.get("run-1")?.lastActivityAt).toBeGreaterThan(0); + }); + + it("throws when the adapter returns a malformed result", async () => { + const extMethod = vi.fn().mockResolvedValue({ notAnAnswer: true }); + injectSession(service, "run-1", { + clientSideConnection: { extMethod }, + }); + + await expect(service.sideQuestion("run-1", "why?")).rejects.toThrow(); + }); + + it("tracks pendingSideQuestions while the extension call is in flight", async () => { + let resolveExtMethod: (value: { answer: string }) => void = () => {}; + const extMethod = vi.fn( + () => + new Promise((resolve) => { + resolveExtMethod = resolve; + }), + ); + injectSession(service, "run-1", { clientSideConnection: { extMethod } }); + const sessions = ( + service as unknown as { + sessions: Map; + } + ).sessions; + + const promise = service.sideQuestion("run-1", "why?"); + expect(sessions.get("run-1")?.pendingSideQuestions).toBe(1); + + resolveExtMethod({ answer: "42" }); + await promise; + + expect(sessions.get("run-1")?.pendingSideQuestions).toBe(0); + }); + + it("decrements pendingSideQuestions when the extension call fails", async () => { + const extMethod = vi.fn().mockRejectedValue(new Error("boom")); + injectSession(service, "run-1", { clientSideConnection: { extMethod } }); + const sessions = ( + service as unknown as { + sessions: Map; + } + ).sessions; + + await expect(service.sideQuestion("run-1", "why?")).rejects.toThrow( + "boom", + ); + + expect(sessions.get("run-1")?.pendingSideQuestions).toBe(0); + }); + }); + describe("channel system prompt local folders", () => { const credentials = { apiHost: "https://app.posthog.com", projectId: 1 }; const FOLDERS_HEADER = diff --git a/packages/workspace-server/src/services/agent/agent.ts b/packages/workspace-server/src/services/agent/agent.ts index a352357e3b..24be12e844 100644 --- a/packages/workspace-server/src/services/agent/agent.ts +++ b/packages/workspace-server/src/services/agent/agent.ts @@ -16,6 +16,7 @@ import { detectRtkBinary, isMcpToolReadOnly, isNotification, + POSTHOG_METHODS, POSTHOG_NOTIFICATIONS, } from "@posthog/agent"; import type { McpToolApprovals } from "@posthog/agent/adapters/claude/mcp/tool-metadata"; @@ -114,7 +115,9 @@ import { type ReconnectSessionInput, type RtkStatus, type SessionResponse, + type SideQuestionOutput, type StartSessionInput, + sideQuestionOutput, } from "./schemas"; export type { InterruptReason }; @@ -292,14 +295,24 @@ interface SessionConfig { spokenNarration?: boolean; } -/** Pull the adapter's `agentCapabilities._meta.posthog.steering` from initialize. */ -function extractSteeringCapability(init: unknown): string | undefined { - const steering = ( +/** Pull the adapter's negotiated `agentCapabilities._meta.posthog` capabilities from initialize. */ +function extractPosthogCapabilities(init: unknown): { + steering?: string; + sideQuestion?: boolean; +} { + const posthog = ( init as { - agentCapabilities?: { _meta?: { posthog?: { steering?: unknown } } }; + agentCapabilities?: { _meta?: { posthog?: Record } }; } - )?.agentCapabilities?._meta?.posthog?.steering; - return typeof steering === "string" ? steering : undefined; + )?.agentCapabilities?._meta?.posthog; + return { + steering: + typeof posthog?.steering === "string" ? posthog.steering : undefined, + sideQuestion: + typeof posthog?.sideQuestion === "boolean" + ? posthog.sideQuestion + : undefined, + }; } interface ManagedSession { @@ -318,8 +331,12 @@ interface ManagedSession { configOptions?: SessionConfigOption[]; /** Adapter's negotiated steering capability from initialize (`_meta.posthog.steering`). */ steering?: string; + /** Adapter's negotiated side-question capability from initialize (`_meta.posthog.sideQuestion`). */ + sideQuestion?: boolean; /** Tracks in-flight MCP tool calls (toolCallId → toolKey) for cancellation */ inFlightMcpToolCalls: Map; + /** Count of "/btw" side questions awaiting a response, so the idle timer does not reap the session mid-answer. */ + pendingSideQuestions: number; /** MCP tool approval states fetched at session start */ mcpToolApprovals: McpToolApprovals; /** Maps tool keys to their installation for backend approval updates */ @@ -575,7 +592,11 @@ export class AgentService extends TypedEventEmitter { private killIdleSession(taskRunId: string): void { const session = this.sessions.get(taskRunId); if (!session) return; - if (session.promptPending || session.inFlightMcpToolCalls.size > 0) { + if ( + session.promptPending || + session.inFlightMcpToolCalls.size > 0 || + session.pendingSideQuestions > 0 + ) { this.recordActivity(taskRunId); return; } @@ -942,10 +963,11 @@ If a repository IS genuinely required, attach one in this priority order: }, }); // The adapter advertises whether mid-turn steering folds natively into the - // running turn (`steering: "native"`) vs needs cancel+resend. Surface it so - // the host gates steer-vs-resend on the negotiated capability, not on a - // hardcoded adapter name (codex-acp advertises "interrupt-resend"). - const steering = extractSteeringCapability(initResult); + // running turn (`steering: "native"`) vs needs cancel+resend, and whether + // it can answer one-shot "/btw" side questions. Surface both so the host + // gates on the negotiated capabilities, not on a hardcoded adapter name + // (codex-acp advertises "interrupt-resend"). + const { steering, sideQuestion } = extractPosthogCapabilities(initResult); const { servers: mcpServers, @@ -1161,7 +1183,9 @@ If a repository IS genuinely required, attach one in this priority order: promptPending: false, configOptions, steering, + sideQuestion, inFlightMcpToolCalls: new Map(), + pendingSideQuestions: 0, mcpToolApprovals: toolApprovals, toolInstallations, evaluatedPrUrls: new Set(), @@ -1377,6 +1401,42 @@ If a repository IS genuinely required, attach one in this priority order: } } + /** + * Answers a one-shot "/btw" side question via the adapter's SIDE_QUESTION + * extension method. Never touches promptPending and never becomes part of + * the conversation; it does count as activity (resets the idle-kill timer), + * the same way refreshSession does. The exchange runs beside the + * conversation (ACP JSON-RPC multiplexes, so this works mid-turn). + * + * `pendingSideQuestions` keeps the session alive if the idle timer fires + * while the extension call is still awaited — otherwise `killIdleSession` + * would see no pending prompt and no in-flight tool call and clean up the + * session out from under this request. + */ + async sideQuestion( + sessionId: string, + question: string, + ): Promise { + const session = this.sessions.get(sessionId); + if (!session) { + throw new Error(`Session not found: ${sessionId}`); + } + + session.lastActivityAt = Date.now(); + this.recordActivity(sessionId); + session.pendingSideQuestions++; + + try { + const result = await session.clientSideConnection.extMethod( + POSTHOG_METHODS.SIDE_QUESTION, + { sessionId: getAgentSessionId(session), question }, + ); + return sideQuestionOutput.parse(result); + } finally { + session.pendingSideQuestions--; + } + } + async cancelSession(sessionId: string): Promise { const session = this.sessions.get(sessionId); if (!session) return false; @@ -2070,6 +2130,7 @@ For git operations while detached: channel: session.channel, configOptions: session.configOptions, steering: session.steering, + sideQuestion: session.sideQuestion, }; } diff --git a/packages/workspace-server/src/services/agent/schemas.ts b/packages/workspace-server/src/services/agent/schemas.ts index 5d8eeab508..a86ed0914a 100644 --- a/packages/workspace-server/src/services/agent/schemas.ts +++ b/packages/workspace-server/src/services/agent/schemas.ts @@ -164,6 +164,10 @@ export const sessionResponseSchema = z.object({ // running turn; "interrupt-resend" (legacy) or absent means the host must // cancel + resend instead. Drives the host's steer-vs-resend decision. steering: z.string().optional(), + // The adapter's negotiated side-question capability from initialize + // (`_meta.posthog.sideQuestion`): true means the adapter can answer a + // one-shot "/btw" question forked off the live transcript. + sideQuestion: z.boolean().optional(), }); export type SessionResponse = z.infer; @@ -194,6 +198,20 @@ export const promptOutput = z.object({ export type PromptOutput = z.infer; +// Side question ("/btw") input/output +export const sideQuestionInput = z.object({ + sessionId: z.string(), + question: z.string().min(1), +}); + +export type SideQuestionInput = z.infer; + +export const sideQuestionOutput = z.object({ + answer: z.string(), +}); + +export type SideQuestionOutput = z.infer; + // Cancel session input export const cancelSessionInput = z.object({ sessionId: z.string(),