-
Notifications
You must be signed in to change notification settings - Fork 38
feat(acp): add remote ACP prototype #1559
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Merged
Merged
Changes from all commits
Commits
Show all changes
6 commits
Select commit
Hold shift + click to select a range
7fc43b7
feat(acp): add remote ACP prototype
gricha 6d7ffd9
fix(acp): map poll aborts to cancellation
gricha 518d230
docs(acp): remove completed implementation plan
gricha 62b787d
feat(acp): cancel active turns
gricha e40a6a9
fix(acp): close cancellation race windows
gricha 8280e21
fix(acp): preserve cancellation recovery
gricha File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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<void> { | ||
| 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<typeof spawn> | undefined; | ||
| let shutdownPromise: Promise<void> | undefined; | ||
|
|
||
| /** Stop the HTTP client and server, revoke the token, and close test adapters. */ | ||
| function shutdown(): Promise<void> { | ||
| 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; | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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<T>( | ||
| run: (context: acp.ClientContext) => Promise<T>, | ||
| ): Promise<T> { | ||
| 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<void> { | ||
| 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`); | ||
| } | ||
| }); |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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. |
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.