904 lines
43 KiB
TypeScript
904 lines
43 KiB
TypeScript
import { LOCAL_QUERY_TOOL_IDS, PRODUCT_MCP_TOOL_IDS, PRODUCT_MCP_WRITE_TOOL_IDS, QUERY_MCP_SERVER } from "./product-mcp-tools.js";
|
|
import { assertWebSearchReady } from "./web-search-check.js";
|
|
import { allowedQueryTools } from "./query-mcp-policy.js";
|
|
import { setTimeout as delayAbortable } from "node:timers/promises";
|
|
import { isMessageNotFoundError, isSessionNotFoundError, type FormDetail, type OpenCodeClient as OpencodeClient } from "@opencode/client";
|
|
import { createRuntimeClient, startOpencode2 } from "./opencode2-server.js";
|
|
import { formAnswers, normalizeRuntimeMessage, Opencode2EventNormalizer } from "./opencode2-events.js";
|
|
import { existsSync, readFileSync } from "node:fs";
|
|
import { createServer } from "node:net";
|
|
import { resolve } from "node:path";
|
|
import { realpath, rm } from "node:fs/promises";
|
|
import { bindScopedConversationWorkspace, createConversationWorkspace, ensureSessionWorkspaceDirectory, isLegacyConversationWorkspace, isManagedConversationWorkspace, migrateConversationWorkspace, resolveConversationWorkspace, resolveStoredConversationWorkspace, sessionWorkspaceDirectory, type SessionWorkspaceScope } from "./conversationWorkspace.js";
|
|
import { getRuntimeSessionContext } from "./sessionContext.js";
|
|
import { TASK_AGENT } from "./task-agent-policy.js";
|
|
import { serviceOpencodeConfig } from "./opencode2-config.js";
|
|
import { mcpReadonlyTools, mcpRuntimeReportSchema, mcpTimeoutDefaults } from "./mcp-config.js";
|
|
import { parseModelReference } from "../chat/modelConfig.js";
|
|
import type { ResponseInput, InputDelivery } from "../chat/response-inputs.js";
|
|
import { assertModelCheckPassed, modelCheckReportSchema, modelCheckTargets } from "./model-check.js";
|
|
|
|
import { config } from "../config.js";
|
|
import { logger } from "../logger.js";
|
|
|
|
const isDevelopmentDebugLoggingEnabled = process.env.NODE_ENV === "development";
|
|
|
|
const logDevelopmentDebug = (
|
|
message: string,
|
|
metadata: Record<string, unknown>,
|
|
) => {
|
|
if (!isDevelopmentDebugLoggingEnabled) {
|
|
return;
|
|
}
|
|
logger.info(metadata, message);
|
|
};
|
|
|
|
export type RuntimeHealth = {
|
|
healthy: boolean;
|
|
version: string;
|
|
};
|
|
|
|
type RuntimeModelOverride = {
|
|
providerID: string;
|
|
modelID: string;
|
|
};
|
|
|
|
export type PermissionReply = "once" | "always" | "reject";
|
|
export type QuestionAnswers = string[][];
|
|
|
|
type RuntimeMessage = {
|
|
info: {
|
|
id: string;
|
|
role: string;
|
|
};
|
|
};
|
|
|
|
const getRuntimeMessageRole = (message: RuntimeMessage) => message.info.role;
|
|
const getRuntimeMessageId = (message: RuntimeMessage) => message.info.id;
|
|
|
|
const isTcpPortAvailable = (hostname: string, port: number) =>
|
|
new Promise<boolean>((resolve, reject) => {
|
|
const server = createServer();
|
|
server.unref();
|
|
server.once("error", (error: NodeJS.ErrnoException) => {
|
|
if (error.code === "EADDRINUSE") {
|
|
resolve(false);
|
|
return;
|
|
}
|
|
reject(error);
|
|
});
|
|
server.listen({ host: hostname, port, exclusive: true }, () => {
|
|
server.close((error) => (error ? reject(error) : resolve(true)));
|
|
});
|
|
});
|
|
|
|
export const resolveEmbeddedOpencodePort = async (
|
|
hostname: string,
|
|
configuredPort: number,
|
|
) => {
|
|
if (configuredPort === 0) {
|
|
return 0;
|
|
}
|
|
return (await isTcpPortAvailable(hostname, configuredPort))
|
|
? configuredPort
|
|
: 0;
|
|
};
|
|
|
|
export class McpPreparationError extends Error {
|
|
readonly code = "MCP_NOT_READY";
|
|
readonly status = 503;
|
|
constructor(readonly reason: string) { super(`MCP 准备失败：${reason}；未启动模型，请重试或检查服务配置。`); }
|
|
}
|
|
|
|
function recoverableMcpConnectionError(error: unknown): boolean {
|
|
const detail = typeof error === "string" ? error : error instanceof Error ? error.message
|
|
: error && typeof error === "object" && "message" in error ? String(error.message) : "";
|
|
if (/401|403|auth|forbidden|configuration|schema|protocol|not.?found/i.test(detail)) return false;
|
|
return /ECONNREFUSED|ECONNRESET|EPIPE|ETIMEDOUT|EHOSTUNREACH|ENETUNREACH|fetch failed|connection (?:closed|reset)|socket hang up|network (?:error|unavailable)|transport.*closed/i.test(detail);
|
|
}
|
|
|
|
export class OpencodeRuntimeAdapter {
|
|
private clientPromise: Promise<OpencodeClient> | null = null;
|
|
private closeServer: (() => void) | null = null;
|
|
private workspaceBindings = new Map<string, Promise<string>>();
|
|
private promptPreparations?: Map<string, Set<AbortController>>;
|
|
|
|
async ensureClient(): Promise<OpencodeClient> {
|
|
if (!this.clientPromise) {
|
|
this.clientPromise = this.bootstrapClient().catch((error) => {
|
|
this.clientPromise = null;
|
|
throw error;
|
|
});
|
|
}
|
|
return this.clientPromise;
|
|
}
|
|
|
|
async health(): Promise<RuntimeHealth> {
|
|
const client = await this.ensureClient();
|
|
const info = await client.server.info();
|
|
return { healthy: true, version: info.version };
|
|
}
|
|
|
|
mcpReadonlyTools(): string[] {
|
|
return mcpReadonlyTools(buildOpencodeConfig());
|
|
}
|
|
|
|
async mcpStatus(sessionId: string, scope?: SessionWorkspaceScope) {
|
|
const client = await this.ensureClient();
|
|
const session = await client.session.get({ sessionID: sessionId });
|
|
let directory: string;
|
|
try {
|
|
directory = resolve(session.location.directory) === process.cwd() ? process.cwd()
|
|
: await resolveStoredConversationWorkspace(session.location.directory, config.CONVERSATION_WORKSPACE_ROOT);
|
|
} catch (error) {
|
|
if (!scope || (error as NodeJS.ErrnoException).code !== "ENOENT" || !isLegacyConversationWorkspace(session.location.directory, config.CONVERSATION_WORKSPACE_ROOT)) throw error;
|
|
directory = await resolveConversationWorkspace(resolve(sessionWorkspaceDirectory(config.CONVERSATION_WORKSPACE_ROOT, { ...scope, sessionId }), "workspace"), config.CONVERSATION_WORKSPACE_ROOT);
|
|
}
|
|
return this.readMcpReport(client, directory);
|
|
}
|
|
|
|
private async readMcpReport(client: OpencodeClient, directory: string, signal?: AbortSignal) {
|
|
const response = await client.rpc.call({
|
|
rpcID: "tjwater.tools", method: "mcp", input: {}, location: { directory },
|
|
}, { signal: signal ? AbortSignal.any([signal, AbortSignal.timeout(15_000)]) : AbortSignal.timeout(15_000) });
|
|
return mcpRuntimeReportSchema.parse(response.output);
|
|
}
|
|
|
|
private async ensureSessionToolsReady(sessionId: string, signal?: AbortSignal) {
|
|
signal?.throwIfAborted();
|
|
const controller = new AbortController();
|
|
const preparations = this.promptPreparations ??= new Map();
|
|
const pending = preparations.get(sessionId) ?? new Set<AbortController>();
|
|
pending.add(controller);
|
|
preparations.set(sessionId, pending);
|
|
const deadline = AbortSignal.timeout(mcpTimeoutDefaults.startup + mcpTimeoutDefaults.catalog);
|
|
const readySignal = AbortSignal.any([controller.signal, deadline, ...(signal ? [signal] : [])]);
|
|
try {
|
|
const client = await this.ensureClient();
|
|
const directory = await this.ensureSessionWorkspace(sessionId);
|
|
readySignal.throwIfAborted();
|
|
const context = getRuntimeSessionContext(sessionId);
|
|
const grants = new Set(this.mcpReadonlyTools());
|
|
// Verify only queries this host-owned scope can expose. Readiness never
|
|
// grants a tool, changes its executor, or enables an optional MCP server.
|
|
const queries = context ? allowedQueryTools(context).filter((name) =>
|
|
(config.TJWATER_MCP_ENABLED || !PRODUCT_MCP_TOOL_IDS.includes(name)) &&
|
|
(grants.has(name) || PRODUCT_MCP_WRITE_TOOL_IDS.includes(name))) : [];
|
|
if (context?.purpose === "task_execution") {
|
|
if (!context.projectId || !context.taskId) throw new McpPreparationError("后台调查缺少可信任务范围");
|
|
if (!config.TJWATER_MCP_ENABLED) throw new McpPreparationError("后台调查所需业务 MCP 已禁用");
|
|
if (!queries.includes(`${QUERY_MCP_SERVER}_get_task_context`))
|
|
throw new McpPreparationError("后台调查所需上下文工具未获授权");
|
|
}
|
|
const required = context?.purpose === "task_execution"
|
|
? ["ask_task_user", "publish_task_report", "publish_task_progress", "publish_task_presentation", "propose_task_action", ...queries]
|
|
: ["ask_user", "final_answer", ...queries];
|
|
let reconnected = false, transientReadRetries = 0;
|
|
for (;;) {
|
|
readySignal.throwIfAborted();
|
|
let tools: Array<{ id: string; direct: boolean }>;
|
|
let report;
|
|
try {
|
|
const [catalog, current] = await Promise.all([
|
|
client.rpc.call({ rpcID: "tjwater.tools", method: "list", input: {}, location: { directory } }, { signal: readySignal }),
|
|
this.readMcpReport(client, directory, readySignal),
|
|
]);
|
|
if (!Array.isArray(catalog.output) || catalog.output.some((tool) =>
|
|
!tool || typeof tool.id !== "string" || typeof tool.direct !== "boolean"))
|
|
throw new McpPreparationError("工具目录格式无效");
|
|
tools = catalog.output;
|
|
report = current;
|
|
} catch (error) {
|
|
readySignal.throwIfAborted();
|
|
if (error instanceof McpPreparationError) throw error;
|
|
if (!recoverableMcpConnectionError(error) || transientReadRetries++ >= 1)
|
|
throw new McpPreparationError("无法读取工具目录或连接状态");
|
|
await delayAbortable(100, undefined, { signal: readySignal });
|
|
continue;
|
|
}
|
|
readySignal.throwIfAborted();
|
|
if (report.configuration_errors.length) throw new McpPreparationError("MCP 配置无效");
|
|
const connection = report.servers.find((server) => server.name === QUERY_MCP_SERVER);
|
|
if (queries.length && connection?.status === "needs_auth") throw new McpPreparationError("业务工具连接需要鉴权");
|
|
if (queries.length && connection?.status === "disabled") throw new McpPreparationError("业务工具连接已禁用");
|
|
if (queries.length && connection?.status === "failed") {
|
|
const native = await client.mcp.list({ location: { directory } }, { signal: readySignal });
|
|
readySignal.throwIfAborted();
|
|
const state = native.data.find((server) => server.name === QUERY_MCP_SERVER)?.status;
|
|
if (state?.status === "connected" || state?.status === "pending") {
|
|
await delayAbortable(100, undefined, { signal: readySignal });
|
|
continue;
|
|
}
|
|
if (reconnected || state?.status !== "failed" || !recoverableMcpConnectionError(state.error))
|
|
throw new McpPreparationError("业务工具连接失败");
|
|
reconnected = true;
|
|
try { await client.mcp.connect({ server: QUERY_MCP_SERVER, location: { directory } }, { signal: readySignal }); }
|
|
catch { readySignal.throwIfAborted(); throw new McpPreparationError("业务工具重连失败"); }
|
|
continue;
|
|
}
|
|
const missing = required.filter((name) => !tools.some((tool) => tool.id === name && tool.direct));
|
|
if (missing.some((name) => !name.startsWith("mcp_"))) throw new McpPreparationError("必要的宿主工具未注册");
|
|
if (!missing.length && (!queries.length || connection?.status === "connected" && queries.every((name) => connection.tools.includes(name)))) return;
|
|
await delayAbortable(100, undefined, { signal: readySignal });
|
|
}
|
|
} catch (error) {
|
|
signal?.throwIfAborted();
|
|
controller.signal.throwIfAborted();
|
|
if (deadline.aborted) throw new McpPreparationError("等待业务工具目录就绪超时");
|
|
if (error instanceof McpPreparationError) throw error;
|
|
throw new McpPreparationError("业务工具准备不可用");
|
|
} finally {
|
|
pending.delete(controller);
|
|
if (!pending.size) preparations.delete(sessionId);
|
|
}
|
|
}
|
|
|
|
private cancelSessionPreparations(sessionId: string) {
|
|
for (const controller of this.promptPreparations?.get(sessionId) ?? [])
|
|
controller.abort(new DOMException("Session preparation cancelled", "AbortError"));
|
|
}
|
|
|
|
async resolveAvailableModel(preferredModelIds: string[]): Promise<string> {
|
|
const client = await this.ensureClient();
|
|
// V2 2.0.24 model.list alone does not await a cold Location's plugin setup.
|
|
await client.rpc.call({ rpcID: "tjwater.tools", method: "list", input: {}, location: { directory: process.cwd() } });
|
|
const models = await client.model.list({ location: { directory: process.cwd() } });
|
|
const available = new Set(models.data.filter((model) => model.enabled && model.capabilities.tools)
|
|
.map((model) => `${model.providerID}/${model.id}`));
|
|
const selected = preferredModelIds.find((modelId) => available.has(modelId));
|
|
if (!selected) throw new Error("no configured operational task model is available from OpenCode");
|
|
return selected;
|
|
}
|
|
|
|
async checkModels(directory = process.cwd()) {
|
|
const client = await this.ensureClient();
|
|
const response = await client.rpc.call({
|
|
rpcID: "tjwater.tools", method: "check_models", input: modelCheckTargets(config.MODEL_CONFIG), location: { directory },
|
|
}, { signal: AbortSignal.timeout(30_000) });
|
|
return modelCheckReportSchema.parse(response.output);
|
|
}
|
|
|
|
async warmup(): Promise<void> {
|
|
const client = await this.ensureClient();
|
|
const health = await this.health();
|
|
if (!health.healthy || !health.version.startsWith("2.")) {
|
|
throw new Error(`OpenCode 2 is required; received ${health.version}`);
|
|
}
|
|
const session = await this.createSession("tjwater-agent-warmup");
|
|
try {
|
|
const readTools = async () => {
|
|
const response = await client.rpc.call({
|
|
rpcID: "tjwater.tools", method: "list", input: {},
|
|
location: { directory: session.location.directory },
|
|
}, { signal: AbortSignal.timeout(10_000) });
|
|
return response.output as { id: string; direct: boolean; sandboxed: boolean; workspaceRoot?: string }[] | undefined;
|
|
};
|
|
let tools = await readTools();
|
|
const required = [...LOCAL_QUERY_TOOL_IDS, ...(config.TJWATER_MCP_ENABLED ? PRODUCT_MCP_TOOL_IDS : []), "shell", "ask_user", "ask_task_user", "final_answer", "present_answer", "present_view", "memory_manager", "skill_manager", "publish_execution_plan", "publish_task_report", "publish_task_progress", "publish_task_presentation", "propose_task_action", "add_task_human_evidence", "request_task_data_access", "todowrite", "send_task_message", "reply_agent_message"];
|
|
let missing = required.filter((name) => !tools?.some((tool) => tool.id === name && tool.direct));
|
|
const catalogDeadline = Date.now() + 10_000;
|
|
// Native MCP connection/catalog publication can finish after plugin setup.
|
|
// Wait only for mandatory query tools; lost plugin tools still fail fast.
|
|
while (missing.length && missing.every((name) => name.startsWith("mcp_")) && Date.now() < catalogDeadline) {
|
|
await new Promise((resolve) => setTimeout(resolve, 100));
|
|
tools = await readTools();
|
|
missing = required.filter((name) => !tools?.some((tool) => tool.id === name && tool.direct));
|
|
}
|
|
if (missing.length) throw new Error(`OpenCode 2 business tools missing: ${missing.join(", ")}`);
|
|
if (tools?.some((tool) => tool.id.startsWith("browser_") || ["web_search", "mcp_tjwater_web_search", "query_project_data", "query_operational_task_data", "get_operational_task_context"].includes(tool.id))) {
|
|
throw new Error("OpenCode retired tools are still registered; reload trusted configuration and plugins");
|
|
}
|
|
if (!tools?.some((tool) => tool.id === "websearch")) throw new Error("OpenCode websearch tool is missing");
|
|
await assertWebSearchReady(client, session.location.directory);
|
|
if (!tools?.some((tool) => tool.id === "shell" && tool.sandboxed)) throw new Error("OpenCode shell sandbox replacement is missing");
|
|
const actualRoot = tools.find((tool) => tool.id === "shell")?.workspaceRoot;
|
|
if (!actualRoot || await realpath(actualRoot) !== await realpath(config.CONVERSATION_WORKSPACE_ROOT)) throw new Error("OpenCode and Agent conversation workspace roots do not match");
|
|
const mcp = await this.readMcpReport(client, session.location.directory);
|
|
if (mcp.configuration_errors.length) throw new Error("MCP configuration invalid");
|
|
assertModelCheckPassed(await this.checkModels(session.location.directory));
|
|
const models = await client.model.list({ location: { directory: session.location.directory } });
|
|
const model = models.data.find((model) => `${model.providerID}/${model.id}` === config.OPENCODE_MODEL);
|
|
if (!model?.enabled || !model.capabilities.tools) {
|
|
throw new Error(`OpenCode 2 model unavailable or does not support tools: ${config.OPENCODE_MODEL}`);
|
|
}
|
|
logger.info({ version: health.version, model: config.OPENCODE_MODEL, toolCount: tools?.length }, "opencode2 warmup tools verified");
|
|
} finally {
|
|
await client.session.remove({ sessionID: session.id });
|
|
await rm(await resolveConversationWorkspace(session.location.directory, config.CONVERSATION_WORKSPACE_ROOT), { recursive: true });
|
|
}
|
|
}
|
|
|
|
async createSession(title?: string, scope?: SessionWorkspaceScope) {
|
|
const client = await this.ensureClient();
|
|
const { providerID, modelID: id } = parseModelReference(config.OPENCODE_MODEL);
|
|
const directory = await createConversationWorkspace(config.CONVERSATION_WORKSPACE_ROOT);
|
|
let createdSessionId: string | undefined;
|
|
try {
|
|
const session = await client.session.create({ title, agent: "instruction", model: { providerID, id }, location: { directory } });
|
|
createdSessionId = session.id;
|
|
if (scope) await this.ensureSessionWorkspace(session.id, scope);
|
|
return session;
|
|
} catch (error) {
|
|
if (createdSessionId) await client.session.remove({ sessionID: createdSessionId }).catch(() => {});
|
|
await rm(directory, { recursive: true, force: true });
|
|
if (scope && createdSessionId) {
|
|
await ensureSessionWorkspaceDirectory(config.CONVERSATION_WORKSPACE_ROOT, { ...scope, sessionId: createdSessionId })
|
|
.then((parent) => rm(resolve(parent, "workspace"), { recursive: true, force: true })).catch(() => {});
|
|
}
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
async ensureSessionWorkspace(sessionId: string, scope?: SessionWorkspaceScope): Promise<string> {
|
|
this.workspaceBindings ??= new Map();
|
|
const pending = this.workspaceBindings.get(sessionId);
|
|
if (pending) return pending;
|
|
const operation = (async () => {
|
|
const client = await this.ensureClient();
|
|
const session = await client.session.get({ sessionID: sessionId });
|
|
const owner = scope ?? getRuntimeSessionContext(sessionId);
|
|
if (owner) {
|
|
let source = session.location.directory;
|
|
if (resolve(source) === process.cwd()) source = await createConversationWorkspace(config.CONVERSATION_WORKSPACE_ROOT);
|
|
const directory = await bindScopedConversationWorkspace(source, config.CONVERSATION_WORKSPACE_ROOT, { ...owner, sessionId });
|
|
if (directory !== resolve(session.location.directory)) await client.session.move({ sessionID: sessionId, directory });
|
|
return directory;
|
|
}
|
|
if (resolve(session.location.directory) !== process.cwd()) {
|
|
const directory = await migrateConversationWorkspace(session.location.directory, config.CONVERSATION_WORKSPACE_ROOT);
|
|
if (directory !== resolve(session.location.directory)) await client.session.move({ sessionID: sessionId, directory });
|
|
return directory;
|
|
}
|
|
// Old sessions used the service checkout. Moving an idle session binds a
|
|
// fresh directory without copying application data, credentials or logs.
|
|
const directory = await createConversationWorkspace(config.CONVERSATION_WORKSPACE_ROOT);
|
|
try { await client.session.move({ sessionID: sessionId, directory }); }
|
|
catch (error) { await rm(directory, { recursive: true }); throw error; }
|
|
return directory;
|
|
})();
|
|
this.workspaceBindings.set(sessionId, operation);
|
|
try { return await operation; }
|
|
finally { this.workspaceBindings.delete(sessionId); }
|
|
}
|
|
|
|
async removeSession(sessionId: string, scope?: SessionWorkspaceScope) {
|
|
const client = await this.ensureClient();
|
|
let session;
|
|
try { session = await client.session.get({ sessionID: sessionId }); }
|
|
catch (error) { if (isSessionNotFoundError(error)) return; throw error; }
|
|
const root = await realpath(config.CONVERSATION_WORKSPACE_ROOT);
|
|
const directory = scope ? await this.ensureSessionWorkspace(sessionId, scope)
|
|
: await resolveStoredConversationWorkspace(session.location.directory, root);
|
|
if (!scope && !isLegacyConversationWorkspace(directory, root) && !isManagedConversationWorkspace(directory, root, sessionId)) {
|
|
throw new Error("cannot remove another session's workspace");
|
|
}
|
|
this.cancelSessionPreparations(sessionId);
|
|
await client.session.interrupt({ sessionID: sessionId });
|
|
await this.waitForSessionIdle(sessionId);
|
|
if ((await client.session.active())[sessionId]) throw new Error("session is still active; cleanup can be retried");
|
|
try { await client.session.remove({ sessionID: sessionId }); }
|
|
catch (error) { if (!isSessionNotFoundError(error)) throw error; }
|
|
await rm(directory, { recursive: true, force: true });
|
|
}
|
|
|
|
async generateText(prompt: string, model?: RuntimeModelOverride, signal?: AbortSignal) {
|
|
const client = await this.ensureClient();
|
|
const selected = model ?? parseModelReference(config.OPENCODE_MODEL);
|
|
const directory = await createConversationWorkspace(config.CONVERSATION_WORKSPACE_ROOT);
|
|
let sessionId: string | undefined;
|
|
try {
|
|
const session = await client.session.create({
|
|
title: "prompt-suggestions", agent: "prompt-suggestions",
|
|
permissions: [{ action: "*", resource: "*", effect: "deny" }],
|
|
model: { providerID: selected.providerID, id: selected.modelID },
|
|
location: { directory },
|
|
}, { signal });
|
|
sessionId = session.id;
|
|
// Await the Location's configured model catalog before one-shot generation.
|
|
await client.rpc.call({ rpcID: "tjwater.tools", method: "list", input: {}, location: { directory } }, { signal });
|
|
return (await client.session.generate({ sessionID: sessionId, prompt }, { signal })).text;
|
|
} finally {
|
|
try {
|
|
if (sessionId) await client.session.remove({ sessionID: sessionId });
|
|
} finally {
|
|
await rm(directory, { recursive: true });
|
|
}
|
|
}
|
|
}
|
|
|
|
async approveSandboxShell(input: {
|
|
sessionId: string; command: string; description?: string;
|
|
callId: string; messageId: string;
|
|
}, signal: AbortSignal) {
|
|
const client = await this.ensureClient();
|
|
const controller = new AbortController();
|
|
const combined = AbortSignal.any([signal, controller.signal, AbortSignal.timeout(5 * 60_000)]);
|
|
const iterator = client.event.subscribe({ signal: combined })[Symbol.asyncIterator]();
|
|
let requestId: string | undefined;
|
|
let replied = false;
|
|
try {
|
|
// Subscribe first: an immediate client reply must never be lost.
|
|
await iterator.next();
|
|
const permission = await client.permission.create({
|
|
sessionID: input.sessionId, action: "shell", resources: [input.command], save: [],
|
|
metadata: { command: input.command, description: input.description ?? "" },
|
|
source: { type: "tool", messageID: input.messageId, id: input.callId },
|
|
}, { signal: combined });
|
|
requestId = permission.id;
|
|
if (permission.effect === "deny") throw Object.assign(new Error("shell permission denied"), { status: 403 });
|
|
if (permission.effect === "allow") throw Object.assign(new Error("sandbox shell requires interactive approval"), { status: 403 });
|
|
while (true) {
|
|
const next = await iterator.next();
|
|
if (next.done) throw new Error("permission event stream ended before approval");
|
|
const event = next.value;
|
|
if (event.type !== "permission.replied" || event.data.sessionID !== input.sessionId || event.data.requestID !== requestId) continue;
|
|
replied = true;
|
|
if (event.data.reply === "reject") throw Object.assign(new Error("shell permission rejected"), { status: 403 });
|
|
return;
|
|
}
|
|
} finally {
|
|
controller.abort();
|
|
await iterator.return?.();
|
|
if (requestId && !replied) await client.permission.reply({ sessionID: input.sessionId, requestID: requestId, decision: "reject" }).catch(() => {});
|
|
}
|
|
}
|
|
|
|
async agentMessageState(sessionId: string, messageId: string): Promise<"missing" | "enqueued" | "delivered"> {
|
|
const client = await this.ensureClient();
|
|
const pending = await client.session.inbox.list({ sessionID: sessionId });
|
|
if (pending.some((item) => item.id === messageId)) return "enqueued";
|
|
try {
|
|
await client.session.message.get({ sessionID: sessionId, messageID: messageId });
|
|
return "delivered";
|
|
} catch (error) {
|
|
// Only a not-found response is safe to treat as absent. A transport
|
|
// failure must not cause a second injection.
|
|
if (error && typeof error === "object" && "status" in error && error.status === 404) return "missing";
|
|
if (isMessageNotFoundError(error)) return "missing";
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
async enqueueAgentMessage(sessionId: string, messageId: string, text: string, metadata: Record<string, string>) {
|
|
const client = await this.ensureClient();
|
|
// resume:false is essential: a race with idle must never start an
|
|
// unowned background turn in a user's main chat.
|
|
await client.session.synthetic({ sessionID: sessionId, id: messageId, text,
|
|
description: "调查协作消息（非用户直接指令）", metadata, delivery: "steer", resume: false });
|
|
}
|
|
|
|
async enqueueUserInput(sessionId: string, input: ResponseInput, options: { resume?: boolean } = {}) {
|
|
const client = await this.ensureClient();
|
|
if (options.resume !== false) await this.ensureSessionToolsReady(sessionId);
|
|
await client.session.prompt({ sessionID: sessionId, id: input.id, text: input.text,
|
|
delivery: input.delivery, resume: options.resume ?? true, metadata: { source: "tjwater_user_input" } });
|
|
}
|
|
|
|
async listPendingUserInputs(sessionId: string): Promise<ResponseInput[]> {
|
|
const client = await this.ensureClient();
|
|
return (await client.session.inbox.list({ sessionID: sessionId })).flatMap((item) =>
|
|
item.type === "user" && item.payload.metadata?.source === "tjwater_user_input"
|
|
? [{ id: item.id, text: item.payload.text, delivery: item.delivery, status: "pending" as const }] : []);
|
|
}
|
|
|
|
async updatePendingUserInput(sessionId: string, id: string, delivery: InputDelivery) {
|
|
const item = (await this.listPendingUserInputs(sessionId)).find((input) => input.id === id);
|
|
if (!item) throw Object.assign(new Error("Message has already been consumed or is unavailable"), { status: 409 });
|
|
const client = await this.ensureClient();
|
|
try { await client.session.inbox.update({ sessionID: sessionId, inboxID: id, delivery }); }
|
|
catch (error) {
|
|
if (!(await this.listPendingUserInputs(sessionId)).some((input) => input.id === id)) {
|
|
throw Object.assign(new Error("Message has already been consumed or is unavailable"), { status: 409 });
|
|
}
|
|
throw error;
|
|
}
|
|
return { ...item, delivery };
|
|
}
|
|
|
|
async cancelPendingUserInputs(sessionId: string) {
|
|
const client = await this.ensureClient();
|
|
for (const input of await this.listPendingUserInputs(sessionId)) {
|
|
try { await client.session.inbox.cancel({ sessionID: sessionId, inboxID: input.id }); }
|
|
catch (error) { if (!(error && typeof error === "object" && "status" in error && error.status === 404)) throw error; }
|
|
}
|
|
}
|
|
|
|
async cancelPendingUserInput(sessionId: string, id: string): Promise<ResponseInput> {
|
|
const item = (await this.listPendingUserInputs(sessionId)).find((input) => input.id === id);
|
|
if (!item) throw Object.assign(new Error("Message has already been consumed or is unavailable"), { status: 409 });
|
|
const client = await this.ensureClient();
|
|
try { await client.session.inbox.cancel({ sessionID: sessionId, inboxID: id }); }
|
|
catch (error) {
|
|
if (!(await this.listPendingUserInputs(sessionId)).some((input) => input.id === id)) {
|
|
throw Object.assign(new Error("Message has already been consumed or is unavailable"), { status: 409 });
|
|
}
|
|
throw error;
|
|
}
|
|
// Native cancel also succeeds for an already consumed ID. Check the native
|
|
// message record so a delivery racing cancellation never restores a sent draft.
|
|
if (await this.agentMessageState(sessionId, id) !== "missing") {
|
|
throw Object.assign(new Error("Message has already been consumed or is unavailable"), { status: 409 });
|
|
}
|
|
return { ...item, status: "cancelled" };
|
|
}
|
|
|
|
async prompt(sessionId: string, text: string, model?: RuntimeModelOverride, messageId?: string, signal?: AbortSignal) {
|
|
const client = await this.ensureClient();
|
|
const startedAt = Date.now();
|
|
logDevelopmentDebug(
|
|
"dispatching opencode session.prompt",
|
|
{
|
|
sessionId,
|
|
model: model ?? null,
|
|
textChars: text.length,
|
|
},
|
|
);
|
|
const context = getRuntimeSessionContext(sessionId);
|
|
// The server owns purpose; a browser prompt or requested model cannot select
|
|
// the background coordinator profile. Re-evaluate for resumed sessions too.
|
|
const agent = context?.purpose === "task_execution" && context.taskId && context.projectId ? TASK_AGENT : "instruction";
|
|
await client.session.switchAgent({ sessionID: sessionId, agent });
|
|
if (model) await client.session.switchModel({ sessionID: sessionId, model: { providerID: model.providerID, id: model.modelID } });
|
|
await this.ensureSessionToolsReady(sessionId, signal);
|
|
signal?.throwIfAborted();
|
|
await client.session.prompt({ sessionID: sessionId, text, ...(messageId ? { id: messageId, delivery: "steer" as const, metadata: { source: "tjwater_host_answer" } } : {}) }, { signal });
|
|
logDevelopmentDebug(
|
|
"opencode session.prompt returned",
|
|
{
|
|
sessionId,
|
|
elapsedMs: Math.max(0, Date.now() - startedAt),
|
|
},
|
|
);
|
|
}
|
|
|
|
private interactionOperations?: Map<string, Promise<unknown>>;
|
|
private async serializeInteraction<T>(sessionId: string, operation: () => Promise<T>): Promise<T> {
|
|
const operations = this.interactionOperations ??= new Map();
|
|
const pending = (operations.get(sessionId) ?? Promise.resolve()).catch(() => undefined).then(operation);
|
|
operations.set(sessionId, pending);
|
|
try { return await pending; }
|
|
finally { if (operations.get(sessionId) === pending) operations.delete(sessionId); }
|
|
}
|
|
|
|
async stopInteractionInput(sessionId: string, messageId: string): Promise<"delivered" | "cancelled"> {
|
|
return this.serializeInteraction(sessionId, async () => {
|
|
const client = await this.ensureClient();
|
|
const state = await this.agentMessageState(sessionId, messageId);
|
|
if (state === "delivered") return "delivered";
|
|
if (state === "enqueued") await client.session.inbox.cancel({ sessionID: sessionId, inboxID: messageId });
|
|
return await this.agentMessageState(sessionId, messageId) === "delivered" ? "delivered" : "cancelled";
|
|
});
|
|
}
|
|
|
|
async continueInteraction(sessionId: string, messageId: string, text: string, model?: RuntimeModelOverride, signal?: AbortSignal): Promise<"started" | "observing" | "completed"> {
|
|
return this.serializeInteraction(sessionId, async () => {
|
|
signal?.throwIfAborted();
|
|
const client = await this.ensureClient();
|
|
const state = await this.agentMessageState(sessionId, messageId);
|
|
const active = Boolean((await client.session.active())[sessionId]);
|
|
signal?.throwIfAborted();
|
|
if (state === "missing") {
|
|
if (active) throw new Error("Previous native turn is still active");
|
|
await this.prompt(sessionId, text, model, messageId, signal);
|
|
return "started";
|
|
}
|
|
if (active) return "observing";
|
|
if (state === "enqueued") {
|
|
// Resume an admitted steering input without constructing another message.
|
|
await this.ensureSessionToolsReady(sessionId, signal);
|
|
signal?.throwIfAborted();
|
|
await client.session.interrupt({ sessionID: sessionId, resume: true });
|
|
return "observing";
|
|
}
|
|
return "completed";
|
|
});
|
|
}
|
|
|
|
async messages(sessionId: string, limit = 20) {
|
|
const client = await this.ensureClient();
|
|
const messages = await client.message.list({ sessionID: sessionId, limit, order: "desc" });
|
|
return messages.data.reverse().map((message) => normalizeRuntimeMessage(message, sessionId));
|
|
}
|
|
|
|
async readInteractionResult(sessionId: string, messageId: string): Promise<string> {
|
|
const client = await this.ensureClient();
|
|
const messages = (await client.message.list({ sessionID: sessionId, limit: 100, order: "desc" })).data.slice().reverse();
|
|
const start = messages.findIndex((message) => message.id === messageId && message.type === "user");
|
|
if (start < 0) throw new Error("Saved answer result is outside the recovery window");
|
|
const tail = messages.slice(start + 1);
|
|
const nextUser = tail.findIndex((message) => message.type === "user");
|
|
const related = nextUser < 0 ? tail : tail.slice(0, nextUser);
|
|
for (const message of related.slice().reverse()) {
|
|
if (message.type !== "assistant" || message.error) continue;
|
|
for (const content of message.content.slice().reverse()) {
|
|
if (content.type === "tool" && content.name === "final_answer" && content.state.status === "completed") {
|
|
for (const item of content.state.content) {
|
|
if (item.type !== "text") continue;
|
|
try { const parsed: unknown = JSON.parse(item.text); if (parsed && typeof parsed === "object" && "answer" in parsed && typeof parsed.answer === "string") return parsed.answer; } catch { /* A non-JSON tool result is not a final answer. */ }
|
|
}
|
|
}
|
|
}
|
|
if (message.time.completed && message.finish === "stop") {
|
|
const text = message.content.filter((part) => part.type === "text").map((part) => part.text).join("\n");
|
|
if (text) return text;
|
|
}
|
|
}
|
|
throw new Error("Answer was consumed but no completed response could be verified; do not resend it");
|
|
}
|
|
|
|
async revertMessage(sessionId: string, messageId: string) {
|
|
const client = await this.ensureClient();
|
|
return client.session.revert.stage({ sessionID: sessionId, messageID: messageId, files: false });
|
|
}
|
|
|
|
async removeMessage(sessionId: string, messageId: string) {
|
|
const client = await this.ensureClient();
|
|
// V2 commits the staged message boundary atomically; individual deletion was removed.
|
|
await client.session.revert.commit({ sessionID: sessionId });
|
|
}
|
|
|
|
async revertToUserMessage(sessionId: string, options: { userOrdinal: number }) {
|
|
const messages = await this.messages(sessionId, 80);
|
|
const userMessages = messages.filter(
|
|
(message) => getRuntimeMessageRole(message) === "user",
|
|
);
|
|
const targetUserMessage = userMessages[options.userOrdinal - 1];
|
|
|
|
if (!targetUserMessage) {
|
|
if (messages.length === 0 && options.userOrdinal === 1) {
|
|
logger.warn(
|
|
{ sessionId, userOrdinal: options.userOrdinal },
|
|
"skipping opencode revert because runtime session has no messages",
|
|
);
|
|
return;
|
|
}
|
|
throw new Error("target user message not found to revert");
|
|
}
|
|
|
|
const targetMessageId = getRuntimeMessageId(targetUserMessage);
|
|
await this.revertMessage(sessionId, targetMessageId);
|
|
await this.removeMessage(sessionId, targetMessageId);
|
|
}
|
|
|
|
async abortSession(sessionId: string) {
|
|
this.cancelSessionPreparations(sessionId);
|
|
const client = await this.ensureClient();
|
|
return client.session.interrupt({ sessionID: sessionId });
|
|
}
|
|
|
|
async waitForSessionIdle(sessionId: string, timeoutMs = config.OPENCODE_TIMEOUT_MS, strict = false) {
|
|
const client = await this.ensureClient();
|
|
const startedAt = Date.now();
|
|
|
|
while (Date.now() - startedAt < timeoutMs) {
|
|
const active = await client.session.active();
|
|
if (!active[sessionId]) {
|
|
return;
|
|
}
|
|
await delay(100);
|
|
}
|
|
|
|
if (strict) throw new Error("Native session did not reach the waiting boundary");
|
|
logger.warn(
|
|
{ sessionId, timeoutMs },
|
|
"timed out waiting for opencode session to become idle",
|
|
);
|
|
}
|
|
|
|
async subscribeEvents() {
|
|
const client = await this.ensureClient();
|
|
const controller = new AbortController();
|
|
const iterator = client.event.subscribe({ signal: controller.signal })[Symbol.asyncIterator]();
|
|
// Await the connected event before dispatching a prompt so its first events cannot be lost.
|
|
const first = await iterator.next();
|
|
const normalizer = new Opencode2EventNormalizer();
|
|
const stream = (async function* () {
|
|
try {
|
|
let next = first;
|
|
while (!next.done) {
|
|
for (const event of normalizer.normalize(next.value)) yield event;
|
|
next = await iterator.next();
|
|
}
|
|
} catch (error) {
|
|
if (!controller.signal.aborted) throw error;
|
|
} finally {
|
|
controller.abort();
|
|
await iterator.return?.();
|
|
}
|
|
})();
|
|
return {
|
|
[Symbol.asyncIterator]() {
|
|
return {
|
|
next: () => stream.next(),
|
|
return: async () => {
|
|
// Cancel pending next() before returning; async generator return alone cannot interrupt it.
|
|
controller.abort();
|
|
return stream.return(undefined);
|
|
},
|
|
};
|
|
},
|
|
};
|
|
}
|
|
|
|
async replyPermission(options: { requestId: string; sessionId?: string; reply: PermissionReply; message?: string }) {
|
|
if (!options.sessionId) throw new Error("OpenCode 2 permission replies require sessionId");
|
|
const client = await this.ensureClient();
|
|
return client.permission.reply({ sessionID: options.sessionId, requestID: options.requestId, decision: options.reply, message: options.message });
|
|
}
|
|
|
|
async replyQuestion(options: { requestId: string; sessionId: string; answers: QuestionAnswers; form: FormDetail }) {
|
|
if (!options.sessionId) throw new Error("OpenCode 2 form replies require sessionId");
|
|
const client = await this.ensureClient();
|
|
return client.session.form.reply({ sessionID: options.sessionId, formID: options.requestId, answer: formAnswers(options.form, options.answers) });
|
|
}
|
|
|
|
async getQuestion(options: { requestId: string; sessionId: string }): Promise<FormDetail | null> {
|
|
const client = await this.ensureClient();
|
|
try { return await client.session.form.get({ sessionID: options.sessionId, formID: options.requestId }); }
|
|
catch (error) {
|
|
const tag = error && typeof error === "object" && "_tag" in error ? error._tag : undefined;
|
|
if (tag === "FormNotFoundError") return null;
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
async rejectQuestion(options: { requestId: string; sessionId?: string }) {
|
|
if (!options.sessionId) throw new Error("OpenCode 2 form cancellation requires sessionId");
|
|
const client = await this.ensureClient();
|
|
return client.session.form.cancel({ sessionID: options.sessionId, formID: options.requestId });
|
|
}
|
|
|
|
async dispose(): Promise<void> {
|
|
for (const sessionId of this.promptPreparations?.keys() ?? []) this.cancelSessionPreparations(sessionId);
|
|
this.closeServer?.();
|
|
this.closeServer = null;
|
|
this.clientPromise = null;
|
|
}
|
|
|
|
private async bootstrapClient(): Promise<OpencodeClient> {
|
|
if (config.OPENCODE_MODE === "client") {
|
|
logger.info(
|
|
{
|
|
baseUrl: config.OPENCODE_CLIENT_BASE_URL,
|
|
mode: config.OPENCODE_MODE,
|
|
},
|
|
"connecting to opencode2 server in client mode",
|
|
);
|
|
return createRuntimeClient(config.OPENCODE_CLIENT_BASE_URL!, config.OPENCODE_CLIENT_PASSWORD);
|
|
}
|
|
|
|
// embedded 模式下，把服务内工具桥地址注入到 opencode 进程环境里，
|
|
// 这样 .opencode/tools 下的自定义工具可以回调本服务。
|
|
process.env.TJWATER_AGENT_WORKSPACE_ROOT = resolve(config.CONVERSATION_WORKSPACE_ROOT);
|
|
process.env.TJWATER_AGENT_INTERNAL_BASE_URL = `http://127.0.0.1:${config.PORT}`;
|
|
process.env.TJWATER_AGENT_INTERNAL_TOKEN =
|
|
config.AGENT_INTERNAL_TOKEN ??
|
|
process.env.TJWATER_AGENT_INTERNAL_TOKEN ??
|
|
"";
|
|
|
|
const port = await resolveEmbeddedOpencodePort(
|
|
config.OPENCODE_HOSTNAME,
|
|
config.OPENCODE_PORT,
|
|
);
|
|
if (port !== config.OPENCODE_PORT) {
|
|
logger.warn(
|
|
{
|
|
hostname: config.OPENCODE_HOSTNAME,
|
|
configuredPort: config.OPENCODE_PORT,
|
|
},
|
|
"configured opencode port is occupied; falling back to a dynamic port",
|
|
);
|
|
}
|
|
|
|
logger.info(
|
|
{
|
|
hostname: config.OPENCODE_HOSTNAME,
|
|
port,
|
|
model: config.OPENCODE_MODEL,
|
|
mode: config.OPENCODE_MODE,
|
|
},
|
|
"starting opencode2 server in embedded mode",
|
|
);
|
|
|
|
const startedAt = Date.now();
|
|
let runtime;
|
|
try {
|
|
runtime = await startOpencode2({
|
|
command: config.OPENCODE_COMMAND,
|
|
hostname: config.OPENCODE_HOSTNAME,
|
|
port,
|
|
timeout: config.OPENCODE_TIMEOUT_MS,
|
|
config: buildOpencodeConfig(),
|
|
});
|
|
} catch (error) {
|
|
if (isMissingOpencodeCli(error)) {
|
|
throw new Error(
|
|
"embedded mode requires the opencode2 CLI to be installed and available in PATH; otherwise set OPENCODE_MODE=client and provide OPENCODE_CLIENT_BASE_URL",
|
|
);
|
|
}
|
|
throw error;
|
|
}
|
|
|
|
logger.info(
|
|
{
|
|
elapsedMs: Math.max(0, Date.now() - startedAt),
|
|
hostname: config.OPENCODE_HOSTNAME,
|
|
baseUrl: runtime.url,
|
|
mode: config.OPENCODE_MODE,
|
|
},
|
|
"opencode2 server started in embedded mode",
|
|
);
|
|
|
|
this.closeServer = () => {
|
|
runtime.close();
|
|
};
|
|
|
|
return runtime.client;
|
|
}
|
|
}
|
|
|
|
export const opencodeRuntime = new OpencodeRuntimeAdapter();
|
|
|
|
function buildOpencodeConfig(): Record<string, unknown> {
|
|
return serviceOpencodeConfig(deepMerge(readProjectOpencodeConfig(), readEnvOpencodeConfig()), config.MODEL_CONFIG, config.CONVERSATION_WORKSPACE_ROOT);
|
|
}
|
|
|
|
function readProjectOpencodeConfig(): Record<string, unknown> {
|
|
const path = resolve(process.cwd(), "opencode.json");
|
|
if (!existsSync(path)) {
|
|
return {};
|
|
}
|
|
return parseConfigJson(readFileSync(path, "utf8"), path);
|
|
}
|
|
|
|
function readEnvOpencodeConfig(): Record<string, unknown> {
|
|
const content = process.env.OPENCODE_CONFIG_CONTENT;
|
|
if (!content?.trim()) {
|
|
return {};
|
|
}
|
|
return parseConfigJson(content, "OPENCODE_CONFIG_CONTENT");
|
|
}
|
|
|
|
function parseConfigJson(content: string, source: string): Record<string, unknown> {
|
|
const parsed = JSON.parse(content) as unknown;
|
|
if (typeof parsed !== "object" || parsed === null || Array.isArray(parsed)) {
|
|
throw new Error(`${source} must contain a JSON object`);
|
|
}
|
|
return parsed as Record<string, unknown>;
|
|
}
|
|
|
|
function deepMerge(
|
|
left: Record<string, unknown>,
|
|
right: Record<string, unknown>,
|
|
): Record<string, unknown> {
|
|
const next = { ...left };
|
|
for (const [key, value] of Object.entries(right)) {
|
|
const existing = next[key];
|
|
if (isPlainObject(existing) && isPlainObject(value)) {
|
|
next[key] = deepMerge(existing, value);
|
|
} else {
|
|
next[key] = value;
|
|
}
|
|
}
|
|
return next;
|
|
}
|
|
|
|
function isPlainObject(value: unknown): value is Record<string, unknown> {
|
|
return typeof value === "object" && value !== null && !Array.isArray(value);
|
|
}
|
|
|
|
function isMissingOpencodeCli(error: unknown): error is NodeJS.ErrnoException {
|
|
return (
|
|
typeof error === "object" &&
|
|
error !== null &&
|
|
"code" in error &&
|
|
(error as NodeJS.ErrnoException).code === "ENOENT"
|
|
);
|
|
}
|
|
|
|
function delay(ms: number) {
|
|
return new Promise<void>((resolve) => {
|
|
setTimeout(resolve, ms);
|
|
});
|
|
}
|