Initial segregate: proxy source, deploy manifests, ImagePolicy Flux CRs.
build-and-push / build (push) Failing after 2m45s
build-and-push / build (push) Failing after 2m45s
Move workload out of infra with cursor-devbox-style GitRepository + run_number-sha image automation.
This commit is contained in:
Vsevolod Sauta
@@ -0,0 +1,464 @@
|
||||
/**
|
||||
* Cursor-agent session manager (ACP transport).
|
||||
*
|
||||
* Each HTTP chat request creates an ACP session (`agent acp`):
|
||||
* initialize → session/new (MCP stdio bridge) → session/prompt
|
||||
*
|
||||
* Client tools are advertised via in-process ProxyMcpServer. MCP tools/call
|
||||
* parks until the HTTP client returns role:"tool" with the result.
|
||||
*
|
||||
* Built-in cursor-agent tools are denied via session/request_permission.
|
||||
* Images are sent as native ACP image ContentBlocks (not via Read).
|
||||
*/
|
||||
|
||||
import { spawn } from "node:child_process";
|
||||
import { mkdtempSync, rmSync } from "node:fs";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { AcpClient } from "./acp-client.mjs";
|
||||
import { ProxyMcpServer } from "./mcp-server.mjs";
|
||||
|
||||
export const SESSION_TTL_MS = parseInt(process.env.PROXY_SESSION_TTL_MS || "120000", 10);
|
||||
export const MAX_CONCURRENT_SESSIONS = parseInt(process.env.PROXY_MAX_CONCURRENT_SESSIONS || "4", 10);
|
||||
export const TOOL_CALL_BATCH_MS = parseInt(process.env.PROXY_TOOL_CALL_BATCH_MS || "50", 10);
|
||||
const REAL_AGENT = process.env.CURSOR_AGENT_BIN || "/usr/local/bin/agent.real";
|
||||
const MCP_BRIDGE = process.env.MCP_BRIDGE_BIN || "/app/scripts/mcp-stdio-bridge.sh";
|
||||
|
||||
/** @type {Map<string, CursorSession>} */
|
||||
const SESSIONS = new Map();
|
||||
|
||||
/**
|
||||
* Resolve ACP model id from session/new response and requested OpenAI model id.
|
||||
* @param {object} sessionResult session/new result
|
||||
* @param {string} requested
|
||||
*/
|
||||
export function resolveModelId(sessionResult, requested) {
|
||||
const available = sessionResult.models?.availableModels || [];
|
||||
if (available.length > 0) {
|
||||
if (!requested || requested === "auto") {
|
||||
return sessionResult.models.currentModelId || available[0].modelId;
|
||||
}
|
||||
const exact = available.find((m) => m.modelId === requested || m.name === requested);
|
||||
if (exact) return exact.modelId;
|
||||
const byName = available.find((m) => m.name?.toLowerCase() === requested.toLowerCase());
|
||||
if (byName) return byName.modelId;
|
||||
const prefix = available.find(
|
||||
(m) => m.modelId.startsWith(`${requested}[`) || m.modelId.startsWith(requested)
|
||||
);
|
||||
if (prefix) return prefix.modelId;
|
||||
const contains = available.find((m) => m.modelId.includes(requested));
|
||||
if (contains) return contains.modelId;
|
||||
return sessionResult.models.currentModelId || available[0].modelId;
|
||||
}
|
||||
|
||||
const modelOption = (sessionResult.configOptions || []).find(
|
||||
(o) => o.id === "model" || o.category === "model"
|
||||
);
|
||||
return pickModelValueId(modelOption, requested);
|
||||
}
|
||||
|
||||
/**
|
||||
* Pick an ACP model config value for the requested OpenAI model id.
|
||||
* @param {object} modelOption session/new config option with id "model"
|
||||
* @param {string} requested e.g. "auto", "gpt-5.2"
|
||||
*/
|
||||
export function pickModelValueId(modelOption, requested) {
|
||||
const options = modelOption?.options || [];
|
||||
if (!options.length) return null;
|
||||
|
||||
if (!requested || requested === "auto") {
|
||||
const auto = options.find((o) => o.value === "default[]" || o.name === "Auto");
|
||||
return auto?.value || modelOption.currentValue || options[0].value;
|
||||
}
|
||||
|
||||
const exact = options.find((o) => o.value === requested);
|
||||
if (exact) return exact.value;
|
||||
|
||||
const byName = options.find((o) => o.name?.toLowerCase() === requested.toLowerCase());
|
||||
if (byName) return byName.value;
|
||||
|
||||
const prefix = options.find(
|
||||
(o) => o.value.startsWith(`${requested}[`) || o.value.startsWith(requested)
|
||||
);
|
||||
if (prefix) return prefix.value;
|
||||
|
||||
const contains = options.find((o) => o.value.includes(requested));
|
||||
if (contains) return contains.value;
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
function pickPermissionOption(options, allow) {
|
||||
if (!Array.isArray(options) || options.length === 0) return null;
|
||||
const kindMatch = allow
|
||||
? (o) => o.kind === "allow_once" || o.kind === "allow_always"
|
||||
: (o) => o.kind === "reject_once" || o.kind === "reject_always";
|
||||
return options.find(kindMatch) || (allow ? options[0] : options[options.length - 1]);
|
||||
}
|
||||
|
||||
export function extractToolNameFromPermission(params) {
|
||||
const tool = params?.toolCall;
|
||||
if (!tool) return "";
|
||||
|
||||
if (tool.rawInput && typeof tool.rawInput === "object" && typeof tool.rawInput.name === "string") {
|
||||
let name = tool.rawInput.name;
|
||||
if (name.startsWith("mcp_client_")) {
|
||||
name = name.slice("mcp_client_".length);
|
||||
}
|
||||
return name;
|
||||
}
|
||||
|
||||
if (typeof tool.title === "string") {
|
||||
const clientMcp = tool.title.match(/^client-([^:]+):\s*(\S+)/);
|
||||
if (clientMcp) {
|
||||
return clientMcp[2];
|
||||
}
|
||||
const m = tool.title.match(/^(\S+)/);
|
||||
if (m) return m[1].replace(/:$/, "");
|
||||
}
|
||||
|
||||
return "";
|
||||
}
|
||||
|
||||
function mapStopReason(stopReason) {
|
||||
if (stopReason === "end_turn") return "stop";
|
||||
if (stopReason === "cancelled") return "error";
|
||||
return "stop";
|
||||
}
|
||||
|
||||
export class CursorSession {
|
||||
constructor({ sessionId, tools, model, log }) {
|
||||
this.sessionId = sessionId;
|
||||
this.tools = tools || [];
|
||||
this.clientToolNames = new Set(this.tools.map((t) => t.function?.name).filter(Boolean));
|
||||
this.model = model || "auto";
|
||||
this.log = log || console.error;
|
||||
this.detectedModel = model || "auto";
|
||||
this.createdAt = Date.now();
|
||||
this.lastActivityAt = Date.now();
|
||||
this.workDir = null;
|
||||
this.cursorProc = null;
|
||||
this.acp = null;
|
||||
this.mcp = null;
|
||||
this.acpSessionId = null;
|
||||
this.timeoutHandle = null;
|
||||
this._finishToolCallsTimer = null;
|
||||
this.terminated = false;
|
||||
this._handlers = null;
|
||||
this._hasToolCallInTurn = false;
|
||||
this._promptInFlight = false;
|
||||
}
|
||||
|
||||
async start() {
|
||||
this.workDir = mkdtempSync(join(tmpdir(), `cursor-${this.sessionId}-`));
|
||||
|
||||
const mcpTools = this.tools.map((t) => ({
|
||||
name: t.function.name,
|
||||
description: t.function.description,
|
||||
inputSchema: t.function.parameters,
|
||||
}));
|
||||
|
||||
const socketPath = join(this.workDir, "mcp.sock");
|
||||
this.mcp = new ProxyMcpServer({
|
||||
socketPath,
|
||||
tools: mcpTools,
|
||||
onToolCall: ({ toolCallId, name, arguments: args }) => {
|
||||
if (!this._handlers) return;
|
||||
this._hasToolCallInTurn = true;
|
||||
this._handlers.onToolCall({
|
||||
id: toolCallId,
|
||||
name,
|
||||
arguments: JSON.stringify(args || {}),
|
||||
});
|
||||
this._scheduleFinishToolCallsTurn();
|
||||
},
|
||||
log: (...args) => this.log(`[mcp:${this.sessionId}]`, ...args),
|
||||
});
|
||||
await this.mcp.start();
|
||||
|
||||
const args = ["acp"];
|
||||
const env = { ...process.env };
|
||||
if (process.env.CURSOR_API_KEY) env.CURSOR_API_KEY = process.env.CURSOR_API_KEY;
|
||||
|
||||
this.cursorProc = spawn(REAL_AGENT, args, {
|
||||
cwd: this.workDir,
|
||||
env,
|
||||
stdio: ["pipe", "pipe", "pipe"],
|
||||
});
|
||||
|
||||
this.acp = new AcpClient({
|
||||
stdin: this.cursorProc.stdin,
|
||||
stdout: this.cursorProc.stdout,
|
||||
log: (...a) => this.log(`[acp:${this.sessionId}]`, ...a),
|
||||
});
|
||||
|
||||
this.acp.onSessionUpdate = (params) => this._onSessionUpdate(params);
|
||||
this.acp.onIncomingRequest = (method, params) => this._onIncomingRequest(method, params);
|
||||
|
||||
this.cursorProc.stderr.on("data", (d) =>
|
||||
this.log(`[cursor:${this.sessionId} stderr] ${d.toString().trim()}`)
|
||||
);
|
||||
this.cursorProc.on("close", (code) => {
|
||||
this.log(`[cursor:${this.sessionId}] process exited code=${code}`);
|
||||
if (this._handlers) this._finishTurn("error", null);
|
||||
});
|
||||
this.cursorProc.on("error", (e) => {
|
||||
this.log(`[cursor:${this.sessionId}] spawn error: ${e.message}`);
|
||||
if (this._handlers) this._finishTurn("error", null);
|
||||
});
|
||||
|
||||
await this.acp.initialize();
|
||||
await this.acp.authenticateIfNeeded();
|
||||
|
||||
const mcpServers = [
|
||||
{
|
||||
name: "client",
|
||||
command: MCP_BRIDGE,
|
||||
args: [socketPath],
|
||||
env: [],
|
||||
},
|
||||
];
|
||||
|
||||
let newSessionResult;
|
||||
try {
|
||||
newSessionResult = await this.acp.request("session/new", {
|
||||
cwd: this.workDir,
|
||||
mcpServers,
|
||||
});
|
||||
} catch (e) {
|
||||
if (e.code === -32000) {
|
||||
await this.acp.request("authenticate", { methodId: "cursor_login" });
|
||||
newSessionResult = await this.acp.request("session/new", {
|
||||
cwd: this.workDir,
|
||||
mcpServers,
|
||||
});
|
||||
} else {
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
|
||||
this.acpSessionId = newSessionResult.sessionId;
|
||||
const modelId = resolveModelId(newSessionResult, this.model);
|
||||
const currentModelId = newSessionResult.models?.currentModelId;
|
||||
if (modelId && modelId !== currentModelId) {
|
||||
await this.acp.request("session/set_config_option", {
|
||||
sessionId: this.acpSessionId,
|
||||
configId: "model",
|
||||
value: modelId,
|
||||
});
|
||||
}
|
||||
this.detectedModel = modelId || currentModelId || this.model;
|
||||
|
||||
this._resetTimer();
|
||||
return this;
|
||||
}
|
||||
|
||||
async _onIncomingRequest(method, params) {
|
||||
if (method === "session/request_permission") {
|
||||
const toolName = extractToolNameFromPermission(params);
|
||||
const allow = this.clientToolNames.has(toolName);
|
||||
const option = pickPermissionOption(params.options, allow);
|
||||
if (!option) {
|
||||
return { outcome: { outcome: "cancelled" } };
|
||||
}
|
||||
const optionId = option.optionId || option.id;
|
||||
return { outcome: { outcome: "selected", optionId } };
|
||||
}
|
||||
|
||||
if (method === "cursor/ask_question") {
|
||||
const first = params?.questions?.[0]?.options?.[0]?.id;
|
||||
return { answers: first ? { [params.questions[0].id]: first } : {} };
|
||||
}
|
||||
|
||||
if (method === "cursor/create_plan") {
|
||||
return { approved: true };
|
||||
}
|
||||
|
||||
if (
|
||||
method === "cursor/update_todos" ||
|
||||
method === "cursor/task" ||
|
||||
method === "cursor/generate_image"
|
||||
) {
|
||||
return {};
|
||||
}
|
||||
|
||||
this.log(`[acp] unhandled incoming method: ${method}`);
|
||||
return {};
|
||||
}
|
||||
|
||||
_onSessionUpdate(params) {
|
||||
if (!this._handlers || params.sessionId !== this.acpSessionId) return;
|
||||
const update = params.update;
|
||||
if (!update) return;
|
||||
|
||||
if (update.sessionUpdate === "agent_message_chunk") {
|
||||
const content = update.content;
|
||||
if (content?.type === "text" && content.text) {
|
||||
this._handlers.onAssistantDelta(content.text);
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
// agent_thought_chunk and other updates are ignored for OpenAI output.
|
||||
}
|
||||
|
||||
/**
|
||||
* @param {Array<object>|null} blocks ACP ContentBlocks, or null for follow-up only
|
||||
*/
|
||||
runTurn(blocks, handlers) {
|
||||
this._handlers = handlers;
|
||||
this.lastActivityAt = Date.now();
|
||||
this._resetTimer();
|
||||
|
||||
if (blocks === null) {
|
||||
// Follow-up after tool_calls: same session/prompt is still in flight.
|
||||
this._hasToolCallInTurn = false;
|
||||
return;
|
||||
}
|
||||
|
||||
this._hasToolCallInTurn = false;
|
||||
this._promptInFlight = true;
|
||||
this.acp
|
||||
.request("session/prompt", {
|
||||
sessionId: this.acpSessionId,
|
||||
prompt: blocks,
|
||||
})
|
||||
.then((result) => {
|
||||
this._promptInFlight = false;
|
||||
if (!this._handlers) return;
|
||||
if (this.mcp.pending.size > 0 || this._hasToolCallInTurn) {
|
||||
this._scheduleFinishToolCallsTurn();
|
||||
return;
|
||||
}
|
||||
this._finishTurn(mapStopReason(result.stopReason), null);
|
||||
})
|
||||
.catch((e) => {
|
||||
this._promptInFlight = false;
|
||||
this.log(`[acp] session/prompt error: ${e.message}`);
|
||||
if (this._handlers) this._finishTurn("error", null);
|
||||
});
|
||||
}
|
||||
|
||||
resolveToolCall(toolCallId, result) {
|
||||
const ok = this.mcp.resolveToolCall(toolCallId, result);
|
||||
if (ok) {
|
||||
this.lastActivityAt = Date.now();
|
||||
this._resetTimer();
|
||||
}
|
||||
return ok;
|
||||
}
|
||||
|
||||
_scheduleFinishToolCallsTurn() {
|
||||
if (this._finishToolCallsTimer) clearTimeout(this._finishToolCallsTimer);
|
||||
this._finishToolCallsTimer = setTimeout(() => {
|
||||
this._finishToolCallsTimer = null;
|
||||
if (!this._handlers || !this._hasToolCallInTurn) return;
|
||||
if (this.mcp.pending.size === 0) return;
|
||||
this._finishTurn("tool_calls", null);
|
||||
}, TOOL_CALL_BATCH_MS);
|
||||
}
|
||||
|
||||
async destroy() {
|
||||
if (this.terminated) return;
|
||||
this.terminated = true;
|
||||
if (this._finishToolCallsTimer) clearTimeout(this._finishToolCallsTimer);
|
||||
if (this.timeoutHandle) clearTimeout(this.timeoutHandle);
|
||||
try {
|
||||
this.acp?.close();
|
||||
if (this.cursorProc && this.cursorProc.exitCode === null) {
|
||||
this.cursorProc.kill("SIGTERM");
|
||||
await new Promise((r) => {
|
||||
const kill = setTimeout(() => {
|
||||
try {
|
||||
this.cursorProc?.kill("SIGKILL");
|
||||
} catch {
|
||||
/* ignore */
|
||||
}
|
||||
r();
|
||||
}, 5000);
|
||||
this.cursorProc?.once("close", () => {
|
||||
clearTimeout(kill);
|
||||
r();
|
||||
});
|
||||
});
|
||||
}
|
||||
} catch {
|
||||
/* ignore */
|
||||
}
|
||||
try {
|
||||
await this.mcp?.stop();
|
||||
} catch {
|
||||
/* ignore */
|
||||
}
|
||||
try {
|
||||
if (this.workDir) rmSync(this.workDir, { recursive: true, force: true });
|
||||
} catch {
|
||||
/* ignore */
|
||||
}
|
||||
SESSIONS.delete(this.sessionId);
|
||||
}
|
||||
|
||||
_resetTimer() {
|
||||
if (this.timeoutHandle) clearTimeout(this.timeoutHandle);
|
||||
this.timeoutHandle = setTimeout(async () => {
|
||||
this.log(`[cursor:${this.sessionId}] TTL expired (idle ${SESSION_TTL_MS}ms)`);
|
||||
await this.destroy();
|
||||
}, SESSION_TTL_MS);
|
||||
}
|
||||
|
||||
_finishTurn(reason, usage) {
|
||||
if (!this._handlers) return;
|
||||
const handlers = this._handlers;
|
||||
this._handlers = null;
|
||||
void Promise.resolve(handlers.onFinish(reason, usage));
|
||||
}
|
||||
}
|
||||
|
||||
export function getActiveSessionCount() {
|
||||
return SESSIONS.size;
|
||||
}
|
||||
|
||||
export async function createSession({ tools, model, log }) {
|
||||
if (SESSIONS.size >= MAX_CONCURRENT_SESSIONS) {
|
||||
const err = new Error(`too many concurrent sessions (max ${MAX_CONCURRENT_SESSIONS})`);
|
||||
err.code = "too_many_sessions";
|
||||
throw err;
|
||||
}
|
||||
|
||||
const sessionId = `s_${randomUUID().replace(/-/g, "").slice(0, 16)}`;
|
||||
const session = new CursorSession({ sessionId, tools, model, log });
|
||||
SESSIONS.set(sessionId, session);
|
||||
try {
|
||||
await session.start();
|
||||
return session;
|
||||
} catch (e) {
|
||||
await session.destroy();
|
||||
SESSIONS.delete(sessionId);
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
|
||||
export function findSessionForFollowUp(body) {
|
||||
const toolResults = (body.messages || []).filter((m) => m.role === "tool");
|
||||
if (toolResults.length === 0) return null;
|
||||
for (const session of SESSIONS.values()) {
|
||||
if (!session.mcp) continue;
|
||||
for (const tr of toolResults) {
|
||||
if (session.mcp.pending.has(tr.tool_call_id)) {
|
||||
return { session, toolResults };
|
||||
}
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
export function getSession(sessionId) {
|
||||
return SESSIONS.get(sessionId);
|
||||
}
|
||||
|
||||
export async function destroyAllSessions() {
|
||||
const all = [...SESSIONS.values()];
|
||||
SESSIONS.clear();
|
||||
await Promise.all(all.map((s) => s.destroy()));
|
||||
}
|
||||
Reference in New Issue
Block a user