root
379 lines
17 KiB
TypeScript
379 lines
17 KiB
TypeScript
import { randomUUID } from "node:crypto";
|
|
import { mkdir, readFile, rename, writeFile } from "node:fs/promises";
|
|
import path from "node:path";
|
|
|
|
import {
|
|
cancelAgentRunChatTurn,
|
|
describeAgentRunAdapterAvailability,
|
|
steerAgentRunChatTurn,
|
|
submitAgentRunChatTurn,
|
|
} from "../cloud/code-agent-agentrun-adapter.ts";
|
|
import { createCodeAgentTraceStore } from "../cloud/code-agent-trace-store.ts";
|
|
import type { WorkbenchApplication, WorkbenchEventPublisher, WorkbenchSteerInput, WorkbenchTurnInput } from "./contracts.ts";
|
|
|
|
type NativeState = {
|
|
sessions: Record<string, Record<string, any>>;
|
|
turns: Record<string, Record<string, any>>;
|
|
};
|
|
|
|
export function createNativeAgentRunWorkbenchApplication(options: {
|
|
stateFile: string;
|
|
env: Record<string, string | undefined>;
|
|
fetchImpl?: typeof fetch;
|
|
now?: () => string;
|
|
eventPublisher?: WorkbenchEventPublisher | null;
|
|
}): WorkbenchApplication {
|
|
const env = options.env;
|
|
const fetchImpl = options.fetchImpl ?? fetch;
|
|
const now = options.now ?? (() => new Date().toISOString());
|
|
const traceStore = createCodeAgentTraceStore();
|
|
const eventPublisher = options.eventPublisher ?? null;
|
|
const mutate = async <T>(fn: (state: NativeState) => T | Promise<T>) => {
|
|
const state = await load(options.stateFile);
|
|
const result = await fn(state);
|
|
await mkdir(path.dirname(options.stateFile), { recursive: true });
|
|
const temporary = `${options.stateFile}.${process.pid}.tmp`;
|
|
await writeFile(temporary, `${JSON.stringify(state, null, 2)}\n`, "utf8");
|
|
await rename(temporary, options.stateFile);
|
|
return result;
|
|
};
|
|
|
|
return {
|
|
async snapshot() { return load(options.stateFile); },
|
|
async health() {
|
|
const state = await load(options.stateFile);
|
|
const availability = describeAgentRunAdapterAvailability(env, { params: { providerProfile: env.HWLAB_CODE_AGENT_DEFAULT_PROVIDER_PROFILE } });
|
|
return {
|
|
serviceId: "hwlab-workbench-native-agentrun",
|
|
status: availability.ready ? "ready" : "blocked",
|
|
mode: "agentrun-native",
|
|
providerProfile: text(env.HWLAB_CODE_AGENT_DEFAULT_PROVIDER_PROFILE),
|
|
agentRun: availability.agentRun,
|
|
sessionCount: Object.keys(state.sessions).length,
|
|
turnCount: Object.keys(state.turns).length,
|
|
valuesPrinted: false,
|
|
};
|
|
},
|
|
async createSession({ actor, params }) {
|
|
return mutate((state) => {
|
|
const sessionId = validOptionalId(params.sessionId, "ses_") ?? `ses_${randomUUID()}`;
|
|
const conversationId = validOptionalId(params.conversationId, "cnv_") ?? `cnv_${randomUUID()}`;
|
|
const timestamp = now();
|
|
const session = {
|
|
sessionId,
|
|
conversationId,
|
|
projectId: text(params.projectId) || "pikasTech/HWLAB",
|
|
ownerUserId: actor.id,
|
|
ownerRole: actor.role ?? "user",
|
|
status: "idle",
|
|
providerProfile: text(params.providerProfile) || text(env.HWLAB_CODE_AGENT_DEFAULT_PROVIDER_PROFILE) || "gpt.pika",
|
|
createdAt: timestamp,
|
|
updatedAt: timestamp,
|
|
mode: "agentrun-native",
|
|
messages: [] as Record<string, unknown>[],
|
|
};
|
|
state.sessions[sessionId] = session;
|
|
return { status: "created", session, valuesRedacted: true, secretMaterialStored: false };
|
|
});
|
|
},
|
|
async admitTurn(input) {
|
|
const admitted = await mutate((state) => {
|
|
const session = ownedSession(state, input.sessionId, input.actor.id);
|
|
const timestamp = validTimestamp(input.params.submittedAt) ?? now();
|
|
const message = requiredText(input.params.message, "message");
|
|
state.turns[input.traceId] = {
|
|
traceId: input.traceId,
|
|
sessionId: input.sessionId,
|
|
conversationId: session.conversationId,
|
|
ownerUserId: input.actor.id,
|
|
status: "admitted",
|
|
message,
|
|
createdAt: timestamp,
|
|
updatedAt: timestamp,
|
|
terminal: false,
|
|
};
|
|
session.messages.push({ messageId: `msg_user_${input.traceId.slice(4)}`, role: "user", content: message, traceId: input.traceId, status: "completed", createdAt: timestamp });
|
|
session.status = "running";
|
|
session.lastTraceId = input.traceId;
|
|
session.updatedAt = timestamp;
|
|
return { ...input, params: { ...input.params, message, submittedAt: timestamp, providerProfile: text(input.params.providerProfile) || session.providerProfile } };
|
|
});
|
|
await eventPublisher?.publish({
|
|
traceId: admitted.traceId,
|
|
sessionId: admitted.sessionId,
|
|
event: {
|
|
type: "backend_status",
|
|
phase: "request-admitted",
|
|
status: "running",
|
|
summary: "请求已接纳，正在启动 AgentRun",
|
|
message: "请求已接纳，正在启动 AgentRun",
|
|
createdAt: admitted.params.submittedAt,
|
|
terminal: false,
|
|
valuesPrinted: false,
|
|
},
|
|
});
|
|
return admitted;
|
|
},
|
|
async dispatchTurn(input: WorkbenchTurnInput) {
|
|
const state = await load(options.stateFile);
|
|
const session = ownedSession(state, input.sessionId, input.actor.id);
|
|
const params = {
|
|
...input.params,
|
|
sessionId: input.sessionId,
|
|
conversationId: session.conversationId,
|
|
projectId: session.projectId,
|
|
ownerUserId: input.actor.id,
|
|
ownerRole: input.actor.role ?? "user",
|
|
providerProfile: text(input.params.providerProfile) || session.providerProfile,
|
|
shortConnection: true,
|
|
};
|
|
const publishingTrace = publishingDispatchTraceStore(traceStore, eventPublisher, input);
|
|
let admitted;
|
|
try {
|
|
admitted = await submitAgentRunChatTurn({ traceId: input.traceId, params, options: { env, fetchImpl }, traceStore: publishingTrace.store });
|
|
} catch (error) {
|
|
await publishingTrace.flush().catch(() => undefined);
|
|
throw error;
|
|
}
|
|
await publishingTrace.flush().catch(() => undefined);
|
|
await projectRunning(admitted, input.traceId);
|
|
return {
|
|
status: "running",
|
|
terminal: false,
|
|
traceId: input.traceId,
|
|
sessionId: input.sessionId,
|
|
agentRun: admitted.agentRun,
|
|
terminalAuthority: "hwlab.event.v1",
|
|
resultSynthesized: false,
|
|
valuesPrinted: false,
|
|
};
|
|
},
|
|
async steerTurn(input: WorkbenchSteerInput) {
|
|
const state = await load(options.stateFile);
|
|
const session = ownedSession(state, input.sessionId, input.actor.id);
|
|
const turn = state.turns[input.targetTraceId];
|
|
if (!turn || turn.ownerUserId !== input.actor.id || turn.sessionId !== input.sessionId) {
|
|
throw codedError("turn_not_found", "turn is not visible to the actor");
|
|
}
|
|
if (!turn.agentRun?.runId || !turn.agentRun?.commandId) {
|
|
throw codedError("turn_not_running", "turn has no active AgentRun command");
|
|
}
|
|
const currentResult = {
|
|
status: "running",
|
|
traceId: input.targetTraceId,
|
|
sessionId: input.sessionId,
|
|
conversationId: turn.conversationId ?? session.conversationId,
|
|
agentRun: turn.agentRun,
|
|
};
|
|
const steered = await steerAgentRunChatTurn({
|
|
traceId: input.targetTraceId,
|
|
currentResult,
|
|
params: {
|
|
...input.params,
|
|
sessionId: input.sessionId,
|
|
conversationId: currentResult.conversationId,
|
|
targetTraceId: input.targetTraceId,
|
|
steerTraceId: input.steerTraceId,
|
|
},
|
|
options: { env, fetchImpl, now },
|
|
traceStore,
|
|
});
|
|
if (!steered) throw codedError("turn_not_running", "turn has no active AgentRun command");
|
|
const actualTraceId = requiredText(steered.traceId, "steered.traceId");
|
|
const delivery = requiredText(steered.delivery, "steered.delivery");
|
|
await mutate((nextState) => {
|
|
const nextSession = ownedSession(nextState, input.sessionId, input.actor.id);
|
|
const targetTurn = nextState.turns[input.targetTraceId];
|
|
if (!targetTurn) throw codedError("turn_not_found", "turn is not visible to the actor");
|
|
if (delivery === "steer") {
|
|
Object.assign(targetTurn, { agentRun: steered.agentRun, updatedAt: now() });
|
|
return;
|
|
}
|
|
if (delivery !== "turn") throw codedError("agentrun_session_send_delivery_invalid", "AgentRun session send returned an unsupported delivery");
|
|
const timestamp = validTimestamp(input.params.submittedAt) ?? now();
|
|
nextState.turns[actualTraceId] = {
|
|
traceId: actualTraceId,
|
|
targetTraceId: input.targetTraceId,
|
|
requestedDelivery: "steer",
|
|
delivery: "turn",
|
|
sessionId: input.sessionId,
|
|
conversationId: targetTurn.conversationId ?? nextSession.conversationId,
|
|
ownerUserId: input.actor.id,
|
|
status: "running",
|
|
message: requiredText(input.params.message, "message"),
|
|
createdAt: timestamp,
|
|
updatedAt: timestamp,
|
|
terminal: false,
|
|
runId: steered.agentRun?.runId ?? null,
|
|
commandId: steered.agentRun?.commandId ?? null,
|
|
runnerJobId: steered.agentRun?.runnerJobId ?? null,
|
|
agentRun: steered.agentRun ?? null,
|
|
};
|
|
nextSession.status = "running";
|
|
nextSession.lastTraceId = actualTraceId;
|
|
nextSession.updatedAt = timestamp;
|
|
});
|
|
return {
|
|
...steered,
|
|
sessionId: input.sessionId,
|
|
terminalAuthority: "hwlab.event.v1",
|
|
resultSynthesized: false,
|
|
valuesPrinted: false,
|
|
};
|
|
},
|
|
async close() { await eventPublisher?.close?.(); },
|
|
async cancelTurn(input) {
|
|
const state = await load(options.stateFile);
|
|
const turn = state.turns[input.traceId];
|
|
if (!turn || turn.ownerUserId !== input.actor.id) throw codedError("turn_not_found", "turn is not visible to the actor");
|
|
const currentResult = turn.agentRun ? { status: "running", agentRun: turn.agentRun } : null;
|
|
const canceled = await cancelAgentRunChatTurn({ traceId: input.traceId, currentResult, options: { env, fetchImpl }, traceStore });
|
|
if (!canceled) throw codedError("turn_not_running", "turn has no active AgentRun command");
|
|
return {
|
|
status: "cancel-requested",
|
|
terminal: false,
|
|
traceId: input.traceId,
|
|
agentRun: canceled.agentRun,
|
|
terminalAuthority: "hwlab.event.v1",
|
|
resultSynthesized: false,
|
|
valuesPrinted: false,
|
|
};
|
|
},
|
|
async providerModelCatalog(profileValue) {
|
|
const profile = text(profileValue) || "gpt.pika";
|
|
const backendProfile = profile === "gpt.pika" ? "gpt-pika" : profile === "codex-api" ? "codex" : profile;
|
|
const fallback = nativeProviderModelFallback(profile);
|
|
try {
|
|
const managerUrl = requiredText(env.AGENTRUN_MGR_URL ?? env.HWLAB_CODE_AGENT_AGENTRUN_MGR_URL, "AGENTRUN_MGR_URL");
|
|
const apiKey = requiredText(env.AGENTRUN_API_KEY ?? env.HWLAB_CODE_AGENT_AGENTRUN_API_KEY, "AGENTRUN_API_KEY");
|
|
const response = await fetchImpl(`${managerUrl.replace(/\/+$/u, "")}/api/v1/provider-profiles/${encodeURIComponent(backendProfile)}/models`, { headers: { accept: "application/json", authorization: `Bearer ${apiKey}` } });
|
|
const payload = await response.json().catch(() => null) as Record<string, any> | null;
|
|
if (!response.ok || !payload) throw new Error(`AgentRun model catalog returned HTTP ${response.status}`);
|
|
const data = payload.data && typeof payload.data === "object" ? payload.data : payload;
|
|
return { ok: true, ...data, profile, valuesPrinted: false };
|
|
} catch (error: any) {
|
|
return { ...fallback, warning: { ...fallback.warning, message: error?.message ?? String(error) } };
|
|
}
|
|
},
|
|
};
|
|
|
|
async function projectRunning(result: any, traceId: string) {
|
|
return mutate((state) => {
|
|
const turn = state.turns[traceId];
|
|
if (!turn) throw codedError("turn_not_admitted", "turn must be admitted before dispatch");
|
|
Object.assign(turn, {
|
|
status: "running",
|
|
updatedAt: now(),
|
|
runId: result.agentRun?.runId ?? null,
|
|
commandId: result.agentRun?.commandId ?? null,
|
|
runnerJobId: result.agentRun?.runnerJobId ?? null,
|
|
agentRun: result.agentRun ?? null,
|
|
});
|
|
});
|
|
}
|
|
}
|
|
|
|
function nativeProviderModelFallback(profile: string) {
|
|
const defaults: Record<string, { model: string; effort: string }> = {
|
|
"gpt.pika": { model: "gpt-5.5", effort: "medium" },
|
|
"codex-api": { model: "gpt-5.5", effort: "medium" },
|
|
deepseek: { model: "deepseek-chat", effort: "medium" },
|
|
"dsflash-go": { model: "deepseek-v4-flash", effort: "xhigh" },
|
|
"minimax-m3": { model: "MiniMax-M3", effort: "medium" }
|
|
};
|
|
const selected = defaults[profile] ?? { model: profile, effort: "medium" };
|
|
return { ok: true, status: "degraded", profile, source: "profile-default", items: [{ id: selected.model, valuesPrinted: false }], count: 1, defaultModel: selected.model, reasoningEfforts: ["low", "medium", "high", "xhigh"], defaultReasoningEffort: selected.effort, warning: { code: "provider-model-catalog-unavailable", message: "AgentRun model catalog is unavailable", blocking: false, valuesPrinted: false }, valuesPrinted: false };
|
|
}
|
|
|
|
function publishingDispatchTraceStore(base: ReturnType<typeof createCodeAgentTraceStore>, publisher: WorkbenchEventPublisher | null, input: WorkbenchTurnInput) {
|
|
const pending: Promise<unknown>[] = [];
|
|
const store = {
|
|
...base,
|
|
append(traceId: string, event: Record<string, any> = {}, meta: Record<string, unknown> = {}) {
|
|
const normalized = base.append(traceId, event, meta);
|
|
if (publisher && dispatchVisibilityEvent(event)) {
|
|
pending.push(publishDispatchVisibility(publisher, input, { ...event, createdAt: normalized?.createdAt ?? event.createdAt }));
|
|
}
|
|
return normalized;
|
|
},
|
|
};
|
|
return { store, async flush() { await Promise.all(pending); } };
|
|
}
|
|
|
|
function dispatchVisibilityEvent(event: Record<string, any>): boolean {
|
|
const label = text(event.label);
|
|
const visible = label.startsWith("agentrun:dispatch-retry:")
|
|
|| label.startsWith("agentrun:dispatch-retry-exhausted:")
|
|
|| label.startsWith("agentrun:dispatch-failed:");
|
|
return visible && (event.willRetry === true || event.terminal === true);
|
|
}
|
|
|
|
async function publishDispatchVisibility(publisher: WorkbenchEventPublisher, input: WorkbenchTurnInput, event: Record<string, any>) {
|
|
const retrying = event.willRetry === true;
|
|
const exhausted = event.terminal === true;
|
|
const failureKind = text(event.failureKind || event.errorCode) || "agentrun-dispatch-error";
|
|
const summary = dispatchFailureSummary(failureKind);
|
|
await publisher.publish({
|
|
traceId: input.traceId,
|
|
sessionId: input.sessionId,
|
|
event: {
|
|
type: retrying ? "backend_status" : "error",
|
|
phase: retrying ? "dispatch-retry-scheduled" : "dispatch-retry-exhausted",
|
|
failureDomain: "infrastructure",
|
|
component: "agentrun-manager",
|
|
code: failureKind,
|
|
failureKind,
|
|
summary,
|
|
message: summary,
|
|
retryable: event.retryable === true,
|
|
willRetry: retrying,
|
|
retryPhase: retrying ? "retryScheduled" : "retryExhausted",
|
|
attempt: event.retryAttempt,
|
|
maxAttempts: event.retryMax,
|
|
backoffMs: event.retryDelayMs,
|
|
nextRetryAt: event.nextRetryAt,
|
|
createdAt: event.createdAt,
|
|
runId: event.runId ?? null,
|
|
commandId: event.commandId ?? null,
|
|
terminal: false,
|
|
valuesPrinted: false,
|
|
},
|
|
});
|
|
if (!exhausted) return;
|
|
await publisher.publish({
|
|
traceId: input.traceId,
|
|
sessionId: input.sessionId,
|
|
event: {
|
|
type: "terminal_status",
|
|
terminalStatus: "failed",
|
|
status: "failed",
|
|
failureKind,
|
|
errorCode: failureKind,
|
|
message: `${summary}，有限重试已耗尽。`,
|
|
createdAt: event.createdAt,
|
|
runId: event.runId ?? null,
|
|
commandId: event.commandId ?? null,
|
|
terminal: true,
|
|
valuesPrinted: false,
|
|
},
|
|
});
|
|
}
|
|
|
|
function dispatchFailureSummary(failureKind: string): string {
|
|
if (failureKind === "connection-refused" || failureKind === "agentrun-connect-failed") return "AgentRun 服务暂时不可达";
|
|
if (failureKind.includes("timeout")) return "AgentRun 服务响应超时";
|
|
return "AgentRun 调度基础设施暂时不可用";
|
|
}
|
|
|
|
async function load(stateFile: string): Promise<NativeState> {
|
|
const parsed = JSON.parse(await readFile(stateFile, "utf8").catch(() => '{"sessions":{},"turns":{}}'));
|
|
return { sessions: object(parsed.sessions), turns: object(parsed.turns) };
|
|
}
|
|
function object(value: unknown) { return value && typeof value === "object" && !Array.isArray(value) ? value as Record<string, Record<string, any>> : {}; }
|
|
function text(value: unknown) { return String(value ?? "").trim(); }
|
|
function validTimestamp(value: unknown) { const timestamp = text(value); return timestamp && Number.isFinite(Date.parse(timestamp)) ? new Date(Date.parse(timestamp)).toISOString() : null; }
|
|
function requiredText(value: unknown, field: string) { const result = text(value); if (!result) throw codedError("invalid_input", `${field} is required`); return result; }
|
|
function validOptionalId(value: unknown, prefix: string) { const id = text(value); if (!id) return null; if (!id.startsWith(prefix)) throw codedError("invalid_input", `${prefix.slice(0, -1)} id is invalid`); return id; }
|
|
function ownedSession(state: NativeState, sessionId: string, ownerUserId: string) { const session = state.sessions[sessionId]; if (!session || session.ownerUserId !== ownerUserId) throw codedError("session_not_found", "session is not visible to the actor"); return session; }
|
|
function codedError(code: string, message: string) { return Object.assign(new Error(message), { code }); }
|