fix: shorten AgentRun admission and stabilize Workbench projection (#1909)
* fix: shorten AgentRun chat admission path * fix: stabilize Workbench turn projection display
This commit is contained in:
Lyon
@@ -274,8 +274,8 @@ export async function submitAgentRunChatTurn({ params = {}, options = {}, traceI
|
||||
type: "request",
|
||||
status: "accepted",
|
||||
label: "agentrun:request:accepted",
|
||||
message: "HWLAB Code Agent request accepted by the AgentRun v0.1 adapter; hwlab-cloud-api will reuse an active runner when the HWLAB session has an active AgentRun reuse window, otherwise it will create run/command/runner-job over the k3s Service DNS.",
|
||||
waitingFor: "agentrun-run-reuse-or-create",
|
||||
message: "HWLAB Code Agent request accepted by the AgentRun v0.1 adapter; hwlab-cloud-api will create a fresh run/command tied to the session-scoped AgentRun PVC, then ensure the runner Job outside the browser admission path.",
|
||||
waitingFor: "agentrun-run-create",
|
||||
adapter: ADAPTER_ID,
|
||||
managerHost: new URL(managerUrl).hostname,
|
||||
...codeAgentOtelTraceFields(traceId, env),
|
||||
@@ -382,7 +382,7 @@ export async function submitAgentRunChatTurn({ params = {}, options = {}, traceI
|
||||
let sessionId = baseSessionId;
|
||||
let sessionReset = false;
|
||||
let finalDispatchParams = params;
|
||||
let run, command, runnerJob, mapping;
|
||||
let run, command, mapping;
|
||||
for (let attempt = 0; attempt < 2; attempt += 1) {
|
||||
let attemptParams = params;
|
||||
if (sessionReset) {
|
||||
@@ -458,36 +458,93 @@ export async function submitAgentRunChatTurn({ params = {}, options = {}, traceI
|
||||
runId, commandId, backendProfile, waitingFor: "agentrun-runner-job-create", valuesPrinted: false,
|
||||
}, backendProfile));
|
||||
const runnerJobInput = buildAgentRunRunnerJobInput({ env, traceId, commandId, ownerApiKey, toolCapabilities, backendProfile });
|
||||
try {
|
||||
const runnerJobStartedAt = Date.now();
|
||||
runnerJob = await agentRunDispatchJson({ fetchImpl, managerUrl, path: "/api/v1/runs/" + encodeURIComponent(runId) + "/runner-jobs", method: "POST", body: runnerJobInput, timeoutMs, env, traceStore, traceId, backendProfile, runId, commandId, stage: "runner-job-create" });
|
||||
void emitCodeAgentOtelSpan("agentrun_runner_job_create", traceId, env, { startTimeMs: runnerJobStartedAt, attributes: { runId, commandId, backendProfile, managerUrl, jobName: runnerJob?.jobName ?? runnerJob?.jobIdentity?.name ?? null } });
|
||||
} catch (error) {
|
||||
if (attempt === 0 && await shouldResetSessionAfterEviction("session-store-evicted", error?.message)) {
|
||||
sessionReset = true;
|
||||
continue;
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
mapping = agentRunMapping({ env, managerUrl, backendProfile, run, command, runnerJob, traceId, startedAt, params: dispatchParams });
|
||||
mapping = agentRunMapping({ env, managerUrl, backendProfile, run, command, runnerJob: null, traceId, startedAt, params: dispatchParams });
|
||||
traceStore.append(traceId, agentRunTraceEvent({
|
||||
type: "backend",
|
||||
status: "running",
|
||||
label: "agentrun:runner-job:queued",
|
||||
message: "AgentRun run/command admission is durable; hwlab-cloud-api will create the runner Job asynchronously so the browser request is not blocked by Kubernetes Job creation.",
|
||||
runId: mapping.runId,
|
||||
commandId: mapping.commandId,
|
||||
backendProfile,
|
||||
waitingFor: "agentrun-runner-job-create",
|
||||
valuesPrinted: false
|
||||
}, mapping));
|
||||
scheduleAgentRunRunnerJobCreateAfterAdmission({ fetchImpl, managerUrl, runnerJobInput, timeoutMs, env, traceStore, traceId, backendProfile, run, command, mapping, options, params: dispatchParams });
|
||||
finalDispatchParams = dispatchParams;
|
||||
break;
|
||||
}
|
||||
return decorateAgentRunRunningResult({ base: initialAgentRunChatResult({ params: finalDispatchParams, options, traceId }), mapping, traceStore, traceId });
|
||||
}
|
||||
|
||||
function scheduleAgentRunRunnerJobCreateAfterAdmission(args) {
|
||||
setTimeout(() => {
|
||||
void ensureAgentRunRunnerJobCreateAfterAdmission(args).catch((error) => {
|
||||
markAgentRunRunnerJobCreateFailedAfterAdmission(args, error);
|
||||
});
|
||||
}, 0);
|
||||
}
|
||||
|
||||
async function ensureAgentRunRunnerJobCreateAfterAdmission({ fetchImpl, managerUrl, runnerJobInput, timeoutMs, env = process.env, traceStore = defaultCodeAgentTraceStore, traceId, backendProfile, mapping = {}, options = {}, params = {} }) {
|
||||
const runId = requiredString(mapping.runId, "mapping.runId");
|
||||
const commandId = requiredString(mapping.commandId, "mapping.commandId");
|
||||
const runnerJobStartedAt = Date.now();
|
||||
const runnerJob = await agentRunDispatchJson({ fetchImpl, managerUrl, path: "/api/v1/runs/" + encodeURIComponent(runId) + "/runner-jobs", method: "POST", body: runnerJobInput, timeoutMs, env, traceStore, traceId, backendProfile, runId, commandId, stage: "runner-job-create-async" });
|
||||
const nextMapping = agentRunMappingWithRunnerJob(mapping, runnerJob);
|
||||
void emitCodeAgentOtelSpan("agentrun_runner_job_create_async", traceId, env, { startTimeMs: runnerJobStartedAt, attributes: { runId, commandId, backendProfile, managerUrl, jobName: nextMapping.jobName ?? null } });
|
||||
traceStore.append(traceId, agentRunTraceEvent({
|
||||
type: "backend",
|
||||
status: "running",
|
||||
label: "agentrun:runner-job:created",
|
||||
message: `AgentRun runner Job ${mapping.jobName ?? "unknown"} created in namespace ${mapping.namespace ?? DEFAULT_RUNNER_NAMESPACE}.`,
|
||||
runId: mapping.runId,
|
||||
commandId: mapping.commandId,
|
||||
attemptId: mapping.attemptId,
|
||||
runnerId: mapping.runnerId,
|
||||
jobName: mapping.jobName,
|
||||
namespace: mapping.namespace,
|
||||
message: `AgentRun runner Job ${nextMapping.jobName ?? "unknown"} created in namespace ${nextMapping.namespace ?? DEFAULT_RUNNER_NAMESPACE}.`,
|
||||
runId: nextMapping.runId,
|
||||
commandId: nextMapping.commandId,
|
||||
attemptId: nextMapping.attemptId,
|
||||
runnerId: nextMapping.runnerId,
|
||||
jobName: nextMapping.jobName,
|
||||
namespace: nextMapping.namespace,
|
||||
waitingFor: "agentrun-result",
|
||||
valuesPrinted: false
|
||||
}, mapping));
|
||||
return decorateAgentRunRunningResult({ base: initialAgentRunChatResult({ params: finalDispatchParams, options, traceId }), mapping, traceStore, traceId });
|
||||
}, nextMapping));
|
||||
const current = options.codeAgentChatResults?.get?.(traceId);
|
||||
const base = current && typeof current === "object" ? current : initialAgentRunChatResult({ params, options, traceId });
|
||||
options.codeAgentChatResults?.set?.(traceId, decorateAgentRunRunningResult({ base: { ...base, agentRun: nextMapping }, mapping: nextMapping, traceStore, traceId }));
|
||||
}
|
||||
|
||||
function markAgentRunRunnerJobCreateFailedAfterAdmission({ traceStore = defaultCodeAgentTraceStore, traceId, backendProfile, mapping = {}, options = {}, params = {} }, error) {
|
||||
const failureKind = agentRunDispatchFailureKind(error);
|
||||
const retryAttempt = Number(error?.retryAttempt ?? 0);
|
||||
const retryMax = Number(error?.retryMax ?? agentRunDispatchRetryPolicy().maxRetries);
|
||||
const message = `AgentRun runner Job creation failed after durable run/command admission; Code Agent stopped this turn instead of silently waiting. failureKind=${failureKind}`;
|
||||
traceStore.append(traceId, agentRunTraceEvent({
|
||||
type: "error",
|
||||
eventType: "error",
|
||||
status: "failed",
|
||||
label: `agentrun:runner-job:create-async-failed:${failureKind}`,
|
||||
errorCode: failureKind,
|
||||
failureKind,
|
||||
retryAttempt: Number.isFinite(retryAttempt) ? retryAttempt : 0,
|
||||
retryMax: Number.isFinite(retryMax) ? retryMax : null,
|
||||
retryExhausted: error?.retryExhausted === true,
|
||||
runId: mapping.runId ?? null,
|
||||
commandId: mapping.commandId ?? null,
|
||||
waitingFor: "agentrun-runner-job-failed",
|
||||
terminal: true,
|
||||
message,
|
||||
valuesPrinted: false
|
||||
}, mapping || backendProfile));
|
||||
const current = options.codeAgentChatResults?.get?.(traceId);
|
||||
const base = current && typeof current === "object"
|
||||
? current
|
||||
: decorateAgentRunRunningResult({ base: initialAgentRunChatResult({ params, options, traceId }), mapping, traceStore, traceId });
|
||||
const failed = agentRunResultToCodeAgentPayload({
|
||||
base: { ...base, agentRun: { ...mapping, status: "runner-job-failed", commandState: "failed", terminalStatus: "failed", failureKind, valuesPrinted: false } },
|
||||
result: { terminalStatus: "failed", failureKind, failureMessage: message, runId: mapping.runId ?? null, commandId: mapping.commandId ?? null, valuesPrinted: false },
|
||||
traceStore,
|
||||
traceId,
|
||||
appendResultEvent: false
|
||||
});
|
||||
options.codeAgentChatResults?.set?.(traceId, failed);
|
||||
}
|
||||
|
||||
async function agentRunDispatchJson({ fetchImpl, managerUrl, path, method = "GET", body = undefined, timeoutMs, env = process.env, traceStore = defaultCodeAgentTraceStore, traceId, backendProfile, runId = null, commandId = null, stage = "agentrun-dispatch", terminalOnFailure = true } = {}) {
|
||||
@@ -2810,6 +2867,7 @@ function resolveAgentRunBackendProfile(env = process.env, params = {}) {
|
||||
|
||||
function agentRunMapping({ env, managerUrl, backendProfile, run, command, runnerJob, traceId, startedAt, params = {} }) {
|
||||
const threadReused = Boolean(safeOpaqueId(params.threadId));
|
||||
const runnerJobCreated = Boolean(runnerJob);
|
||||
return {
|
||||
adapter: ADAPTER_ID,
|
||||
managerUrl,
|
||||
@@ -2822,7 +2880,7 @@ function agentRunMapping({ env, managerUrl, backendProfile, run, command, runner
|
||||
runnerJobId: runnerJob?.id ?? null,
|
||||
jobName: runnerJob?.jobName ?? runnerJob?.jobIdentity?.name ?? null,
|
||||
namespace: runnerJob?.namespace ?? runnerJob?.jobIdentity?.namespace ?? DEFAULT_RUNNER_NAMESPACE,
|
||||
status: "runner-job-created",
|
||||
status: runnerJobCreated ? "runner-job-created" : "runner-job-pending",
|
||||
runStatus: run.status ?? null,
|
||||
commandState: command.state ?? null,
|
||||
terminalStatus: null,
|
||||
@@ -2833,6 +2891,7 @@ function agentRunMapping({ env, managerUrl, backendProfile, run, command, runner
|
||||
conversationId: run?.sessionRef?.conversationId ?? null,
|
||||
threadId: run?.sessionRef?.threadId ?? null,
|
||||
runnerReused: false,
|
||||
runnerJobCount: runnerJobCreated ? 1 : 0,
|
||||
threadReused,
|
||||
persistentResume: threadReused,
|
||||
reused: false,
|
||||
@@ -2843,6 +2902,21 @@ function agentRunMapping({ env, managerUrl, backendProfile, run, command, runner
|
||||
};
|
||||
}
|
||||
|
||||
function agentRunMappingWithRunnerJob(mapping = {}, runnerJob = null) {
|
||||
return {
|
||||
...mapping,
|
||||
attemptId: runnerJob?.attemptId ?? runnerJob?.runner?.attemptId ?? mapping.attemptId ?? null,
|
||||
runnerId: runnerJob?.runnerId ?? runnerJob?.runner?.runnerId ?? mapping.runnerId ?? null,
|
||||
runnerJobId: runnerJob?.id ?? mapping.runnerJobId ?? null,
|
||||
jobName: runnerJob?.jobName ?? runnerJob?.jobIdentity?.name ?? mapping.jobName ?? null,
|
||||
namespace: runnerJob?.namespace ?? runnerJob?.jobIdentity?.namespace ?? mapping.namespace ?? DEFAULT_RUNNER_NAMESPACE,
|
||||
status: "runner-job-created",
|
||||
runnerJobCount: Math.max(1, Number(mapping.runnerJobCount ?? 0) + 1),
|
||||
updatedAt: nowIso(),
|
||||
valuesPrinted: false
|
||||
};
|
||||
}
|
||||
|
||||
function agentRunReusedMapping({ previous = {}, run = {}, command = {}, runnerJob = null, traceId, startedAt, backendProfile, managerUrl, env }) {
|
||||
const threadReused = Boolean(safeOpaqueId(run?.sessionRef?.threadId ?? previous.threadId));
|
||||
return {
|
||||
|
||||
Reference in New Issue
Block a user