diff --git a/apps/example/server.ts b/apps/example/server.ts index 5d5d2adce8..e855d49a57 100644 --- a/apps/example/server.ts +++ b/apps/example/server.ts @@ -17,6 +17,7 @@ const [ ]); const app = await createApp({ + experimental: { acp: process.env.NODE_ENV === "development" }, dashboard: { authRequired: exampleDashboardAuthRequired(), allowedGoogleDomains: ["sentry.io"], diff --git a/package.json b/package.json index 752f66bbc7..0ef6116908 100644 --- a/package.json +++ b/package.json @@ -3,6 +3,7 @@ "private": true, "packageManager": "pnpm@11.8.0+sha512.c1f5e7c4cb241c8f174b743851d82f42b802324afc8b0f116b96adb15aa06664948dde36960a3ba1079ba5b4b29dd0140135b94b5b5f5263592249d68e555f26", "scripts": { + "acp:local": "node scripts/acp-local.mjs", "dev": "node scripts/dev-server.mjs", "dev:env": "pnpx vercel env pull .env.local --environment=development && pnpm run cloudflare:token", "cli": "node scripts/cli-with-root-env.mjs", diff --git a/packages/docs/src/content/docs/reference/config-and-env.md b/packages/docs/src/content/docs/reference/config-and-env.md index f2baa8b161..0a3436918f 100644 --- a/packages/docs/src/content/docs/reference/config-and-env.md +++ b/packages/docs/src/content/docs/reference/config-and-env.md @@ -42,8 +42,8 @@ related: | `CRON_SECRET` or `JUNIOR_SCHEDULER_SECRET` | Conditional | Bearer token for the internal heartbeat route; use `CRON_SECRET` with Vercel Cron, or `JUNIOR_SCHEDULER_SECRET` for a non-Vercel heartbeat caller. | | `JUNIOR_TIMEZONE` | No | Default IANA timezone for scheduler authoring when the scheduler plugin is enabled. Defaults to `America/Los_Angeles`. | | `AI_GATEWAY_API_KEY` | No | Fallback AI Gateway auth when Vercel OIDC is unavailable (local/CI/non-Vercel hosts). On Vercel, prefer project OIDC so usage attributes to the project. | -| `BLOB_STORE_ID` | Conditional | Vercel Blob store for durable conversation attachments and published public artifacts. Vercel sets this when an OIDC-enabled Blob store is connected to the project. | -| `BLOB_READ_WRITE_TOKEN` | Conditional | Static Vercel Blob credential when OIDC is unavailable. Vercel sets this for a token-connected store. | +| `BLOB_STORE_ID` | Conditional | Vercel Blob store for durable conversation attachments and published public artifacts. Vercel sets this when an OIDC-enabled Blob store is connected to the project. | +| `BLOB_READ_WRITE_TOKEN` | Conditional | Static Vercel Blob credential when OIDC is unavailable. Vercel sets this for a token-connected store. | For Vercel deployments, create a private Blob store and connect it to the project before using `sendFiles` or `publishImage`. Prefer an OIDC connection. @@ -139,6 +139,8 @@ import { createApp } from "@sentry/junior"; const app = await createApp({ experimental: { + // ACP v1 Streamable HTTP for one-process development and testing. + acp: true, // Model-facing spawnAgent for durable child agent work. Incomplete; keep off // unless you are testing the #879 runtime. subagents: true, @@ -149,6 +151,12 @@ const app = await createApp({ `junior chat` enables experimental `subagents` automatically because it is the local createApp-equivalent entrypoint and already wires the child-worker path. +`acp` mounts `GET`, `POST`, and `DELETE /api/acp`. Every request needs a Junior +personal token in the bearer authorization header. The current transport keeps +connection state in one Node process. Use it only for local or single-process +testing. Run `pnpm acp:local` in this repository for a loopback test with the +official ACP SDK client. + ## Install-wide config defaults Pass `configDefaults` to `createApp()` to set provider defaults across all conversations: diff --git a/packages/docs/src/content/docs/reference/handler-surface.md b/packages/docs/src/content/docs/reference/handler-surface.md index 24c76a9f1b..0eddc2f625 100644 --- a/packages/docs/src/content/docs/reference/handler-surface.md +++ b/packages/docs/src/content/docs/reference/handler-surface.md @@ -29,6 +29,12 @@ Handled `POST` routes: - `/api/internal/plugin/tasks` - `/api/webhooks/:platform` (Slack path is `/api/webhooks/slack`) +When `createApp({ experimental: { acp: true } })` is set, `GET`, `POST`, and +`DELETE /api/acp` expose ACP v1 Streamable HTTP. Every request requires a Junior +personal token in the bearer authorization header. This experimental route +keeps connection state in one Node process. Do not enable it on a multi-process +deployment. + ## Expected behavior - Unknown routes return `404`. diff --git a/packages/junior/package.json b/packages/junior/package.json index 230006f592..856f9a584f 100644 --- a/packages/junior/package.json +++ b/packages/junior/package.json @@ -55,6 +55,7 @@ "prepare": "pnpm run build", "prepack": "pnpm run build", "build": "tsup && tsc -p tsconfig.build.json --emitDeclarationOnly", + "acp:smoke": "pnpm exec tsx scripts/acp-smoke.ts", "db:generate": "pnpm exec drizzle-kit generate --config drizzle.config.ts", "lint": "oxlint --config .oxlintrc.json --deny-warnings src tests scripts bin tsup.config.ts && depcruise --config .dependency-cruiser.mjs src/chat", "lint:fix": "oxlint --config .oxlintrc.json --deny-warnings --fix src tests scripts bin tsup.config.ts", @@ -66,6 +67,7 @@ "test:coverage": "vitest run --maxWorkers=4 --coverage --reporter=default --reporter=junit --outputFile.junit=coverage/results.junit.xml" }, "dependencies": { + "@agentclientprotocol/sdk": "1.3.0", "@ai-sdk/gateway": "^3.0.119", "@chat-adapter/slack": "4.29.0", "@chat-adapter/state-memory": "4.29.0", @@ -101,6 +103,7 @@ "zod": "catalog:" }, "devDependencies": { + "@hono/node-server": "1.19.14", "@emnapi/core": "^1.10.0", "@emnapi/runtime": "^1.10.0", "@sentry/junior-github": "workspace:*", diff --git a/packages/junior/scripts/acp-local-server.ts b/packages/junior/scripts/acp-local-server.ts new file mode 100644 index 0000000000..aeb7042436 --- /dev/null +++ b/packages/junior/scripts/acp-local-server.ts @@ -0,0 +1,136 @@ +/** + * Serve one loopback ACP process, run the official client, and clean up its + * short-lived personal token. This is test equipment, not a product transport. + */ +import { spawn } from "node:child_process"; +import { once } from "node:events"; +import type { AddressInfo } from "node:net"; +import { serve } from "@hono/node-server"; +import { createApp } from "@/app"; +import { migrateSchema } from "@/chat/conversations/sql/migrations"; +import { getSqlExecutor } from "@/chat/db"; +import { + createPersonalToken, + revokePersonalToken, +} from "@/personal-tokens/store"; +import { + closeApiTurnWorkFixture, + createConversationWorkWebHarness, +} from "../tests/fixtures/api-turn"; +import { streamScript } from "../tests/fixtures/conversation-work"; + +const DEFAULT_PORT = 3099; +const DEFAULT_REPLY = "Local Junior ACP completed this Turn."; + +function localPort(): number { + const raw = process.env.JUNIOR_ACP_LOCAL_PORT?.trim(); + if (!raw) return DEFAULT_PORT; + const port = Number.parseInt(raw, 10); + if (!Number.isSafeInteger(port) || port < 0 || port > 65_535) { + throw new Error("JUNIOR_ACP_LOCAL_PORT must be an integer from 0 to 65535"); + } + return port; +} + +await migrateSchema(getSqlExecutor()); +const harness = await createConversationWorkWebHarness({ + modelStream: streamScript( + process.env.JUNIOR_ACP_LOCAL_REPLY?.trim() || DEFAULT_REPLY, + ), +}); +const app = await createApp({ + conversationWork: harness.conversationWork, + experimental: { acp: true, subagents: true }, +}); +let drainActive = false; + +/** Drain queued API Turn work while the smoke client waits for its response. */ +async function drainQueuedWork(): Promise { + if (drainActive || !harness.queue.hasQueuedMessages()) return; + drainActive = true; + try { + await harness.drain(); + } catch (error) { + console.error("Local ACP queue drain failed", error); + exitAfterShutdown(1); + } finally { + drainActive = false; + } +} + +const drainTimer = setInterval(() => void drainQueuedWork(), 10); +const server = serve({ + fetch: app.fetch, + hostname: "127.0.0.1", + port: localPort(), +}); +if (!server.listening) { + await once(server, "listening"); +} + +const token = await createPersonalToken({ + email: harness.actor.email, + name: "Local ACP test", +}); +const address = server.address() as AddressInfo; +const url = `http://127.0.0.1:${address.port}/api/acp`; +console.log(`Local ACP URL: ${url}`); + +let smoke: ReturnType | undefined; +let shutdownPromise: Promise | undefined; + +/** Stop the HTTP client and server, revoke the token, and close test adapters. */ +function shutdown(): Promise { + shutdownPromise ??= (async () => { + clearInterval(drainTimer); + if (smoke?.exitCode === null && smoke.signalCode === null) { + smoke.kill("SIGTERM"); + } + server.close(); + await once(server, "close"); + await revokePersonalToken({ email: harness.actor.email, id: token.id }); + await closeApiTurnWorkFixture(); + })(); + return shutdownPromise; +} + +/** Finish cleanup and exit from a terminal runtime edge. */ +function exitAfterShutdown(code: number): void { + void shutdown().then( + () => process.exit(code), + (error) => { + console.error("Local ACP shutdown failed", error); + process.exit(1); + }, + ); +} + +for (const signal of ["SIGINT", "SIGTERM"] as const) { + process.on(signal, () => { + exitAfterShutdown(signal === "SIGINT" ? 130 : 143); + }); +} + +console.log("Running the official SDK smoke client..."); +let smokeExitCode = 1; +try { + smoke = spawn(process.execPath, ["--import", "tsx", "scripts/acp-smoke.ts"], { + cwd: process.cwd(), + env: { + ...process.env, + JUNIOR_ACP_FOLLOW_UP: + process.env.JUNIOR_ACP_FOLLOW_UP?.trim() || "Send a follow-up.", + JUNIOR_ACP_TOKEN: token.token, + JUNIOR_ACP_URL: url, + }, + stdio: "inherit", + }); + const [code, signal] = await once(smoke, "exit"); + if (signal) { + throw new Error(`Local ACP smoke client stopped with ${signal}`); + } + smokeExitCode = code ?? 1; +} finally { + await shutdown(); +} +if (smokeExitCode !== 0) process.exitCode = smokeExitCode; diff --git a/packages/junior/scripts/acp-smoke.ts b/packages/junior/scripts/acp-smoke.ts new file mode 100644 index 0000000000..423d52c554 --- /dev/null +++ b/packages/junior/scripts/acp-smoke.ts @@ -0,0 +1,100 @@ +import * as acp from "@agentclientprotocol/sdk"; +import { createHttpStream } from "@agentclientprotocol/sdk/experimental/http-client"; + +function requiredEnvironment(name: string): string { + const value = process.env[name]?.trim(); + if (!value) { + throw new Error(`${name} is required`); + } + return value; +} + +const url = requiredEnvironment("JUNIOR_ACP_URL"); +const token = requiredEnvironment("JUNIOR_ACP_TOKEN"); +const prompt = + process.env.JUNIOR_ACP_PROMPT?.trim() || + "Reply with a short confirmation that remote ACP works."; +const savedSessionId = process.env.JUNIOR_ACP_SESSION_ID?.trim(); +const followUp = process.env.JUNIOR_ACP_FOLLOW_UP?.trim(); + +async function withConnection( + run: (context: acp.ClientContext) => Promise, +): Promise { + const stream = createHttpStream(url, { + headers: { Authorization: `Bearer ${token}` }, + }); + try { + return await acp + .client({ name: "junior-acp-smoke" }) + .onNotification(acp.methods.client.session.update, (context) => { + const update = context.params.update; + if ( + (update.sessionUpdate === "user_message_chunk" || + update.sessionUpdate === "agent_message_chunk") && + update.content.type === "text" + ) { + process.stdout.write( + `[${update.sessionUpdate}] ${update.content.text}\n`, + ); + } + }) + .connectWith(stream, run); + } finally { + await stream.writable.close().catch(() => undefined); + } +} + +async function initialize(context: acp.ClientContext): Promise { + const result = await context.request(acp.methods.agent.initialize, { + protocolVersion: acp.PROTOCOL_VERSION, + clientCapabilities: {}, + clientInfo: { name: "junior-acp-smoke", version: "1" }, + }); + if (result.agentCapabilities?.loadSession !== true) { + throw new Error("Junior did not advertise session/load support"); + } +} + +const sessionId = await withConnection(async (context) => { + await initialize(context); + if (savedSessionId) { + await context.request(acp.methods.agent.session.load, { + sessionId: savedSessionId, + cwd: process.cwd(), + mcpServers: [], + }); + } + const activeSessionId = + savedSessionId ?? + ( + await context.request(acp.methods.agent.session.new, { + cwd: process.cwd(), + mcpServers: [], + }) + ).sessionId; + const result = await context.request(acp.methods.agent.session.prompt, { + sessionId: activeSessionId, + prompt: [{ type: "text", text: prompt }], + }); + process.stdout.write(`[stop] ${result.stopReason}\n`); + return activeSessionId; +}); + +process.stdout.write(`[session] ${sessionId}\n`); + +await withConnection(async (context) => { + await initialize(context); + await context.request(acp.methods.agent.session.load, { + sessionId, + cwd: process.cwd(), + mcpServers: [], + }); + process.stdout.write("[reconnect] load complete\n"); + if (followUp) { + const result = await context.request(acp.methods.agent.session.prompt, { + sessionId, + prompt: [{ type: "text", text: followUp }], + }); + process.stdout.write(`[follow-up stop] ${result.stopReason}\n`); + } +}); diff --git a/packages/junior/src/api/acp/README.md b/packages/junior/src/api/acp/README.md new file mode 100644 index 0000000000..3fac02191d --- /dev/null +++ b/packages/junior/src/api/acp/README.md @@ -0,0 +1,46 @@ +# Remote ACP + +Junior exposes ACP v1 Streamable HTTP at `/api/acp` when the app sets +`experimental: { acp: true }`. The route accepts `GET`, `POST`, and `DELETE`. +Every request needs a Junior personal token in an `Authorization: Bearer` +header. + +The adapter maps an ACP session to a private Conversation. It uses the existing +web Actor, API Turn mailbox, worker, event store, and Conversation access rules. +Client paths do not select the Junior sandbox. Client MCP servers, resource +links, media, filesystem callbacks, and terminal callbacks are not supported. +`session/cancel` stops the active Turn and returns the ACP `cancelled` stop +reason. + +The ACP SDK keeps connection state in the Node process. This prototype supports +one process only. Do not use it on a multi-process deployment until the +transport has proven affinity or the SDK provides a released distributed state +backend. Direct tests with T3 Code or Zed and agreed resource-link and tool +behavior are also required before promotion. + +Run the official-SDK smoke client against a single local process through the +existing tunnel: + +```sh +JUNIOR_ACP_URL=https://example.trycloudflare.com/api/acp \ +JUNIOR_ACP_TOKEN=jr_pat_example \ +JUNIOR_ACP_FOLLOW_UP="Send one follow-up reply." \ +pnpm --filter @sentry/junior acp:smoke +``` + +Set `JUNIOR_ACP_SESSION_ID` to load an earlier Conversation before the first +prompt. The client always reconnects once and loads the active session. It +prints the session id so it can be reused. + +## Local Validation + +Run `pnpm acp:local` from the repository root. The command starts the local +Postgres and Redis services, applies core migrations, and opens the real +`/api/acp` route on loopback. It creates a short-lived test token, runs the +official SDK smoke client with two Turns and one reconnect, revokes the token, +and exits. The token does not enter terminal output. The test server uses the +normal auth, Conversation, mailbox, worker, event, and replay paths. It replaces +only Vercel Queue transport and model generation with in-process test adapters. + +This command is test equipment. It does not add a local ACP transport to the +product. The Compose services stay available for later local tests. diff --git a/packages/junior/src/api/acp/route.ts b/packages/junior/src/api/acp/route.ts new file mode 100644 index 0000000000..de3a9b7d22 --- /dev/null +++ b/packages/junior/src/api/acp/route.ts @@ -0,0 +1,553 @@ +/** + * Own the remote ACP HTTP edge. + * + * Every request authenticates one Actor. Connections stay in this app process, + * and sessions map only to that Actor's private Conversations. This prototype + * accepts text prompts and durable replay. It does not accept client tools. + */ +import { randomUUID } from "node:crypto"; +import type { StateAdapter } from "chat"; +import type { User } from "@sentry/junior-plugin-api"; +import * as acp from "@agentclientprotocol/sdk"; +import { AcpServer } from "@agentclientprotocol/sdk/experimental/server"; +import { readConversationAccessFromSql } from "@/api/conversations/access"; +import { + apiTurnIdForMessage, + appendAndEnqueueApiConversationMessage, + recordApiConversationActivity, + webActorFromEmail, +} from "@/chat/api-turns/work"; +import type { WebActor } from "@/chat/actor"; +import type { ConversationEventStore } from "@/chat/conversations/history"; +import { projectConversationMessages } from "@/chat/conversations/message-projection"; +import type { ConversationStore } from "@/chat/conversations/store"; +import { + getConversationEventStore, + getConversationStore, + getDb, +} from "@/chat/db"; +import { resolveViewerUser } from "@/chat/plugins/viewer"; +import { logException, withSpan } from "@/chat/logging"; +import { sleep } from "@/chat/sleep"; +import type { ConversationWorkQueue } from "@/chat/task-execution/queue"; +import { ensureConversationWake } from "@/chat/task-execution/store"; +import type { ApiTurnCancellation } from "@/chat/api-turns/cancellation"; +import { authenticatePersonalToken } from "@/personal-tokens/store"; +import { JUNIOR_VERSION } from "@/version"; + +const ACP_CONNECTION_ID_HEADER = "Acp-Connection-Id"; +const ACP_CONVERSATION_PREFIX = "local:acp:"; +const EVENT_PAGE_SIZE = 50; +const EVENT_POLL_INTERVAL_MS = 25; +const MAX_PROMPT_TEXT_LENGTH = 32_000; + +interface AcpRouteOptions { + cancellation: ApiTurnCancellation; + conversationStore?: ConversationStore; + queue: ConversationWorkQueue; + state?: StateAdapter; +} + +interface AuthenticatedAcpActor { + actor: WebActor; + user: User; +} + +type AcpOperation = + | "initialize" + | "session_cancel" + | "session_load" + | "session_new" + | "session_prompt"; + +function bearerToken(request: Request): string | undefined { + const authorization = request.headers.get("Authorization"); + const match = authorization?.match(/^Bearer ([^\s]+)$/); + return match?.[1]; +} + +/** Resolve one personal token to the Actor and user that own the request. */ +async function authenticateRequest( + request: Request, +): Promise { + const token = bearerToken(request); + if (!token) return undefined; + const email = await authenticatePersonalToken(token); + if (!email) return undefined; + const user = await resolveViewerUser(email); + if (!user) return undefined; + return { + actor: webActorFromEmail( + user.email ?? email, + user.displayName ? { fullName: user.displayName } : undefined, + ), + user, + }; +} + +/** Reject client MCP servers because Junior does not use the client workspace. */ +function rejectUnsupportedMcpServers(mcpServers: readonly unknown[]): void { + if (mcpServers.length > 0) { + throw acp.RequestError.invalidParams( + { field: "mcpServers" }, + "Junior does not accept client MCP servers", + ); + } +} + +/** Require an ACP session that belongs to the authenticated participant. */ +async function requireOwnedSession( + sessionId: string, + user: User, +): Promise { + if (!sessionId.startsWith(ACP_CONVERSATION_PREFIX)) { + throw acp.RequestError.resourceNotFound(sessionId); + } + const access = ( + await readConversationAccessFromSql(getDb(), [sessionId], user) + ).get(sessionId); + if (!access?.isParticipant) { + throw acp.RequestError.resourceNotFound(sessionId); + } +} + +/** Convert supported ACP text blocks to one bounded API Turn message. */ +function promptText(prompt: readonly acp.ContentBlock[]): string { + if (prompt.length === 0) { + throw acp.RequestError.invalidParams( + { field: "prompt" }, + "Junior accepts one or more text blocks only", + ); + } + const blocks: string[] = []; + for (const block of prompt) { + if (block.type !== "text") { + throw acp.RequestError.invalidParams( + { field: "prompt" }, + "Junior accepts one or more text blocks only", + ); + } + if (!block.text.trim()) { + throw acp.RequestError.invalidParams( + { field: "prompt" }, + "Junior accepts non-empty text blocks only", + ); + } + blocks.push(block.text); + } + const text = blocks.join("\n"); + if (text.length > MAX_PROMPT_TEXT_LENGTH) { + throw acp.RequestError.invalidParams( + { field: "prompt" }, + `Junior accepts at most ${MAX_PROMPT_TEXT_LENGTH} prompt characters`, + ); + } + return text; +} + +/** Preserve the JSON-RPC id type in one connection-scoped mailbox key. */ +function requestIdKey(requestId: acp.JsonRpcId): string { + if (requestId === null) return "null"; + return `${typeof requestId}:${requestId}`; +} + +async function latestEventSeq( + eventStore: ConversationEventStore, + conversationId: string, +): Promise { + const page = await eventStore.query(conversationId, { limit: 1 }); + return page.events.at(-1)?.seq ?? 0; +} + +/** Replay durable user and assistant Messages in their stored order. */ +async function replaySession( + eventStore: ConversationEventStore, + sessionId: string, + client: acp.AgentContext, +): Promise { + const history = await eventStore.loadMessageHistory(sessionId); + for (const message of projectConversationMessages(history)) { + if (message.role === "system") continue; + await client.notify(acp.methods.client.session.update, { + sessionId, + update: { + sessionUpdate: + message.role === "user" + ? "user_message_chunk" + : "agent_message_chunk", + content: { type: "text", text: message.text }, + messageId: message.id, + }, + }); + } +} + +/** Stream durable assistant Messages until the matching Turn ends or fails. */ +async function waitForTurn(args: { + afterSeq: number; + cancellation: ApiTurnCancellation; + cancellationSignal: AbortSignal; + client: acp.AgentContext; + eventStore: ConversationEventStore; + sessionId: string; + signal: AbortSignal; + turnId: string; +}): Promise { + let cursor = args.afterSeq; + const assistantPrefix = `${args.turnId}:assistant:`; + const sentMessageIds = new Set(); + + while (true) { + if (args.signal.aborted) { + throw acp.RequestError.requestCancelled(); + } + const page = await args.eventStore.query(args.sessionId, { + afterSeq: cursor, + limit: EVENT_PAGE_SIZE, + types: ["message", "turn_completed", "turn_failed"], + }); + if (page.events.length === 0) { + try { + await sleep(EVENT_POLL_INTERVAL_MS, args.signal); + } catch (error) { + if (args.signal.aborted) { + throw acp.RequestError.requestCancelled(); + } + throw error; + } + continue; + } + + for (const event of page.events) { + cursor = event.seq; + const data = event.data; + if ( + data.type === "message" && + data.role === "assistant" && + data.messageId.startsWith(assistantPrefix) && + !sentMessageIds.has(data.messageId) + ) { + sentMessageIds.add(data.messageId); + await args.client.notify(acp.methods.client.session.update, { + sessionId: args.sessionId, + update: { + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: data.text }, + messageId: data.messageId, + }, + }); + continue; + } + if (data.type === "turn_completed" && data.turnId === args.turnId) { + args.cancellation.finish(args.sessionId, args.cancellationSignal); + return { + stopReason: data.outcome === "cancelled" ? "cancelled" : "end_turn", + }; + } + if (data.type === "turn_failed" && data.turnId === args.turnId) { + args.cancellation.finish(args.sessionId, args.cancellationSignal); + throw acp.RequestError.internalError( + { failureCode: data.failureCode }, + "Junior Turn failed", + ); + } + } + } +} + +/** Trace one protocol operation and capture only unexpected edge failures. */ +async function runAcpOperation( + authenticated: AuthenticatedAcpActor, + operation: AcpOperation, + conversationId: string | undefined, + callback: () => Promise | T, +): Promise { + const name = `acp.${operation}`; + return await withSpan( + name, + name, + { + actorId: authenticated.actor.userId, + conversationId, + platform: "acp", + userId: authenticated.actor.userId, + }, + async () => { + try { + return await callback(); + } catch (error) { + if (!(error instanceof acp.RequestError)) { + logException(error, `${name}.failed`); + } + throw error; + } + }, + ); +} + +/** Build the ACP Agent whose session handlers run as one authenticated Actor. */ +function createActorAgent( + authenticated: AuthenticatedAcpActor, + connectionNonce: string, + options: AcpRouteOptions, +) { + const conversationStore = options.conversationStore ?? getConversationStore(); + const eventStore = getConversationEventStore(); + + return acp + .agent({ name: "junior" }) + .onRequest(acp.methods.agent.initialize, () => + runAcpOperation(authenticated, "initialize", undefined, () => ({ + protocolVersion: acp.PROTOCOL_VERSION, + agentCapabilities: { + loadSession: true, + promptCapabilities: { + image: false, + audio: false, + embeddedContext: false, + }, + }, + authMethods: [], + agentInfo: { name: "junior", version: JUNIOR_VERSION }, + })), + ) + .onRequest(acp.methods.agent.session.new, async (context) => { + return await runAcpOperation( + authenticated, + "session_new", + undefined, + async () => { + rejectUnsupportedMcpServers(context.params.mcpServers); + const sessionId = `${ACP_CONVERSATION_PREFIX}${randomUUID()}`; + await recordApiConversationActivity({ + actor: authenticated.actor, + conversationId: sessionId, + conversationStore, + nowMs: Date.now(), + rootVisibility: "private", + }); + return { sessionId }; + }, + ); + }) + .onRequest(acp.methods.agent.session.load, async (context) => { + return await runAcpOperation( + authenticated, + "session_load", + context.params.sessionId, + async () => { + rejectUnsupportedMcpServers(context.params.mcpServers); + await requireOwnedSession( + context.params.sessionId, + authenticated.user, + ); + await replaySession( + eventStore, + context.params.sessionId, + context.client, + ); + return {}; + }, + ); + }) + .onRequest(acp.methods.agent.session.prompt, async (context) => { + return await runAcpOperation( + authenticated, + "session_prompt", + context.params.sessionId, + async () => { + await requireOwnedSession( + context.params.sessionId, + authenticated.user, + ); + const text = promptText(context.params.prompt); + const currentSeq = await latestEventSeq( + eventStore, + context.params.sessionId, + ); + const cancellationSignal = options.cancellation.begin( + context.params.sessionId, + ); + if (!cancellationSignal) { + throw acp.RequestError.invalidParams( + { field: "sessionId" }, + "This ACP session already has an active prompt", + ); + } + const recordDisconnect = () => + options.cancellation.disconnect( + context.params.sessionId, + cancellationSignal, + ); + if (context.signal.aborted) { + recordDisconnect(); + } else { + context.signal.addEventListener("abort", recordDisconnect, { + once: true, + }); + } + let accepted: Awaited< + ReturnType + >; + try { + accepted = await appendAndEnqueueApiConversationMessage( + { + actor: authenticated.actor, + conversationId: context.params.sessionId, + idempotencyKey: `${connectionNonce}:${requestIdKey(context.requestId)}`, + message: text, + }, + { + conversationStore, + queue: options.queue, + state: options.state, + }, + ); + } catch (error) { + options.cancellation.finish( + context.params.sessionId, + cancellationSignal, + ); + throw error; + } + // A duplicate request can refer to a Turn that ended before currentSeq. + const afterSeq = accepted.status === "duplicate" ? 0 : currentSeq; + return await waitForTurn({ + afterSeq, + cancellation: options.cancellation, + cancellationSignal, + client: context.client, + eventStore, + sessionId: context.params.sessionId, + signal: context.signal, + turnId: apiTurnIdForMessage(accepted.messageId), + }); + }, + ); + }) + .onNotification(acp.methods.agent.session.cancel, async (context) => { + await runAcpOperation( + authenticated, + "session_cancel", + context.params.sessionId, + async () => { + await requireOwnedSession( + context.params.sessionId, + authenticated.user, + ); + if (!options.cancellation.cancel(context.params.sessionId)) { + return; + } + const nowMs = Date.now(); + await ensureConversationWake({ + conversationId: context.params.sessionId, + conversationStore, + idempotencyKey: `acp-cancel:${context.params.sessionId}:${nowMs}`, + nowMs, + queue: options.queue, + replaceExistingWake: true, + state: options.state, + }); + }, + ); + }); +} + +function isJsonRpcId(value: unknown): value is acp.JsonRpcId { + return ( + value === null || + typeof value === "string" || + (typeof value === "number" && Number.isFinite(value)) + ); +} + +/** Reject object-shaped messages that would make the SDK log their raw body. */ +async function hasSafeAcpEnvelope(request: Request): Promise { + if (request.method !== "POST") return true; + let value: unknown; + try { + value = await request.clone().json(); + } catch { + return true; + } + if (typeof value !== "object" || value === null || Array.isArray(value)) { + return true; + } + if (!("jsonrpc" in value) || value.jsonrpc !== "2.0") { + return false; + } + if (!("method" in value) || typeof value.method !== "string") { + return false; + } + return !("id" in value) || isJsonRpcId(value.id); +} + +/** Return whether an SDK initialization response contains a protocol result. */ +async function initializationSucceeded(response: Response): Promise { + try { + const value: unknown = await response.clone().json(); + return typeof value === "object" && value !== null && "result" in value; + } catch { + return false; + } +} + +/** Create one app-scoped remote ACP v1 HTTP handler. */ +export function createAcpHttpHandler( + options: AcpRouteOptions, +): (request: Request) => Promise { + const server = new AcpServer({ agent: acp.agent({ name: "junior" }) }); + const connectionActors = new Map(); + + return async (request) => { + const authenticated = await authenticateRequest(request); + if (!authenticated) { + return new Response("Unauthorized", { status: 401 }); + } + + const connectionId = request.headers.get(ACP_CONNECTION_ID_HEADER); + const connectionActorId = connectionId + ? connectionActors.get(connectionId) + : undefined; + if (connectionId && connectionActorId === undefined) { + return new Response("Unknown Acp-Connection-Id", { status: 404 }); + } + if ( + connectionActorId !== undefined && + connectionActorId !== authenticated.actor.userId + ) { + return new Response("Unknown Acp-Connection-Id", { status: 404 }); + } + if (!(await hasSafeAcpEnvelope(request))) { + return new Response("Invalid JSON-RPC message", { status: 400 }); + } + + const response = await server.handleRequest(request, { + agent: createActorAgent(authenticated, randomUUID(), options), + }); + const responseConnectionId = response.headers.get(ACP_CONNECTION_ID_HEADER); + if (response.ok && responseConnectionId && !connectionId) { + if (await initializationSucceeded(response)) { + connectionActors.set(responseConnectionId, authenticated.actor.userId); + } else { + await server.handleRequest( + new Request(request.url, { + method: "DELETE", + headers: { [ACP_CONNECTION_ID_HEADER]: responseConnectionId }, + }), + ); + const headers = new Headers(response.headers); + headers.delete(ACP_CONNECTION_ID_HEADER); + return new Response(response.body, { + headers, + status: response.status, + statusText: response.statusText, + }); + } + } + if (request.method === "DELETE" && response.ok && connectionId) { + connectionActors.delete(connectionId); + } + return response; + }; +} diff --git a/packages/junior/src/app.ts b/packages/junior/src/app.ts index e8bd583cbe..979700c8c2 100644 --- a/packages/junior/src/app.ts +++ b/packages/junior/src/app.ts @@ -20,6 +20,7 @@ import { executeAgentRun } from "@/chat/agent"; import { normalizeSandboxEgressTracePropagationDomains } from "@/chat/sandbox/egress/tracing"; import { getExperimentalFeatures, + isExperimentalFeatureEnabled, setExperimentalFeatures, type ExperimentalFeaturesConfig, } from "@/chat/experimental"; @@ -70,7 +71,6 @@ import { JUNIOR_PLUGIN_TASK_CALLBACK_ROUTE } from "@/deployment"; import { createVercelConversationWorkCallback, registerVercelConversationWorkDevConsumer, - type VercelConversationWorkCallbackOptions, } from "@/chat/task-execution/vercel-callback"; import { getVercelConversationWorkQueue } from "@/chat/task-execution/vercel-queue"; import { bindSpawnAgent } from "@/chat/agent-invocations/spawn"; @@ -82,6 +82,7 @@ import { createProductionConversationWorkOptions, createProductionSlackWebhookServices, } from "@/chat/app/production"; +import type { ConversationWorkCallbackOptions } from "@/chat/app/conversation-work"; import { createAgentRunner } from "@/chat/runtime/agent-runner"; import { createVercelAttachmentStorage } from "@/chat/attachments/vercel"; import { publicArtifactGET } from "@/handlers/artifacts"; @@ -90,6 +91,7 @@ import { ingestResourceEvent } from "@/chat/resource-events/ingest"; import { createResourceEventTeamIdResolver } from "@/chat/resource-events/workspace"; import { ingestEventTasks } from "@/chat/event-tasks/ingest"; import { receiveLocalOAuthCredential } from "@/chat/local/credential-sync"; +import { createAcpHttpHandler } from "@/api/acp/route"; export { defineJuniorPlugins } from "./plugins"; export { JUNIOR_VERSION } from "./version"; @@ -118,7 +120,7 @@ export interface JuniorAppOptions { /** Install-wide provider defaults. Unregistered `provider.key` entries warn at startup. */ configDefaults?: Record; /** Queue consumer wiring for the durable conversation worker. */ - conversationWork?: VercelConversationWorkCallbackOptions; + conversationWork?: ConversationWorkCallbackOptions; /** Direct plugin set override. Usually omitted when `juniorNitro()` uses a plugin module. */ plugins?: JuniorPluginSet; /** Sandbox execution options. */ @@ -786,9 +788,7 @@ export async function createApp(options?: JuniorAppOptions): Promise { let pluginTaskPOST: | ReturnType | undefined; - let conversationWorkOptions: - | VercelConversationWorkCallbackOptions - | undefined; + let conversationWorkOptions: ConversationWorkCallbackOptions | undefined; const getConversationWorkOptions = () => { conversationWorkOptions ??= options?.conversationWork ?? @@ -798,6 +798,22 @@ export async function createApp(options?: JuniorAppOptions): Promise { }); return conversationWorkOptions; }; + if (isExperimentalFeatureEnabled("acp")) { + const work = getConversationWorkOptions(); + const cancellation = work.apiTurnCancellation; + if (!cancellation) { + throw new Error("Experimental ACP requires API Turn cancellation wiring"); + } + const handleAcpRequest = createAcpHttpHandler({ + cancellation, + conversationStore: work.conversationStore, + queue: work.queue ?? getVercelConversationWorkQueue(), + state: work.state, + }); + app.on(["GET", "POST", "DELETE"], "/api/acp", (c) => + handleAcpRequest(c.req.raw), + ); + } if (process.env.NODE_ENV === "development") { registerVercelConversationWorkDevConsumer(getConversationWorkOptions()); registerVercelPluginTaskDevConsumer(); diff --git a/packages/junior/src/chat/api-turns/cancellation.ts b/packages/junior/src/chat/api-turns/cancellation.ts new file mode 100644 index 0000000000..f7893b2af2 --- /dev/null +++ b/packages/junior/src/chat/api-turns/cancellation.ts @@ -0,0 +1,132 @@ +/** App-scoped control for one active API Turn per Conversation. */ +export interface ApiTurnCancellation { + begin(conversationId: string): AbortSignal | undefined; + cancel(conversationId: string): boolean; + disconnect(conversationId: string, signal: AbortSignal): void; + finish(conversationId: string, signal: AbortSignal): void; + park(conversationId: string, signal: AbortSignal): void; + signal(conversationId: string): AbortSignal | undefined; +} + +interface ActiveApiTurnCancellation { + connected: boolean; + controller: AbortController; + parked: boolean; +} + +/** Create in-process cancellation state for active API Turns. */ +export function createApiTurnCancellation(): ApiTurnCancellation { + const active = new Map(); + + return { + begin(conversationId) { + if (active.has(conversationId)) { + return undefined; + } + const controller = new AbortController(); + active.set(conversationId, { + connected: true, + controller, + parked: false, + }); + return controller.signal; + }, + cancel(conversationId) { + const entry = active.get(conversationId); + if (!entry) { + return false; + } + entry.controller.abort(new Error("API Turn cancelled")); + return true; + }, + disconnect(conversationId, signal) { + const entry = active.get(conversationId); + if (entry?.controller.signal !== signal) { + return; + } + entry.connected = false; + if (entry.parked) { + active.delete(conversationId); + } + }, + finish(conversationId, signal) { + const entry = active.get(conversationId); + if (entry?.controller.signal === signal) { + active.delete(conversationId); + } + }, + park(conversationId, signal) { + const entry = active.get(conversationId); + if (entry?.controller.signal !== signal) { + return; + } + entry.parked = true; + if (!entry.connected) { + active.delete(conversationId); + } + }, + signal(conversationId) { + return active.get(conversationId)?.controller.signal; + }, + }; +} + +/** Close one cancelled API Turn without a failure reply or retry. */ +export async function completeCancelledApiTurn(args: { + acknowledge(): Promise; + actorId: string; + cancellation?: ApiTurnCancellation; + conversation: ThreadConversationState; + conversationId: string; + lifecycle: ConversationTurnLifecycle; + sandboxRef?: SandboxRef; + signal?: AbortSignal; + turnId: string; + userMessageId: string; +}): Promise { + try { + await abandonTurnRecord({ + conversationId: args.conversationId, + turnId: args.turnId, + errorMessage: "API Turn cancelled", + }); + clearPendingAuth(args.conversation, args.turnId); + markConversationMessage(args.conversation, args.userMessageId, { + replied: false, + skippedReason: "turn cancelled", + }); + markTurnClosed({ + conversation: args.conversation, + nowMs: Date.now(), + sessionId: args.turnId, + }); + await deleteWebAuthorization({ + actorId: args.actorId, + conversationId: args.conversationId, + }); + await persistThreadStateById(args.conversationId, { + conversation: args.conversation, + sandboxRef: args.sandboxRef, + }); + await args.lifecycle.complete({ + conversationId: args.conversationId, + createdAtMs: Date.now(), + outcome: "cancelled", + turnId: args.turnId, + }); + } finally { + if (args.signal) { + args.cancellation?.finish(args.conversationId, args.signal); + } + } + await args.acknowledge(); +} +import type { ConversationTurnLifecycle } from "@/chat/conversations/turn-lifecycle"; +import { deleteWebAuthorization } from "@/chat/api-turns/authorization"; +import { persistThreadStateById } from "@/chat/runtime/thread-state"; +import { markTurnClosed } from "@/chat/runtime/turn"; +import type { SandboxRef } from "@/chat/sandbox/ref"; +import { markConversationMessage } from "@/chat/services/conversation-memory"; +import { clearPendingAuth } from "@/chat/services/pending-auth"; +import type { ThreadConversationState } from "@/chat/state/conversation"; +import { abandonTurnRecord } from "@/chat/task-execution/checkpoint"; diff --git a/packages/junior/src/chat/api-turns/routing.ts b/packages/junior/src/chat/api-turns/routing.ts new file mode 100644 index 0000000000..7be19345d8 --- /dev/null +++ b/packages/junior/src/chat/api-turns/routing.ts @@ -0,0 +1,110 @@ +import { z } from "zod"; +import { + getTurnRecord, + listTurnSummaries, +} from "@/chat/task-execution/checkpoint"; +import type { InboundMessage } from "@/chat/task-execution/store"; +import type { ConversationWorkerContext } from "@/chat/task-execution/worker"; + +const apiTurnMailboxMetadataSchema = z + .object({ + authorEmail: z.string().email(), + authorFullName: z.string().min(1).optional(), + authorUserId: z.string().min(1), + authorUserName: z.string().min(1).optional(), + kind: z.literal("api_turn"), + messageId: z.string().min(1), + }) + .strict(); + +export type ApiTurnMailboxMetadata = z.output< + typeof apiTurnMailboxMetadataSchema +>; + +function parseApiTurnMessages( + messages: readonly InboundMessage[], +): Array<{ message: InboundMessage; metadata: ApiTurnMailboxMetadata }> { + if (messages.length === 0) { + return []; + } + const parsed = messages.map((message) => ({ + message, + metadata: apiTurnMailboxMetadataSchema.safeParse(message.input.metadata), + })); + if (parsed.every((entry) => !entry.metadata.success)) { + return []; + } + if (parsed.some((entry) => !entry.metadata.success)) { + throw new Error("Conversation mailbox mixes web turns and other input"); + } + return parsed.map((entry) => { + if (!entry.metadata.success) { + throw new Error("API turn mailbox metadata failed validation"); + } + return { message: entry.message, metadata: entry.metadata.data }; + }); +} + +/** + * Resolve API Turn work from mailbox metadata or an active checkpoint. + * + * Empty resume wakes after yield carry no mailbox rows. Use durable active + * Turn state so these wakes do not fall through to Slack. + */ +export async function resolveApiTurnWork( + context: ConversationWorkerContext, +): Promise< + | { + kind: "mailbox"; + batch: Array<{ + message: InboundMessage; + metadata: ApiTurnMailboxMetadata; + }>; + } + | { kind: "resume"; turnId: string } + | undefined +> { + const batch = parseApiTurnMessages(context.attempt.messages); + if (batch.length > 0) { + return { kind: "mailbox", batch }; + } + if (context.attempt.messages.length > 0) { + return undefined; + } + + const summaries = await listTurnSummaries(context.conversationId); + // Agent dispatch also writes surface "api". Those Turns own a dispatchId + // and must stay on the dispatch router, which runs after this route. + const active = summaries.filter( + (summary) => + summary.surface === "api" && + !summary.dispatchId && + (summary.state === "paused" || summary.state === "running"), + ); + if (active.length > 1) { + throw new Error( + `Conversation ${context.conversationId} has multiple active web turns`, + ); + } + const turnId = active[0]?.turnId; + if (!turnId) { + return undefined; + } + const record = await getTurnRecord(context.conversationId, turnId); + if ( + !record || + record.surface !== "api" || + Boolean(record.dispatchId) || + (record.state !== "paused" && record.state !== "running") + ) { + return undefined; + } + return { kind: "resume", turnId }; +} + +/** Return whether the leased attempt belongs to an API Turn. */ +export async function isApiTurnWork( + context: ConversationWorkerContext, +): Promise { + return (await resolveApiTurnWork(context)) !== undefined; +} diff --git a/packages/junior/src/chat/api-turns/work.ts b/packages/junior/src/chat/api-turns/work.ts index c616963b7c..223340bae3 100644 --- a/packages/junior/src/chat/api-turns/work.ts +++ b/packages/junior/src/chat/api-turns/work.ts @@ -8,7 +8,6 @@ * location context and never publish back to Slack. */ import { createHash } from "node:crypto"; -import { z } from "zod"; import type { StateAdapter } from "chat"; import { createWebSource, @@ -89,19 +88,17 @@ import { createWebAuthorization, deleteWebAuthorization, } from "@/chat/api-turns/authorization"; +import { + completeCancelledApiTurn, + type ApiTurnCancellation, +} from "@/chat/api-turns/cancellation"; +import { + isApiTurnWork, + resolveApiTurnWork, + type ApiTurnMailboxMetadata, +} from "@/chat/api-turns/routing"; -const apiTurnMailboxMetadataSchema = z - .object({ - authorEmail: z.string().email(), - authorFullName: z.string().min(1).optional(), - authorUserId: z.string().min(1), - authorUserName: z.string().min(1).optional(), - kind: z.literal("api_turn"), - messageId: z.string().min(1), - }) - .strict(); - -type ApiTurnMailboxMetadata = z.output; +export { resolveApiTurnWork } from "@/chat/api-turns/routing"; type EnqueueOptions = { conversationStore?: ConversationStore; @@ -299,7 +296,8 @@ export function buildApiTurnInboundMessage(args: { }; } -async function recordApiConversationActivity(args: { +/** Record web activity and materialize a new API Conversation root when needed. */ +export async function recordApiConversationActivity(args: { actor: WebActor; conversationId: string; conversationStore?: ConversationStore; @@ -406,94 +404,6 @@ export async function appendAndEnqueueApiConversationMessage( }; } -function parseApiTurnMessages( - messages: readonly InboundMessage[], -): Array<{ message: InboundMessage; metadata: ApiTurnMailboxMetadata }> { - if (messages.length === 0) { - return []; - } - const parsed = messages.map((message) => ({ - message, - metadata: apiTurnMailboxMetadataSchema.safeParse(message.input.metadata), - })); - if (parsed.every((entry) => !entry.metadata.success)) { - return []; - } - if (parsed.some((entry) => !entry.metadata.success)) { - throw new Error("Conversation mailbox mixes web turns and other input"); - } - return parsed.map((entry) => { - if (!entry.metadata.success) { - throw new Error("API turn mailbox metadata failed validation"); - } - return { message: entry.message, metadata: entry.metadata.data }; - }); -} - -/** - * Resolve API turn work from mailbox metadata or an active API turn checkpoint. - * - * Empty resume wakes after yield carry no mailbox rows; match agent-invocation - * and look up durable active turn state instead of falling through to Slack. - */ -export async function resolveApiTurnWork( - context: ConversationWorkerContext, -): Promise< - | { - kind: "mailbox"; - batch: Array<{ - message: InboundMessage; - metadata: ApiTurnMailboxMetadata; - }>; - } - | { kind: "resume"; turnId: string } - | undefined -> { - const batch = parseApiTurnMessages(context.attempt.messages); - if (batch.length > 0) { - return { kind: "mailbox", batch }; - } - if (context.attempt.messages.length > 0) { - return undefined; - } - - const summaries = await listTurnSummaries(context.conversationId); - // Agent-dispatch also writes surface "api". Those turns own a dispatchId and - // must stay on the dispatch router (this route runs first). - const active = summaries.filter( - (summary) => - summary.surface === "api" && - !summary.dispatchId && - (summary.state === "paused" || summary.state === "running"), - ); - if (active.length > 1) { - throw new Error( - `Conversation ${context.conversationId} has multiple active web turns`, - ); - } - const turnId = active[0]?.turnId; - if (!turnId) { - return undefined; - } - const record = await getTurnRecord(context.conversationId, turnId); - if ( - !record || - record.surface !== "api" || - Boolean(record.dispatchId) || - (record.state !== "paused" && record.state !== "running") - ) { - return undefined; - } - return { kind: "resume", turnId }; -} - -/** True when this leased attempt is API-authored root work. */ -export async function isApiTurnWork( - context: ConversationWorkerContext, -): Promise { - return (await resolveApiTurnWork(context)) !== undefined; -} - function captureApiBoundaryFailure(args: { conversationId: string; error: unknown; @@ -509,9 +419,15 @@ function captureApiBoundaryFailure(args: { return typeof eventId === "string" ? eventId : undefined; } +function hasLostTurnInputCommit(error: unknown): boolean { + const cause = getConversationTurnBoundaryError(error)?.cause ?? error; + return isTurnInputCommitLostError(error) || isTurnInputCommitLostError(cause); +} + /** Build the mailbox consumer for API-authored root turns. */ export function createApiTurnWorker(options: { agentRunner: AgentRunner; + cancellation?: ApiTurnCancellation; turnLifecycle?: ConversationTurnLifecycle; }) { return async ( @@ -713,6 +629,41 @@ export function createApiTurnWorker(options: { let modelFailureEventId: string | undefined; let modelFailureCaptureAttempted = false; let reply: AgentRunResult | undefined; + const cancellationSignal = options.cancellation?.signal( + context.conversationId, + ); + const finishCancellation = (): void => { + if (cancellationSignal) { + options.cancellation?.finish( + context.conversationId, + cancellationSignal, + ); + } + }; + + const completeCancelledTurn = + async (): Promise => { + try { + await completeCancelledApiTurn({ + acknowledge, + actorId: actor.userId, + cancellation: options.cancellation, + conversation, + conversationId: context.conversationId, + lifecycle, + sandboxRef, + signal: cancellationSignal, + turnId, + userMessageId, + }); + } catch (error) { + if (hasLostTurnInputCommit(error)) { + return { status: "lost_lease" }; + } + throw error; + } + return { status: "completed" }; + }; const deliverAssistantMessage = async ( value: AssistantMessage | string, @@ -757,6 +708,9 @@ export function createApiTurnWorker(options: { await persistThreadStateById(context.conversationId, { conversation, }); + if (cancellationSignal?.aborted) { + return await completeCancelledTurn(); + } const piMessages = await loadProjection({ conversationId: context.conversationId, }); @@ -781,6 +735,7 @@ export function createApiTurnWorker(options: { publishExternally: false, source, surface: "api", + ...(cancellationSignal ? { signal: cancellationSignal } : {}), authorization: createWebAuthorization({ actorId: actor.userId, conversationId: context.conversationId, @@ -810,6 +765,10 @@ export function createApiTurnWorker(options: { }, }); + if (cancellationSignal?.aborted) { + return await completeCancelledTurn(); + } + if (outcome.status === "suspended") { return { status: "yielded" }; } @@ -825,9 +784,16 @@ export function createApiTurnWorker(options: { conversation, sandboxRef, }); + if (cancellationSignal) { + options.cancellation?.park( + context.conversationId, + cancellationSignal, + ); + } await acknowledge(); return { status: "completed" }; } + finishCancellation(); reply = outcome.result; modelFailureCaptureAttempted = reply.diagnostics.outcome !== "success"; @@ -922,13 +888,12 @@ export function createApiTurnWorker(options: { await acknowledge(); return { status: "completed" }; } catch (error) { - const cause = getConversationTurnBoundaryError(error)?.cause ?? error; - if ( - isTurnInputCommitLostError(error) || - isTurnInputCommitLostError(cause) - ) { + if (hasLostTurnInputCommit(error)) { return { status: "lost_lease" }; } + if (cancellationSignal?.aborted) { + return await completeCancelledTurn(); + } if (!context.attempt.isFinalAttempt) { throw error; } @@ -971,6 +936,7 @@ export function createApiTurnWorker(options: { failureCode, turnId, }); + finishCancellation(); await acknowledge(); return { status: "completed" }; } diff --git a/packages/junior/src/chat/app/conversation-work.ts b/packages/junior/src/chat/app/conversation-work.ts index 2c236214a4..09898f19de 100644 --- a/packages/junior/src/chat/app/conversation-work.ts +++ b/packages/junior/src/chat/app/conversation-work.ts @@ -19,10 +19,11 @@ import { createAgentInvocationWorker, routeAgentInvocationWork, } from "@/chat/agent-invocations/work"; +import { createApiTurnWorker, routeApiTurnWork } from "@/chat/api-turns/work"; import { - createApiTurnWorker, - routeApiTurnWork, -} from "@/chat/api-turns/work"; + createApiTurnCancellation, + type ApiTurnCancellation, +} from "@/chat/api-turns/cancellation"; import { getDispatchConversationId, getDispatchInputMessageIds, @@ -39,15 +40,22 @@ interface ConversationWorkOptions { state?: StateAdapter; } +export type ConversationWorkCallbackOptions = + VercelConversationWorkCallbackOptions & { + /** App-scoped control required by the experimental ACP route. */ + apiTurnCancellation?: ApiTurnCancellation; + }; + /** * Compose conversation work once for production and integration tests. * Environment-specific queue, state, Slack, and agent adapters stop here. */ export function createConversationWork( options: ConversationWorkOptions, -): VercelConversationWorkCallbackOptions & { +): ConversationWorkCallbackOptions & { runtime: ReturnType; } { + const apiTurnCancellation = createApiTurnCancellation(); const services: JuniorRuntimeServiceOverrides = { ...options.services, replyExecutor: { @@ -112,11 +120,13 @@ export function createConversationWork( fallbackWorker: slackWorker, }); return { + apiTurnCancellation, conversationStore: options.conversationStore, queue: options.queue, run: routeApiTurnWork({ apiTurnWorker: createApiTurnWorker({ agentRunner: options.agentRunner, + cancellation: apiTurnCancellation, }), fallbackWorker: routeAgentInvocationWork({ invocationWorker: createAgentInvocationWorker({ diff --git a/packages/junior/src/chat/app/production.ts b/packages/junior/src/chat/app/production.ts index 567914be5d..c48d1df83d 100644 --- a/packages/junior/src/chat/app/production.ts +++ b/packages/junior/src/chat/app/production.ts @@ -12,11 +12,13 @@ import { createChatSdkLogger } from "@/chat/logging"; import { createJuniorSlackAdapter } from "@/chat/slack/adapter"; import type { SlackWebhookServices } from "@/chat/ingress/slack-webhook"; import { getVercelConversationWorkQueue } from "@/chat/task-execution/vercel-queue"; -import type { VercelConversationWorkCallbackOptions } from "@/chat/task-execution/vercel-callback"; import type { JuniorRuntimeServiceOverrides } from "@/chat/app/services"; import { getConversationStore } from "@/chat/db"; import type { ConversationStore } from "@/chat/conversations/store"; -import { createConversationWork } from "@/chat/app/conversation-work"; +import { + createConversationWork, + type ConversationWorkCallbackOptions, +} from "@/chat/app/conversation-work"; let productionSlackAdapter: SlackAdapter | undefined; let productionSlackRuntime: ReturnType | undefined; @@ -95,7 +97,7 @@ export function getProductionSlackWebhookServices(): SlackWebhookServices { export function createProductionConversationWorkOptions(options: { agentRunner: AgentRunner; services?: JuniorRuntimeServiceOverrides; -}): VercelConversationWorkCallbackOptions { +}): ConversationWorkCallbackOptions { const conversationStore = getProductionConversationStore(); return createConversationWork({ agentRunner: options.agentRunner, diff --git a/packages/junior/src/chat/conversations/history.ts b/packages/junior/src/chat/conversations/history.ts index 49a00806cb..3ebc4cc5c9 100644 --- a/packages/junior/src/chat/conversations/history.ts +++ b/packages/junior/src/chat/conversations/history.ts @@ -327,7 +327,7 @@ const turnCompletedEventDataSchema = z .object({ type: z.literal("turn_completed"), turnId: z.string().min(1), - outcome: z.enum(["success", "no_reply"]), + outcome: z.enum(["success", "no_reply", "cancelled"]), }) .strict(); diff --git a/packages/junior/src/chat/conversations/turn-lifecycle.ts b/packages/junior/src/chat/conversations/turn-lifecycle.ts index 771913b37a..9d53b0e7e4 100644 --- a/packages/junior/src/chat/conversations/turn-lifecycle.ts +++ b/packages/junior/src/chat/conversations/turn-lifecycle.ts @@ -16,7 +16,7 @@ export interface StartConversationTurnInput { export interface CompleteConversationTurnInput { conversationId: string; createdAtMs: number; - outcome: "success" | "no_reply"; + outcome: "success" | "no_reply" | "cancelled"; turnId: string; } diff --git a/packages/junior/src/chat/experimental.ts b/packages/junior/src/chat/experimental.ts index 94c405f6ac..74e78797f9 100644 --- a/packages/junior/src/chat/experimental.ts +++ b/packages/junior/src/chat/experimental.ts @@ -3,7 +3,7 @@ * Add new keys here as features graduate from private experiments; remove them * once they become stable defaults. */ -export const EXPERIMENTAL_FEATURES = ["subagents"] as const; +export const EXPERIMENTAL_FEATURES = ["acp", "subagents"] as const; /** One known experimental feature name. */ export type ExperimentalFeature = (typeof EXPERIMENTAL_FEATURES)[number]; diff --git a/packages/junior/tests/fixtures/api-turn.ts b/packages/junior/tests/fixtures/api-turn.ts index 537d707e2e..ba7717ae7f 100644 --- a/packages/junior/tests/fixtures/api-turn.ts +++ b/packages/junior/tests/fixtures/api-turn.ts @@ -14,7 +14,10 @@ import { createAndEnqueueApiConversation, webActorFromEmail, } from "@/chat/api-turns/work"; -import { createConversationWork } from "@/chat/app/conversation-work"; +import { + createConversationWork, + type ConversationWorkCallbackOptions, +} from "@/chat/app/conversation-work"; import type { ConversationStore } from "@/chat/conversations/store"; import { closeDb, @@ -22,6 +25,7 @@ import { getConversationStore, } from "@/chat/db"; import type { AgentRunner } from "@/chat/runtime/agent-runner"; +import type { AgentRun } from "@/chat/agent/types"; import { disconnectStateAdapter, getStateAdapter } from "@/chat/state/adapter"; import { processConversationQueueMessage } from "@/chat/task-execution/vercel-callback"; import type { ConversationWorkerContext } from "@/chat/task-execution/worker"; @@ -91,7 +95,9 @@ export function emptyApiTurnAttempt(args: { export type ConversationWorkWebHarness = { actor: typeof apiTurnTestActor; + agentRuns: AgentRun[]; agentRunner: AgentRunner; + conversationWork: ConversationWorkCallbackOptions; conversationStore: ConversationStore; queue: ConversationWorkQueueTestAdapter; state: StateAdapter; @@ -132,8 +138,12 @@ export async function createConversationWorkWebHarness( const state = getStateAdapter(); await state.connect(); let modelStream = options.modelStream ?? streamReplies("Web turn complete."); + const agentRuns: AgentRun[] = []; const agentRunner: AgentRunner = options.agentRunner ?? { - run: async (request) => await executeAgentRun(request, modelStream), + run: async (request) => { + agentRuns.push(request); + return await executeAgentRun(request, modelStream); + }, }; const work = createConversationWork({ agentRunner, @@ -155,7 +165,15 @@ export async function createConversationWorkWebHarness( return { actor, + agentRuns, agentRunner, + conversationWork: { + apiTurnCancellation: work.apiTurnCancellation, + conversationStore, + queue, + run: work.run, + state, + }, conversationStore, queue, state, diff --git a/packages/junior/tests/integration/acp-http.test.ts b/packages/junior/tests/integration/acp-http.test.ts new file mode 100644 index 0000000000..adcea3e03e --- /dev/null +++ b/packages/junior/tests/integration/acp-http.test.ts @@ -0,0 +1,837 @@ +import * as acp from "@agentclientprotocol/sdk"; +import { createHttpStream } from "@agentclientprotocol/sdk/experimental/http-client"; +import type { Hono } from "hono"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { createApp } from "@/app"; +import { getConversationEventStore } from "@/chat/db"; +import { createPersonalToken } from "@/personal-tokens/store"; +import { + closeApiTurnWorkFixture, + createConversationWorkWebHarness, +} from "../fixtures/api-turn"; +import { deferred, streamReplies } from "../fixtures/conversation-work"; +import { createModelStream } from "../fixtures/model-stream"; + +const ACP_URL = "http://junior.test/api/acp"; + +function appFetch(app: Hono): typeof globalThis.fetch { + return async (input, init) => + await app.fetch(new Request(input, init as RequestInit)); +} + +function initializeRequest(token?: string): Request { + return new Request(ACP_URL, { + method: "POST", + headers: { + "Content-Type": "application/json", + ...(token ? { Authorization: `Bearer ${token}` } : {}), + }, + body: JSON.stringify({ + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { + protocolVersion: acp.PROTOCOL_VERSION, + clientCapabilities: {}, + }, + }), + }); +} + +async function withAcpClient(args: { + app: Hono; + onUpdate?: (update: acp.SessionUpdate) => void; + run: (context: acp.ClientContext) => Promise; + token: string; +}): Promise { + const stream = createHttpStream(ACP_URL, { + fetch: appFetch(args.app), + headers: { Authorization: `Bearer ${args.token}` }, + }); + try { + return await acp + .client({ name: "junior-acp-test" }) + .onNotification(acp.methods.client.session.update, (context) => { + args.onUpdate?.(context.params.update); + }) + .connectWith(stream, args.run); + } finally { + await stream.writable.close().catch(() => undefined); + } +} + +/** Drive explicit JSON-RPC ids through the official HTTP transport. */ +async function withRawAcpConnection(args: { + app: Hono; + run: ( + request: ( + id: acp.JsonRpcId, + method: string, + params: unknown, + ) => Promise, + ) => Promise; + token: string; +}): Promise { + const stream = createHttpStream(ACP_URL, { + fetch: appFetch(args.app), + headers: { Authorization: `Bearer ${args.token}` }, + }); + const reader = stream.readable.getReader(); + const writer = stream.writable.getWriter(); + const request = async ( + id: acp.JsonRpcId, + method: string, + params: unknown, + ): Promise => { + await writer.write({ jsonrpc: "2.0", id, method, params }); + while (true) { + const next = await reader.read(); + if (next.done) { + throw new Error("ACP stream closed before the response arrived"); + } + if (!("id" in next.value) || next.value.id !== id) continue; + if ("method" in next.value) continue; + if ("error" in next.value) { + throw new Error( + `ACP request failed: ${next.value.error.code} ${next.value.error.message}`, + ); + } + return next.value.result; + } + }; + try { + return await args.run(request); + } finally { + await writer.close().catch(() => undefined); + writer.releaseLock(); + await reader.cancel().catch(() => undefined); + reader.releaseLock(); + } +} + +describe("remote ACP HTTP", () => { + afterEach(async () => { + await closeApiTurnWorkFixture(); + }); + + it("does not mount the endpoint without the experimental flag", async () => { + const harness = await createConversationWorkWebHarness(); + const app = await createApp({ + conversationWork: harness.conversationWork, + experimental: { subagents: true }, + }); + + const response = await app.fetch(initializeRequest()); + + expect(response.status).toBe(404); + }); + + it("requires a valid personal bearer token before ACP dispatch", async () => { + const harness = await createConversationWorkWebHarness(); + const app = await createApp({ + conversationWork: harness.conversationWork, + experimental: { acp: true, subagents: true }, + }); + + const missing = await app.fetch(initializeRequest()); + const invalid = await app.fetch(initializeRequest("jr_pat_invalid")); + + expect(missing.status).toBe(401); + expect(invalid.status).toBe(401); + }); + + it("rejects unsafe envelopes and failed initialization", async () => { + const harness = await createConversationWorkWebHarness(); + const app = await createApp({ + conversationWork: harness.conversationWork, + experimental: { acp: true, subagents: true }, + }); + const token = await createPersonalToken({ + email: harness.actor.email, + name: "ACP envelope validation", + }); + const malformed = await app.request(ACP_URL, { + method: "POST", + headers: { + Authorization: `Bearer ${token.token}`, + "Content-Type": "application/json", + }, + body: JSON.stringify({ jsonrpc: "2.0", privatePrompt: "sentinel" }), + }); + const wrongVersion = await app.request(ACP_URL, { + method: "POST", + headers: { + Authorization: `Bearer ${token.token}`, + "Content-Type": "application/json", + }, + body: JSON.stringify({ + jsonrpc: "1.0", + id: 1, + method: "initialize", + params: {}, + }), + }); + const nonFiniteId = await app.request(ACP_URL, { + method: "POST", + headers: { + Authorization: `Bearer ${token.token}`, + "Content-Type": "application/json", + }, + body: '{"jsonrpc":"2.0","id":1e400,"method":"session/new"}', + }); + const failedInitialize = await app.request(ACP_URL, { + method: "POST", + headers: { + Authorization: `Bearer ${token.token}`, + "Content-Type": "application/json", + }, + body: JSON.stringify({ + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { clientCapabilities: {} }, + }), + }); + + expect(malformed.status).toBe(400); + expect(wrongVersion.status).toBe(400); + expect(nonFiniteId.status).toBe(400); + expect(failedInitialize.status).toBe(200); + expect(failedInitialize.headers.get("Acp-Connection-Id")).toBeNull(); + await expect(failedInitialize.json()).resolves.toMatchObject({ + error: { code: -32602 }, + }); + }); + + it("runs, reloads, and protects a private Conversation through the official client", async () => { + const harness = await createConversationWorkWebHarness({ + modelStream: streamReplies("First ACP reply."), + }); + const app = await createApp({ + conversationWork: harness.conversationWork, + experimental: { acp: true, subagents: true }, + }); + const ownerToken = await createPersonalToken({ + email: harness.actor.email, + name: "ACP owner", + }); + const otherToken = await createPersonalToken({ + email: "bob@example.com", + name: "ACP other actor", + }); + const firstUpdates: acp.SessionUpdate[] = []; + let resolveFirstSession!: (sessionId: string) => void; + const firstSession = new Promise((resolve) => { + resolveFirstSession = resolve; + }); + + const firstRun = withAcpClient({ + app, + token: ownerToken.token, + onUpdate: (update) => firstUpdates.push(update), + run: async (context) => { + const initialized = await context.request( + acp.methods.agent.initialize, + { + protocolVersion: acp.PROTOCOL_VERSION, + clientCapabilities: {}, + }, + ); + expect(initialized).toMatchObject({ + protocolVersion: acp.PROTOCOL_VERSION, + authMethods: [], + }); + expect(initialized.agentCapabilities).toEqual({ + loadSession: true, + promptCapabilities: { + audio: false, + embeddedContext: false, + image: false, + }, + }); + const session = await context.request(acp.methods.agent.session.new, { + cwd: "/client/workspace", + mcpServers: [], + }); + resolveFirstSession(session.sessionId); + const result = await context.request(acp.methods.agent.session.prompt, { + sessionId: session.sessionId, + prompt: [ + { type: "text", text: "First" }, + { type: "text", text: "ACP prompt." }, + ], + }); + return { result, sessionId: session.sessionId }; + }, + }); + + const sessionId = await firstSession; + await expect( + harness.conversationStore.get({ conversationId: sessionId }), + ).resolves.toMatchObject({ + conversationId: sessionId, + source: "web", + visibility: "private", + }); + await vi.waitFor(() => { + expect(harness.queue.hasQueuedMessages()).toBe(true); + }); + await harness.drain(); + const first = await firstRun; + + expect(first.result).toEqual({ stopReason: "end_turn" }); + expect(first.sessionId).toBe(sessionId); + expect(firstUpdates).toEqual([ + expect.objectContaining({ + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: "First ACP reply." }, + }), + ]); + expect(harness.agentRuns).toHaveLength(1); + expect(harness.agentRuns[0]).toMatchObject({ + publishExternally: false, + source: { platform: "web", visibility: "private" }, + }); + await expect(harness.historyTexts(sessionId)).resolves.toEqual([ + "First\nACP prompt.", + "First ACP reply.", + ]); + + const rawInitialize = await app.fetch(initializeRequest(ownerToken.token)); + const connectionId = rawInitialize.headers.get("Acp-Connection-Id"); + expect(connectionId).toBeTruthy(); + const crossActorConnection = await app.request(ACP_URL, { + method: "GET", + headers: { + Accept: "text/event-stream", + Authorization: `Bearer ${otherToken.token}`, + "Acp-Connection-Id": connectionId!, + }, + }); + expect(crossActorConnection.status).toBe(404); + await app.request(ACP_URL, { + method: "DELETE", + headers: { + Authorization: `Bearer ${ownerToken.token}`, + "Acp-Connection-Id": connectionId!, + }, + }); + + const crossActorSessionErrors = await withAcpClient({ + app, + token: otherToken.token, + run: async (context) => { + await context.request(acp.methods.agent.initialize, { + protocolVersion: acp.PROTOCOL_VERSION, + clientCapabilities: {}, + }); + let loadError: unknown; + let promptError: unknown; + try { + await context.request(acp.methods.agent.session.load, { + sessionId, + cwd: "/client/workspace", + mcpServers: [], + }); + } catch (error) { + loadError = error; + } + try { + await context.request(acp.methods.agent.session.prompt, { + sessionId, + prompt: [{ type: "text", text: "Cross-Actor prompt." }], + }); + } catch (error) { + promptError = error; + } + return { loadError, promptError }; + }, + }); + expect(crossActorSessionErrors).toEqual({ + loadError: expect.objectContaining({ code: -32002 }), + promptError: expect.objectContaining({ code: -32002 }), + }); + expect(harness.queue.hasQueuedMessages()).toBe(false); + await expect(harness.historyTexts(sessionId)).resolves.toEqual([ + "First\nACP prompt.", + "First ACP reply.", + ]); + + harness.setModelStream(streamReplies("Second ACP reply.")); + const secondUpdates: acp.SessionUpdate[] = []; + let resolveSecondPrompt!: () => void; + const secondPromptStarted = new Promise((resolve) => { + resolveSecondPrompt = resolve; + }); + const secondRun = withAcpClient({ + app, + token: ownerToken.token, + onUpdate: (update) => secondUpdates.push(update), + run: async (context) => { + await context.request(acp.methods.agent.initialize, { + protocolVersion: acp.PROTOCOL_VERSION, + clientCapabilities: {}, + }); + await context.request(acp.methods.agent.session.load, { + sessionId, + cwd: "/different/client/workspace", + mcpServers: [], + }); + resolveSecondPrompt(); + return await context.request(acp.methods.agent.session.prompt, { + sessionId, + prompt: [{ type: "text", text: "Follow up." }], + }); + }, + }); + + await secondPromptStarted; + await vi.waitFor(() => { + expect(harness.queue.hasQueuedMessages()).toBe(true); + }); + await harness.drain(); + await expect(secondRun).resolves.toEqual({ stopReason: "end_turn" }); + expect(secondUpdates).toEqual([ + expect.objectContaining({ + sessionUpdate: "user_message_chunk", + content: { type: "text", text: "First\nACP prompt." }, + }), + expect.objectContaining({ + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: "First ACP reply." }, + }), + expect.objectContaining({ + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: "Second ACP reply." }, + }), + ]); + await expect(harness.historyTexts(sessionId)).resolves.toEqual([ + "First\nACP prompt.", + "First ACP reply.", + "Follow up.", + "Second ACP reply.", + ]); + }, 20_000); + + it("deduplicates repeated request ids without colliding id types", async () => { + const harness = await createConversationWorkWebHarness({ + modelStream: streamReplies("Typed id reply."), + }); + const app = await createApp({ + conversationWork: harness.conversationWork, + experimental: { acp: true, subagents: true }, + }); + const token = await createPersonalToken({ + email: harness.actor.email, + name: "ACP idempotency", + }); + + await withRawAcpConnection({ + app, + token: token.token, + run: async (request) => { + await request(0, acp.methods.agent.initialize, { + protocolVersion: acp.PROTOCOL_VERSION, + clientCapabilities: {}, + }); + const created = await request(1, acp.methods.agent.session.new, { + cwd: "/client/workspace", + mcpServers: [], + }); + if ( + typeof created !== "object" || + created === null || + !("sessionId" in created) || + typeof created.sessionId !== "string" + ) { + throw new Error("ACP session/new returned no session id"); + } + const sessionId = created.sessionId; + const first = request(2, acp.methods.agent.session.prompt, { + sessionId, + prompt: [{ type: "text", text: "Numeric request id." }], + }); + await vi.waitFor(() => { + expect(harness.queue.hasQueuedMessages()).toBe(true); + }); + await harness.drain(); + await expect(first).resolves.toEqual({ stopReason: "end_turn" }); + + await expect( + request(2, acp.methods.agent.session.prompt, { + sessionId, + prompt: [{ type: "text", text: "Numeric request id." }], + }), + ).resolves.toEqual({ stopReason: "end_turn" }); + expect(harness.queue.hasQueuedMessages()).toBe(false); + expect(harness.agentRuns).toHaveLength(1); + + harness.setModelStream(streamReplies("Typed id reply.")); + const typed = request("2", acp.methods.agent.session.prompt, { + sessionId, + prompt: [{ type: "text", text: "String request id." }], + }); + const typedResult = typed.then((result) => { + expect(result).toEqual({ stopReason: "end_turn" }); + }); + await vi.waitFor(() => { + expect(harness.queue.hasQueuedMessages()).toBe(true); + }); + await harness.drain(); + await typedResult; + expect(harness.agentRuns).toHaveLength(2); + await expect(harness.historyTexts(sessionId)).resolves.toEqual([ + "Numeric request id.", + "Typed id reply.", + "String request id.", + "Typed id reply.", + ]); + }, + }); + }, 20_000); + + it("finishes durable work after the ACP connection closes", async () => { + const harness = await createConversationWorkWebHarness({ + modelStream: streamReplies("Completed after disconnect."), + }); + const app = await createApp({ + conversationWork: harness.conversationWork, + experimental: { acp: true, subagents: true }, + }); + const token = await createPersonalToken({ + email: harness.actor.email, + name: "ACP disconnect", + }); + const stream = createHttpStream(ACP_URL, { + fetch: appFetch(app), + headers: { Authorization: `Bearer ${token.token}` }, + }); + let resolveSession!: (sessionId: string) => void; + const session = new Promise((resolve) => { + resolveSession = resolve; + }); + const connected = acp + .client({ name: "junior-acp-disconnect-test" }) + .connectWith(stream, async (context) => { + await context.request(acp.methods.agent.initialize, { + protocolVersion: acp.PROTOCOL_VERSION, + clientCapabilities: {}, + }); + const created = await context.request(acp.methods.agent.session.new, { + cwd: "/client/workspace", + mcpServers: [], + }); + resolveSession(created.sessionId); + return await context.request(acp.methods.agent.session.prompt, { + sessionId: created.sessionId, + prompt: [{ type: "text", text: "Keep running." }], + }); + }); + const connectionClosed = connected.catch(() => undefined); + + const sessionId = await session; + await vi.waitFor(() => { + expect(harness.queue.hasQueuedMessages()).toBe(true); + }); + await stream.writable.close(); + await connectionClosed; + await harness.drain(); + + const replayed: acp.SessionUpdate[] = []; + await withAcpClient({ + app, + token: token.token, + onUpdate: (update) => replayed.push(update), + run: async (context) => { + await context.request(acp.methods.agent.initialize, { + protocolVersion: acp.PROTOCOL_VERSION, + clientCapabilities: {}, + }); + await context.request(acp.methods.agent.session.load, { + sessionId, + cwd: "/client/workspace", + mcpServers: [], + }); + }, + }); + + await expect(harness.historyTexts(sessionId)).resolves.toEqual([ + "Keep running.", + "Completed after disconnect.", + ]); + expect(replayed).toEqual([ + expect.objectContaining({ + sessionUpdate: "user_message_chunk", + content: { type: "text", text: "Keep running." }, + }), + expect.objectContaining({ + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: "Completed after disconnect." }, + }), + ]); + }, 20_000); + + it("cancels the active Turn and accepts a later prompt", async () => { + const modelStarted = deferred(); + const releaseModel = deferred(); + const harness = await createConversationWorkWebHarness({ + modelStream: createModelStream([ + { + type: "text", + text: "This reply must not be stored.", + onRequest: () => modelStarted.resolve(), + waitFor: releaseModel.promise, + }, + ]), + }); + const app = await createApp({ + conversationWork: harness.conversationWork, + experimental: { acp: true, subagents: true }, + }); + const token = await createPersonalToken({ + email: harness.actor.email, + name: "ACP cancellation", + }); + const sessionCreated = deferred(); + let cancelActiveTurn: (() => Promise) | undefined; + + const cancelledRun = withAcpClient({ + app, + token: token.token, + run: async (context) => { + await context.request(acp.methods.agent.initialize, { + protocolVersion: acp.PROTOCOL_VERSION, + clientCapabilities: {}, + }); + const session = await context.request(acp.methods.agent.session.new, { + cwd: "/client/workspace", + mcpServers: [], + }); + sessionCreated.resolve(session.sessionId); + cancelActiveTurn = async () => { + await context.notify(acp.methods.agent.session.cancel, { + sessionId: session.sessionId, + }); + }; + return await context.request(acp.methods.agent.session.prompt, { + sessionId: session.sessionId, + prompt: [{ type: "text", text: "Cancel this Turn." }], + }); + }, + }); + + const sessionId = await sessionCreated.promise; + await vi.waitFor(() => { + expect(harness.queue.hasQueuedMessages()).toBe(true); + }); + const draining = harness.drain(); + await modelStarted.promise; + if (!cancelActiveTurn) { + throw new Error("ACP cancellation handler was not ready"); + } + await cancelActiveTurn(); + await vi.waitFor(() => { + expect(harness.agentRuns[0]?.signal?.aborted).toBe(true); + }); + releaseModel.resolve(); + await draining; + + await expect(cancelledRun).resolves.toEqual({ stopReason: "cancelled" }); + expect(harness.agentRuns).toHaveLength(1); + await expect(harness.historyTexts(sessionId)).resolves.toEqual([ + "Cancel this Turn.", + ]); + const terminalEvents = await getConversationEventStore().query(sessionId, { + limit: 50, + types: ["turn_completed", "turn_failed"], + }); + expect(terminalEvents.events).toEqual([ + expect.objectContaining({ + data: expect.objectContaining({ + outcome: "cancelled", + type: "turn_completed", + }), + }), + ]); + + harness.setModelStream(streamReplies("Reply after cancellation.")); + const followUpStarted = deferred(); + const followUp = withAcpClient({ + app, + token: token.token, + run: async (context) => { + await context.request(acp.methods.agent.initialize, { + protocolVersion: acp.PROTOCOL_VERSION, + clientCapabilities: {}, + }); + await context.request(acp.methods.agent.session.load, { + sessionId, + cwd: "/client/workspace", + mcpServers: [], + }); + followUpStarted.resolve(); + return await context.request(acp.methods.agent.session.prompt, { + sessionId, + prompt: [{ type: "text", text: "Continue after cancellation." }], + }); + }, + }); + + await followUpStarted.promise; + await vi.waitFor(() => { + expect(harness.queue.hasQueuedMessages()).toBe(true); + }); + await harness.drain(); + await expect(followUp).resolves.toEqual({ stopReason: "end_turn" }); + expect(harness.agentRuns).toHaveLength(2); + await expect(harness.historyTexts(sessionId)).resolves.toEqual([ + "Cancel this Turn.", + "Continue after cancellation.", + "Reply after cancellation.", + ]); + }, 20_000); + + it("maps one durable failed Turn to a protocol error", async () => { + const harness = await createConversationWorkWebHarness({ + modelStream: createModelStream([ + { type: "error", errorMessage: "model unavailable" }, + ]), + }); + const app = await createApp({ + conversationWork: harness.conversationWork, + experimental: { acp: true, subagents: true }, + }); + const token = await createPersonalToken({ + email: harness.actor.email, + name: "ACP failed Turn", + }); + let sessionId: string | undefined; + const failed = withAcpClient({ + app, + token: token.token, + run: async (context) => { + await context.request(acp.methods.agent.initialize, { + protocolVersion: acp.PROTOCOL_VERSION, + clientCapabilities: {}, + }); + const session = await context.request(acp.methods.agent.session.new, { + cwd: "/client/workspace", + mcpServers: [], + }); + sessionId = session.sessionId; + return await context.request(acp.methods.agent.session.prompt, { + sessionId, + prompt: [{ type: "text", text: "Fail this Turn." }], + }); + }, + }); + const failedResult = failed.then( + () => { + throw new Error("Expected the ACP prompt to fail"); + }, + (error: unknown) => { + expect(error).toMatchObject({ + code: -32603, + data: { failureCode: "model_execution_failed" }, + }); + }, + ); + + await vi.waitFor(() => { + expect(harness.queue.hasQueuedMessages()).toBe(true); + }); + await harness.drain(); + await failedResult; + if (!sessionId) throw new Error("ACP session was not created"); + const events = await getConversationEventStore().query(sessionId, { + limit: 50, + types: ["turn_failed"], + }); + expect(events.events).toEqual([ + expect.objectContaining({ + data: expect.objectContaining({ + failureCode: "model_execution_failed", + type: "turn_failed", + }), + }), + ]); + expect(harness.agentRuns).toHaveLength(1); + }, 20_000); + + it("rejects unsupported MCP and prompt content at the protocol boundary", async () => { + const harness = await createConversationWorkWebHarness(); + const app = await createApp({ + conversationWork: harness.conversationWork, + experimental: { acp: true, subagents: true }, + }); + const token = await createPersonalToken({ + email: harness.actor.email, + name: "ACP validation", + }); + + const errors = await withAcpClient({ + app, + token: token.token, + run: async (context) => { + await context.request(acp.methods.agent.initialize, { + protocolVersion: acp.PROTOCOL_VERSION, + clientCapabilities: {}, + }); + let mcpError: unknown; + try { + await context.request(acp.methods.agent.session.new, { + cwd: "/client/workspace", + mcpServers: [ + { command: "example", args: [], env: [], name: "example" }, + ], + }); + } catch (error) { + mcpError = error; + } + const session = await context.request(acp.methods.agent.session.new, { + cwd: "/client/workspace", + mcpServers: [], + }); + const promptErrors: unknown[] = []; + for (const prompt of [ + [ + { + type: "resource_link" as const, + name: "client file", + uri: "file:///client/workspace/file.ts", + }, + ], + [ + { + type: "image" as const, + data: "AA==", + mimeType: "image/png", + }, + ], + [{ type: "text" as const, text: " " }], + [{ type: "text" as const, text: "x".repeat(32_001) }], + ]) { + try { + await context.request(acp.methods.agent.session.prompt, { + sessionId: session.sessionId, + prompt, + }); + } catch (error) { + promptErrors.push(error); + } + } + return { mcpError, promptErrors }; + }, + }); + + expect(errors.mcpError).toMatchObject({ code: -32602 }); + expect(errors.promptErrors).toHaveLength(4); + expect(errors.promptErrors).toEqual([ + expect.objectContaining({ code: -32602 }), + expect.objectContaining({ code: -32602 }), + expect.objectContaining({ code: -32602 }), + expect.objectContaining({ code: -32602 }), + ]); + expect(harness.queue.hasQueuedMessages()).toBe(false); + }); +}); diff --git a/packages/junior/tests/integration/api-turn-work.test.ts b/packages/junior/tests/integration/api-turn-work.test.ts index 36edc3ee44..315577ff58 100644 --- a/packages/junior/tests/integration/api-turn-work.test.ts +++ b/packages/junior/tests/integration/api-turn-work.test.ts @@ -1,5 +1,6 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import { createWebSource } from "@sentry/junior-plugin-api"; +import { createApiTurnCancellation } from "@/chat/api-turns/cancellation"; import { appendAndEnqueueApiConversationMessage, apiTurnIdForMessage, @@ -253,6 +254,66 @@ describe("api turn conversation work", () => { ); }); + it("reports a lost lease and releases cancellation when ack fails", async () => { + const { actor, conversationStore, queue, state } = + await createApiTurnWorkFixture(); + const accepted = await createAndEnqueueApiConversation( + { + actor, + idempotencyKey: "cancel-lost-lease-1", + message: "Cancel before this Turn starts.", + }, + { conversationStore, queue, state }, + ); + const destination = { + platform: "local" as const, + conversationId: accepted.conversationId, + }; + const inbound = buildApiTurnInboundMessage({ + actor, + conversationId: accepted.conversationId, + destination, + message: "Cancel before this Turn starts.", + messageId: accepted.messageId, + }); + const cancellation = createApiTurnCancellation(); + const signal = cancellation.begin(accepted.conversationId); + if (!signal) throw new Error("Expected an active Turn signal"); + cancellation.cancel(accepted.conversationId); + const agentRuns: AgentRun[] = []; + const worker = createApiTurnWorker({ + agentRunner: createModelAgentRunnerForRun((run) => { + agentRuns.push(run); + return createModelStream([ + { type: "text", text: "Cancelled Turn must not reach the agent." }, + ]); + }), + cancellation, + }); + + await expect( + worker({ + attempt: { + ack: async () => { + throw new Error("lease lost"); + }, + conversationId: accepted.conversationId, + destination, + drain: async () => [], + isFinalAttempt: false, + messages: [inbound], + }, + checkIn: async () => true, + conversationId: accepted.conversationId, + destination, + publishExternally: false, + shouldYield: () => false, + }), + ).resolves.toEqual({ status: "lost_lease" }); + expect(agentRuns).toHaveLength(0); + expect(cancellation.begin(accepted.conversationId)).toBeDefined(); + }); + it("routes empty resume wakes to the active API turn", async () => { const { actor, conversationStore, queue, state } = await createApiTurnWorkFixture(); diff --git a/packages/junior/tests/unit/chat/api-turn-cancellation.test.ts b/packages/junior/tests/unit/chat/api-turn-cancellation.test.ts new file mode 100644 index 0000000000..19771ef689 --- /dev/null +++ b/packages/junior/tests/unit/chat/api-turn-cancellation.test.ts @@ -0,0 +1,46 @@ +import { describe, expect, it } from "vitest"; +import { createApiTurnCancellation } from "@/chat/api-turns/cancellation"; + +describe("API Turn cancellation", () => { + it("keeps disconnected running work active until it finishes", () => { + const cancellation = createApiTurnCancellation(); + const signal = cancellation.begin("conversation-1"); + if (!signal) throw new Error("Expected an active Turn signal"); + + cancellation.disconnect("conversation-1", signal); + + expect(cancellation.begin("conversation-1")).toBeUndefined(); + cancellation.finish("conversation-1", signal); + expect(cancellation.begin("conversation-1")).toBeDefined(); + }); + + it.each(["disconnect-first", "park-first"] as const)( + "releases disconnected auth work when %s", + (order) => { + const cancellation = createApiTurnCancellation(); + const signal = cancellation.begin("conversation-1"); + if (!signal) throw new Error("Expected an active Turn signal"); + + if (order === "disconnect-first") { + cancellation.disconnect("conversation-1", signal); + cancellation.park("conversation-1", signal); + } else { + cancellation.park("conversation-1", signal); + cancellation.disconnect("conversation-1", signal); + } + + expect(cancellation.begin("conversation-1")).toBeDefined(); + }, + ); + + it("ignores cancellation after the Turn finishes", () => { + const cancellation = createApiTurnCancellation(); + const signal = cancellation.begin("conversation-1"); + if (!signal) throw new Error("Expected an active Turn signal"); + + cancellation.finish("conversation-1", signal); + + expect(cancellation.cancel("conversation-1")).toBe(false); + expect(signal.aborted).toBe(false); + }); +}); diff --git a/packages/junior/tests/unit/cli/init-cli.test.ts b/packages/junior/tests/unit/cli/init-cli.test.ts index dde3674c15..ad2ccf1158 100644 --- a/packages/junior/tests/unit/cli/init-cli.test.ts +++ b/packages/junior/tests/unit/cli/init-cli.test.ts @@ -28,9 +28,13 @@ function normalizeText(source: string): string { return source.trim().replace(/\n{3,}/g, "\n\n"); } -function removeExampleDashboardServerConfig(source: string): string { +function removeExampleOnlyServerConfig(source: string): string { return normalizeText( source + .replace( + ' experimental: { acp: process.env.NODE_ENV === "development" },\n', + "", + ) .replace( / \{\n exampleDashboardAuthRequired,\n exampleDashboardComponentGallery,\n exampleDashboardMockConversations,\n \},\n/, "", @@ -303,7 +307,7 @@ allowBuilds: "utf8", ); expect(normalizeText(scaffoldServer)).toEqual( - removeExampleDashboardServerConfig(exampleServer), + removeExampleOnlyServerConfig(exampleServer), ); const scaffoldTsConfig = readJsonFile>( diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index b1efd684d4..3ae28f10bf 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -166,6 +166,9 @@ importers: packages/junior: dependencies: + '@agentclientprotocol/sdk': + specifier: 1.3.0 + version: 1.3.0(zod@4.4.3) '@ai-sdk/gateway': specifier: ^3.0.119 version: 3.0.119(zod@4.4.3) @@ -272,6 +275,9 @@ importers: '@emnapi/runtime': specifier: ^1.10.0 version: 1.10.0 + '@hono/node-server': + specifier: 1.19.14 + version: 1.19.14(hono@4.12.27) '@sentry/junior-github': specifier: workspace:* version: file:packages/junior-github(@neondatabase/serverless@1.1.0)(@opentelemetry/api@1.9.1)(@types/pg@8.15.6)(pg@8.21.0) @@ -665,6 +671,11 @@ importers: packages: + '@agentclientprotocol/sdk@1.3.0': + resolution: {integrity: sha512-i3h/efaeuMUFAO1HSfo97QZQnnvMd7wWBYtBsdL6UMZg3a78sk3Ffya5Xu7C7tYsXomXoDXJBAzQF2PcFKAhIQ==} + peerDependencies: + zod: ^3.25.0 || ^4.0.0 + '@ai-sdk/gateway@3.0.119': resolution: {integrity: sha512-VAhfRWC+JexZakkVfmjaJKaTj00x7/UHdE8kMWL3NhuQAlf8oXtg9r4dfvFZrByXxchGRBvYE3biEUyibkg0xg==} engines: {node: '>=18'} @@ -8418,6 +8429,10 @@ packages: snapshots: + '@agentclientprotocol/sdk@1.3.0(zod@4.4.3)': + dependencies: + zod: 4.4.3 + '@ai-sdk/gateway@3.0.119(zod@4.4.3)': dependencies: '@ai-sdk/provider': 3.0.10 @@ -11447,6 +11462,7 @@ snapshots: '@sentry/junior@file:packages/junior': dependencies: + '@agentclientprotocol/sdk': 1.3.0(zod@4.4.3) '@ai-sdk/gateway': 3.0.119(zod@4.4.3) '@chat-adapter/slack': 4.29.0(ai@6.0.190(zod@4.4.3))(zod@4.4.3) '@chat-adapter/state-memory': 4.29.0(ai@6.0.190(zod@4.4.3))(zod@4.4.3) diff --git a/scripts/acp-local.mjs b/scripts/acp-local.mjs new file mode 100644 index 0000000000..bbaea891cc --- /dev/null +++ b/scripts/acp-local.mjs @@ -0,0 +1,74 @@ +import { spawn, spawnSync } from "node:child_process"; +import { fileURLToPath } from "node:url"; +import path from "node:path"; +import { + applyJuniorDevelopmentDefaults, + loadEnvFiles, +} from "./lib/load-env-files.mjs"; + +const workspaceRoot = path.resolve( + path.dirname(fileURLToPath(import.meta.url)), + "..", +); +const exampleRoot = path.join(workspaceRoot, "apps", "example"); +const packageRoot = path.join(workspaceRoot, "packages", "junior"); +const tsconfigPath = path.join(packageRoot, "tsconfig.json"); + +loadEnvFiles([workspaceRoot, exampleRoot]); +applyJuniorDevelopmentDefaults(process.env); + +const compose = spawnSync( + "docker", + ["compose", "up", "-d", "--wait", "postgres", "redis"], + { + cwd: workspaceRoot, + env: process.env, + stdio: "inherit", + }, +); +if (compose.error) { + console.error(`Could not start local services: ${compose.error.message}`); + process.exit(1); +} +if (compose.signal) { + process.kill(process.pid, compose.signal); +} +if (compose.status !== 0) { + process.exit(compose.status ?? 1); +} + +const child = spawn( + "node", + ["--import", "tsx", "scripts/acp-local-server.ts"], + { + cwd: packageRoot, + env: { + ...process.env, + DATABASE_URL: "postgresql://junior:junior@127.0.0.1:54322/junior", + JUNIOR_DATABASE_DRIVER: "postgres", + JUNIOR_STATE_ADAPTER: "redis", + JUNIOR_STATE_KEY_PREFIX: `junior:acp-local:${process.pid}`, + NODE_ENV: "test", + REDIS_URL: "redis://127.0.0.1:6382", + TSX_TSCONFIG_PATH: tsconfigPath, + }, + stdio: "inherit", + }, +); + +child.on("error", (error) => { + console.error(`Could not start local ACP server: ${error.message}`); + process.exit(1); +}); +child.on("exit", (code, signal) => { + if (signal) { + process.removeAllListeners(signal); + process.kill(process.pid, signal); + return; + } + process.exit(code ?? 1); +}); + +for (const signal of ["SIGINT", "SIGTERM"]) { + process.on(signal, () => child.kill(signal)); +}