528 lines
19 KiB
JavaScript
528 lines
19 KiB
JavaScript
import { rmSync } from "node:fs";
|
|
import { mkdir, mkdtemp, readFile, writeFile } from "node:fs/promises";
|
|
import { tmpdir } from "node:os";
|
|
import { join } from "node:path";
|
|
import { Agent, setGlobalDispatcher } from "undici";
|
|
import { createAgentSession, DefaultResourceLoader, ModelRuntime, SessionManager, SettingsManager } from "@earendil-works/pi-coding-agent";
|
|
import { createAssistantMessageEventStream } from "@earendil-works/pi-ai";
|
|
import { Type } from "typebox";
|
|
|
|
// Node 自带 undici 的 header/body 超时默认 300 秒，短于文本任务 8 分钟和视频任务 60 分钟。
|
|
// 设为 0 后只由 Go 侧任务期限取消 bridge；这里不再提前切断仍在排队或生成中的步骤。
|
|
setGlobalDispatcher(new Agent({
|
|
connectTimeout: 30_000,
|
|
headersTimeout: 0,
|
|
bodyTimeout: 0,
|
|
}));
|
|
|
|
const providerID = "yingce";
|
|
const api = "openai-completions";
|
|
|
|
|
|
async function readRequest() {
|
|
let raw = "";
|
|
for await (const chunk of process.stdin) raw += chunk;
|
|
if (!raw.trim()) throw new Error("Agent runtime input is empty");
|
|
return JSON.parse(raw);
|
|
}
|
|
|
|
function emit(event, payload = {}) {
|
|
process.stdout.write(`${JSON.stringify({ event, ...payload })}\n`);
|
|
}
|
|
|
|
function createMessage(model, result) {
|
|
const content = [];
|
|
if (result.text) content.push({ type: "text", text: result.text });
|
|
for (const call of result.toolCalls ?? []) {
|
|
content.push({
|
|
type: "toolCall",
|
|
id: call.id,
|
|
...(call.item_id ? { itemId: call.item_id } : {}),
|
|
name: call.name,
|
|
arguments: call.arguments ?? {},
|
|
});
|
|
}
|
|
return {
|
|
role: "assistant",
|
|
content,
|
|
api: model.api,
|
|
provider: model.provider,
|
|
model: model.id,
|
|
usage: {
|
|
input: Number(result.usage?.input ?? 0),
|
|
output: Number(result.usage?.output ?? 0),
|
|
cacheRead: Number(result.usage?.cacheRead ?? 0),
|
|
cacheWrite: Number(result.usage?.cacheWrite ?? 0),
|
|
totalTokens: Number(result.usage?.totalTokens ?? 0),
|
|
cost: result.usage?.cost ?? { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
|
|
},
|
|
stopReason: content.some((item) => item.type === "toolCall") ? "toolUse" : "stop",
|
|
timestamp: Date.now(),
|
|
};
|
|
}
|
|
|
|
async function bridge(request, path, body, signal) {
|
|
const response = await fetch(new URL(path, request.bridgeURL), {
|
|
method: "POST",
|
|
headers: {
|
|
"content-type": "application/json",
|
|
authorization: `Bearer ${request.bridgeToken}`,
|
|
},
|
|
body: JSON.stringify(body),
|
|
signal,
|
|
});
|
|
const data = await response.json().catch(() => ({}));
|
|
if (!response.ok) throw new Error(data.error || `Agent bridge returned HTTP ${response.status}`);
|
|
return data;
|
|
}
|
|
|
|
async function run() {
|
|
const request = await readRequest();
|
|
if (!request.bridgeURL || !request.bridgeToken || !request.model?.id) {
|
|
throw new Error("Agent runtime is missing its bridge, session, or model configuration");
|
|
}
|
|
|
|
// 每次运行使用独立的空配置目录：SDK 的 auth.json / models.json / 锁文件都落在这里，
|
|
// 不读取宿主 ~/.pi，也不会执行预置 auth.json 里的 !command。运行结束即删除。
|
|
// 会话文件和工作目录同样放在这里：它们只是本轮的工作副本，真实来源是服务端
|
|
// 数据库里的会话快照（请求带 sessionJSONL，运行中回传）。不使用服务端的路径：
|
|
// 独立容器里没有服务端的数据目录；空工作目录也不会被发现任何项目级配置或技能。
|
|
const isolatedDir = await mkdtemp(join(tmpdir(), "agent-runtime-"));
|
|
process.env.HOME = isolatedDir;
|
|
process.env.PI_CODING_AGENT_DIR = isolatedDir;
|
|
process.on("exit", () => {
|
|
try { rmSync(isolatedDir, { recursive: true, force: true }); } catch {}
|
|
});
|
|
const agentDir = isolatedDir;
|
|
const workDir = join(isolatedDir, "work");
|
|
const sessionDir = join(isolatedDir, "sessions");
|
|
await mkdir(workDir, { mode: 0o700 });
|
|
await mkdir(sessionDir, { mode: 0o700 });
|
|
// Lifecycle state, not prompt wording, identifies SDK summarization calls.
|
|
// Keep the id through the closing event delivery so the requested/end pair
|
|
// can always be correlated, while the closing phase no longer labels a
|
|
// following normal model request as a summary request.
|
|
let activeCompaction = null;
|
|
const eventChain = { current: Promise.resolve() };
|
|
const modelRuntime = await ModelRuntime.create({
|
|
authPath: join(isolatedDir, "auth.json"),
|
|
modelsPath: join(isolatedDir, "models.json"),
|
|
});
|
|
modelRuntime.registerProvider(providerID, {
|
|
name: "影策模型任务",
|
|
baseUrl: "http://agent-runtime.invalid/v1",
|
|
apiKey: "managed-by-go-bridge",
|
|
api,
|
|
authHeader: false,
|
|
models: [{
|
|
id: request.model.id,
|
|
name: request.model.name || request.model.id,
|
|
api,
|
|
reasoning: Boolean(request.model.reasoning),
|
|
input: request.model.input?.length ? request.model.input : ["text", "image"],
|
|
cost: request.model.cost ?? { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 },
|
|
contextWindow: Math.max(1, Number(request.model.contextWindow || 128000)),
|
|
maxTokens: Math.max(1, Number(request.model.maxTokens || 8192)),
|
|
}],
|
|
streamSimple(model, context, options) {
|
|
const stream = createAssistantMessageEventStream();
|
|
const partial = {
|
|
role: "assistant",
|
|
content: [],
|
|
api: model.api,
|
|
provider: model.provider,
|
|
model: model.id,
|
|
usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, totalTokens: 0, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 } },
|
|
stopReason: "pending",
|
|
timestamp: Date.now(),
|
|
};
|
|
void (async () => {
|
|
stream.push({ type: "start", partial });
|
|
// 审批暂停时 session.abort() 是异步生效的，SDK 可能在它生效前就开下一轮。
|
|
// 这一轮不能发给服务端：运行正在等审批，再跑一步会多计一次费，
|
|
// 新的工具调用还会和待审批的调用对不上，让整轮以检查点校验失败收场。
|
|
if (pausedForApproval) {
|
|
stream.push({ type: "error", reason: "aborted", error: { ...partial, stopReason: "aborted", errorMessage: "paused for approval" } });
|
|
stream.end();
|
|
return;
|
|
}
|
|
try {
|
|
const purpose = activeCompaction?.phase === "summarizing" ? "compaction" : "conversation";
|
|
// Deliver lifecycle events/snapshots before scheduling the next request.
|
|
// In particular, compaction must be visible before its model call begins.
|
|
await eventChain.current;
|
|
const result = await bridge(request, "/model", {
|
|
modelId: model.id,
|
|
purpose,
|
|
systemPrompt: context.systemPrompt,
|
|
messages: context.messages,
|
|
tools: (context.tools ?? []).map((tool) => ({
|
|
name: tool.name,
|
|
description: tool.description,
|
|
parameters: tool.parameters,
|
|
})),
|
|
thinkingLevel: options?.reasoning ?? "off",
|
|
contextUsage: session.getContextUsage(),
|
|
}, options?.signal);
|
|
for (const text of result.steeringMessages ?? []) {
|
|
await session.steer(text);
|
|
}
|
|
const message = createMessage(model, result);
|
|
if (result.text) {
|
|
partial.content.push({ type: "text", text: result.text });
|
|
stream.push({ type: "text_start", contentIndex: 0, partial });
|
|
stream.push({ type: "text_delta", contentIndex: 0, delta: result.text, partial });
|
|
stream.push({ type: "text_end", contentIndex: 0, content: result.text, partial });
|
|
}
|
|
let contentIndex = result.text ? 1 : 0;
|
|
for (const call of result.toolCalls ?? []) {
|
|
const toolCall = message.content.find((item) => item.type === "toolCall" && item.id === call.id);
|
|
partial.content.push(toolCall);
|
|
stream.push({ type: "toolcall_start", contentIndex, partial });
|
|
stream.push({ type: "toolcall_end", contentIndex, toolCall, partial });
|
|
contentIndex++;
|
|
}
|
|
stream.push({ type: "done", reason: message.stopReason, message });
|
|
stream.end(message);
|
|
} catch (error) {
|
|
stream.push({ type: "error", reason: "error", error: {
|
|
...partial,
|
|
stopReason: "error",
|
|
errorMessage: error instanceof Error ? error.message : String(error),
|
|
} });
|
|
stream.end();
|
|
}
|
|
})();
|
|
return stream;
|
|
},
|
|
});
|
|
|
|
const model = modelRuntime.getModel(providerID, request.model.id);
|
|
if (!model) throw new Error("Agent model registration failed");
|
|
|
|
let session;
|
|
// 工具进入审批时主动中止本轮；这是暂停，不是失败。
|
|
let pausedForApproval = false;
|
|
const tools = (request.tools ?? []).map((tool) => ({
|
|
name: tool.name,
|
|
label: tool.label || tool.name,
|
|
description: tool.description || tool.name,
|
|
parameters: Type.Unsafe(tool.parameters ?? { type: "object", properties: {} }),
|
|
executionMode: tool.executionMode || "sequential",
|
|
async execute(toolCallId, params, signal) {
|
|
const result = await bridge(request, "/tool", {
|
|
callId: toolCallId,
|
|
name: tool.name,
|
|
arguments: params,
|
|
}, signal);
|
|
if (result.pause) {
|
|
pausedForApproval = true;
|
|
emit("approval_wait", { approvalId: result.approvalId, toolName: tool.name, callId: toolCallId });
|
|
session.abort();
|
|
return {
|
|
content: [{ type: "text", text: result.content || "操作正在等待用户审批。" }],
|
|
details: { approvalId: result.approvalId, paused: true },
|
|
isError: true,
|
|
terminate: true,
|
|
};
|
|
}
|
|
// 工具结果必须是非空文本：交回 "null" 会让模型返回空回复。
|
|
let text = typeof result.content === "string" ? result.content : JSON.stringify(result.content ?? result ?? {});
|
|
if (!text || text === "null") text = JSON.stringify({ error: "工具没有返回结果" });
|
|
return {
|
|
content: [{ type: "text", text }],
|
|
details: result.details,
|
|
isError: Boolean(result.isError),
|
|
terminate: Boolean(result.terminate),
|
|
};
|
|
},
|
|
}));
|
|
|
|
let sessionManager;
|
|
if (request.sessionJSONL) {
|
|
const sessionFile = join(sessionDir, "session.jsonl");
|
|
await writeFile(sessionFile, request.sessionJSONL, { mode: 0o600, flag: "wx" });
|
|
sessionManager = SessionManager.open(sessionFile, sessionDir, workDir);
|
|
} else {
|
|
sessionManager = SessionManager.create(workDir, sessionDir);
|
|
}
|
|
// 严格隔离：不探索文件系统的 packages/skills/extensions/项目配置。
|
|
// 技能由服务端的读取工具提供，这里不再按路径加载；projectTrusted=false 让 SDK
|
|
// 不读取 <cwd>/.pi 设置，也不向上层目录收集 .agents/skills。
|
|
// 同一个设置管理器也交给 createAgentSession，否则 SDK 会另建一个读文件的实例。
|
|
const settingsManager = SettingsManager.inMemory({}, { projectTrusted: false });
|
|
const resourceLoader = new DefaultResourceLoader({
|
|
cwd: workDir,
|
|
agentDir,
|
|
noExtensions: true,
|
|
noSkills: true,
|
|
noPromptTemplates: true,
|
|
noThemes: true,
|
|
noContextFiles: true,
|
|
additionalSkillPaths: [],
|
|
systemPrompt: request.systemPrompt,
|
|
settingsManager,
|
|
});
|
|
await resourceLoader.reload();
|
|
|
|
const contextWindow = Math.max(1, Number(request.model?.contextWindow || 128000));
|
|
const sessionConfig = {
|
|
cwd: workDir,
|
|
agentDir,
|
|
settingsManager,
|
|
modelRuntime,
|
|
model,
|
|
resourceLoader,
|
|
sessionManager,
|
|
customTools: tools,
|
|
};
|
|
|
|
if (request.profile) {
|
|
sessionConfig.profile = {
|
|
revision: request.profile.revision,
|
|
hash: request.profile.hash,
|
|
layers: request.profile.layers || [],
|
|
};
|
|
}
|
|
|
|
if (request.memory?.enabled) {
|
|
sessionConfig.memory = {
|
|
enabled: true,
|
|
storePath: request.memory.storePath,
|
|
userId: request.userId,
|
|
canvasId: request.canvasId,
|
|
maxEntries: request.memory.maxEntries || 1000,
|
|
};
|
|
}
|
|
|
|
if (request.canvas) {
|
|
sessionConfig.canvas = request.canvas;
|
|
}
|
|
|
|
if (request.features) {
|
|
sessionConfig.features = {
|
|
planning: request.features.planningEnabled ?? true,
|
|
forms: request.features.formsEnabled ?? true,
|
|
approval: request.features.approvalRequired ?? [],
|
|
};
|
|
}
|
|
|
|
if (request.permissions) {
|
|
sessionConfig.permissions = request.permissions;
|
|
}
|
|
|
|
// 只启用平台声明的工具，不使用 SDK 默认工具
|
|
const platformToolNames = tools.map(t => t.name);
|
|
sessionConfig.tools = platformToolNames;
|
|
|
|
({ session } = await createAgentSession(sessionConfig));
|
|
|
|
const compactionConfig = request.compaction || {
|
|
enabled: true,
|
|
strategy: "balanced",
|
|
reserveTokens: Math.max(1024, Math.ceil(contextWindow * 0.2)),
|
|
keepRecentTokens: Math.min(20000, Math.max(1024, Math.floor(contextWindow * 0.2))),
|
|
};
|
|
|
|
session.settingsManager.applyOverrides({
|
|
compaction: compactionConfig,
|
|
// 重试只由 Go 模型桥决定（按上游真实 HTTP 状态区分临时故障与参数错误）；
|
|
// 这里再重试一层会让 400 也被重发、让 500 被重试 3×3 次。
|
|
retry: { enabled: false },
|
|
});
|
|
|
|
let runtimeError = null;
|
|
// 事件链串行执行。已有一次快照在排队时，它执行时读到的已是最新文件，
|
|
// 不必为每条新条目再整份上传一次。
|
|
let snapshotQueued = false;
|
|
const enqueueEvent = (payload) => {
|
|
const pending = eventChain.current.then(() => bridge(request, "/event", payload));
|
|
eventChain.current = pending;
|
|
return pending;
|
|
};
|
|
const enqueueSessionSnapshot = () => {
|
|
if (snapshotQueued) return;
|
|
snapshotQueued = true;
|
|
eventChain.current = eventChain.current.then(async () => {
|
|
snapshotQueued = false;
|
|
const sessionFile = sessionManager.getSessionFile();
|
|
if (!sessionFile) return;
|
|
let sessionJSONL;
|
|
try {
|
|
sessionJSONL = await readFile(sessionFile, "utf8");
|
|
} catch (error) {
|
|
if (error?.code === "ENOENT") return;
|
|
throw error;
|
|
}
|
|
return bridge(request, "/event", {
|
|
type: "session_snapshot",
|
|
sessionJSONL,
|
|
contextUsage: session.getContextUsage(),
|
|
});
|
|
});
|
|
};
|
|
|
|
session.subscribe((event) => {
|
|
// 转发完整的 Pi 事件流到 Go 后端
|
|
if (event.type === "message_start") {
|
|
const message = event.message;
|
|
if (message?.role === "assistant") {
|
|
enqueueEvent({
|
|
type: "message_start",
|
|
role: "assistant",
|
|
messageId: event.messageId,
|
|
});
|
|
}
|
|
} else if (event.type === "message_delta") {
|
|
enqueueEvent({
|
|
type: "message_delta",
|
|
delta: event.delta,
|
|
messageId: event.messageId,
|
|
});
|
|
} else if (event.type === "message_end") {
|
|
const message = event.message;
|
|
if (message?.role !== "assistant") return;
|
|
if (pausedForApproval || message.stopReason === "aborted") return;
|
|
if (message.stopReason === "error" || message.errorMessage) {
|
|
runtimeError = new Error(message.errorMessage || "Agent assistant message failed");
|
|
return;
|
|
}
|
|
enqueueEvent({
|
|
type: "message_end",
|
|
message,
|
|
contextUsage: session.getContextUsage(),
|
|
});
|
|
} else if (event.type === "thinking_start") {
|
|
enqueueEvent({
|
|
type: "thinking_start",
|
|
thinkingId: event.thinkingId,
|
|
});
|
|
} else if (event.type === "thinking_delta") {
|
|
enqueueEvent({
|
|
type: "thinking_delta",
|
|
delta: event.delta,
|
|
thinkingId: event.thinkingId,
|
|
});
|
|
} else if (event.type === "thinking_end") {
|
|
enqueueEvent({
|
|
type: "thinking_end",
|
|
content: event.content,
|
|
thinkingId: event.thinkingId,
|
|
});
|
|
} else if (event.type === "planning_start") {
|
|
enqueueEvent({
|
|
type: "planning_start",
|
|
planId: event.planId,
|
|
});
|
|
} else if (event.type === "planning_update") {
|
|
enqueueEvent({
|
|
type: "planning_update",
|
|
plan: event.plan,
|
|
planId: event.planId,
|
|
});
|
|
} else if (event.type === "planning_end") {
|
|
enqueueEvent({
|
|
type: "planning_end",
|
|
plan: event.plan,
|
|
planId: event.planId,
|
|
});
|
|
} else if (event.type === "form_start") {
|
|
enqueueEvent({
|
|
type: "form_start",
|
|
form: event.form,
|
|
formId: event.formId,
|
|
});
|
|
} else if (event.type === "form_update") {
|
|
enqueueEvent({
|
|
type: "form_update",
|
|
form: event.form,
|
|
formId: event.formId,
|
|
});
|
|
} else if (event.type === "form_submit") {
|
|
enqueueEvent({
|
|
type: "form_submit",
|
|
answers: event.answers,
|
|
formId: event.formId,
|
|
});
|
|
} else if (event.type === "tool_call_start") {
|
|
enqueueEvent({
|
|
type: "tool_call_start",
|
|
toolCall: event.toolCall,
|
|
callId: event.callId,
|
|
});
|
|
} else if (event.type === "tool_call_end") {
|
|
enqueueEvent({
|
|
type: "tool_call_end",
|
|
result: event.result,
|
|
callId: event.callId,
|
|
});
|
|
} else if (event.type === "approval_requested") {
|
|
enqueueEvent({
|
|
type: "approval_requested",
|
|
approval: event.approval,
|
|
approvalId: event.approvalId,
|
|
});
|
|
} else if (event.type === "entry_appended") {
|
|
enqueueSessionSnapshot();
|
|
} else if (event.type === "compaction_start" || event.type === "compaction_end") {
|
|
// Persist the compaction entry before reporting it to the server. If event
|
|
// delivery fails, the summary and retained tail are still recoverable.
|
|
if (event.type === "compaction_start") activeCompaction = { id: crypto.randomUUID(), phase: "summarizing" };
|
|
const compaction = activeCompaction;
|
|
if (event.type === "compaction_end" && compaction) compaction.phase = "closing";
|
|
if (event.type === "compaction_end") enqueueSessionSnapshot();
|
|
const eventDelivery = enqueueEvent({
|
|
type: event.type,
|
|
compactionId: compaction?.id,
|
|
reason: event.reason,
|
|
aborted: event.aborted,
|
|
willRetry: event.willRetry,
|
|
errorMessage: event.errorMessage,
|
|
hasResult: Boolean(event.result),
|
|
tokensBefore: event.result?.tokensBefore,
|
|
estimatedTokensAfter: event.result?.estimatedTokensAfter,
|
|
contextUsage: session.getContextUsage(),
|
|
});
|
|
if (event.type === "compaction_end" && compaction) {
|
|
eventDelivery.then(
|
|
() => {
|
|
if (activeCompaction === compaction) activeCompaction = null;
|
|
},
|
|
() => {
|
|
if (activeCompaction === compaction) activeCompaction = null;
|
|
},
|
|
);
|
|
}
|
|
}
|
|
});
|
|
|
|
try {
|
|
try {
|
|
await session.prompt(request.prompt);
|
|
} catch (error) {
|
|
if (!pausedForApproval) throw error;
|
|
}
|
|
await eventChain.current;
|
|
if (runtimeError && !pausedForApproval) throw runtimeError;
|
|
const sessionFile = sessionManager.getSessionFile();
|
|
const sessionJSONL = await readFile(sessionFile, "utf8");
|
|
await bridge(request, "/event", {
|
|
type: "session_snapshot",
|
|
sessionJSONL,
|
|
contextUsage: session.getContextUsage(),
|
|
});
|
|
emit("settled", {
|
|
contextUsage: session.getContextUsage(),
|
|
sessionFile,
|
|
entries: sessionManager.getEntries().length,
|
|
});
|
|
} finally {
|
|
session.dispose();
|
|
}
|
|
}
|
|
|
|
run().catch((error) => {
|
|
emit("runtime_error", { message: error instanceof Error ? error.message : String(error) });
|
|
process.exitCode = 1;
|
|
});
|