fix: 增强长 turn liveness 可见性

2026-06-10 12:05:21 +08:00
parent 43d42fe087
commit 43c47d3fa9
7 changed files with 362 additions and 15 deletions
@@ -0,0 +1,132 @@
+import assert from "node:assert/strict";
+import { setTimeout as sleep } from "node:timers/promises";
+import { ManagerClient } from "../../mgr/client.js";
+import { startManagerServer } from "../../mgr/server.js";
+import { MemoryAgentRunStore } from "../../mgr/store.js";
+import type { JsonRecord } from "../../common/types.js";
+import { assertNoSecretLeak, type SelfTestCase, type SelfTestContext } from "../harness.js";
+import { summarizeQueueCommanderSnapshot } from "../../../scripts/src/cli.js";
+
+const selfTest: SelfTestCase = async (context: SelfTestContext) => {
+  const store = new MemoryAgentRunStore();
+  const server = await startManagerServer({ port: 0, host: "127.0.0.1", sourceCommit: "self-test", store });
+  try {
+    const client = new ManagerClient(server.baseUrl);
+
+    const tool = await createActiveRun(client, context, "timeout-liveness-tool", 120_000);
+    await client.post(`/api/v1/runs/${tool.runId}/events`, { type: "tool_call", payload: { commandId: tool.commandId, itemId: "tool_live", toolName: "commandExecution", status: "running", command: "hwpod workspace apply-patch" } });
+    const toolResult = await commandResult(client, tool);
+    const toolLive = toolResult.liveness as JsonRecord;
+    assert.equal(toolLive.phase, "waiting-tool");
+    assert.equal(((toolLive.lastActivity as JsonRecord).activityKind), "tool-in-flight");
+    assert.equal(((toolLive.lastActivity as JsonRecord).sourceSeq), 4);
+    assert.equal(((toolLive.timeoutBudget as JsonRecord).state), "within-budget");
+    assert.ok(Array.isArray(toolLive.recoveryActions));
+
+    const assistant = await createActiveRun(client, context, "timeout-liveness-assistant", 120_000);
+    await client.post(`/api/v1/runs/${assistant.runId}/events`, { type: "assistant_message", payload: { commandId: assistant.commandId, itemId: "msg_progress", progress: true, text: "正在生成 apply-patch，先汇总待改文件。" } });
+    const assistantLive = (await commandResult(client, assistant)).liveness as JsonRecord;
+    assert.equal(assistantLive.phase, "waiting-model-output");
+    assert.equal(((assistantLive.lastActivity as JsonRecord).activityKind), "assistant-progress");
+
+    const inactive = await createActiveRun(client, context, "timeout-liveness-inactive", 40);
+    await sleep(36);
+    const inactiveLive = (await commandResult(client, inactive)).liveness as JsonRecord;
+    assert.equal(inactiveLive.phase, "runner-stdio-inactive");
+    assert.ok(["approaching-hard-timeout", "overdue"].includes(String((inactiveLive.timeoutBudget as JsonRecord).state)));
+
+    const terminal = await createActiveRun(client, context, "timeout-liveness-terminal", 50);
+    await client.post(`/api/v1/runs/${terminal.runId}/events`, { type: "error", payload: { commandId: terminal.commandId, failureKind: "backend-timeout", phase: "turn:hard-timeout", message: "codex stdio turn hard timed out after 50ms" } });
+    await client.patch(`/api/v1/commands/${terminal.commandId}/status`, { terminalStatus: "failed", failureKind: "backend-timeout", failureMessage: "codex stdio turn hard timed out after 50ms" });
+    await client.patch(`/api/v1/runs/${terminal.runId}/status`, { terminalStatus: "failed", failureKind: "backend-timeout", failureMessage: "codex stdio turn hard timed out after 50ms" });
+    const terminalResult = await commandResult(client, terminal);
+    const terminalLive = terminalResult.liveness as JsonRecord;
+    assert.equal(terminalResult.terminalStatus, "failed");
+    assert.equal(terminalLive.phase, "terminal");
+    assert.equal(((terminalLive.timeoutBudget as JsonRecord).state), "timed-out");
+    assert.ok((terminalLive.recoveryActions as JsonRecord[]).some((action) => action.action === "resume-session"));
+    assert.ok((terminalLive.recoveryActions as JsonRecord[]).some((action) => action.action === "split-task"));
+
+    const session = await client.get(`/api/v1/sessions/${terminal.sessionId}?readerId=timeout-liveness`) as JsonRecord;
+    assert.equal(((session.liveness as JsonRecord).phase), "terminal");
+    assert.ok(Array.isArray(((session.supervisor as JsonRecord).recoveryActions)), "session show must keep terminal recovery actions");
+
+    const task = await client.post("/api/v1/queue/tasks", queueTask(context, terminal.sessionId, 50)) as JsonRecord;
+    store.updateQueueTaskAttempt(String(task.id), {
+      state: "running",
+      latestAttempt: { attemptId: "attempt_timeout_liveness", state: "running", runId: terminal.runId, commandId: terminal.commandId, runnerJobId: null, sessionId: terminal.sessionId, sessionPath: `/api/v1/sessions/${terminal.sessionId}` },
+      sessionPath: `/api/v1/sessions/${terminal.sessionId}`,
+    });
+    const commander = await client.get("/api/v1/queue/commander?queue=timeout-liveness&readerId=timeout-liveness") as JsonRecord;
+    const commanderItem = ((commander.items as JsonRecord[]) ?? []).find((item) => item.id === task.id) as JsonRecord;
+    assert.equal(((commanderItem.supervisor as JsonRecord).phase), "terminal");
+    assert.equal((((commanderItem.supervisor as JsonRecord).timeoutBudget as JsonRecord).state), "timed-out");
+    const commanderSummary = summarizeQueueCommanderSnapshot(commander, { limit: 5 });
+    const summaryItem = ((commanderSummary.items as JsonRecord[]) ?? []).find((item) => item.id === task.id) as JsonRecord;
+    assert.equal(((summaryItem.supervisor as JsonRecord).phase), "terminal");
+    assert.equal(JSON.stringify(commanderSummary).includes("hwpod workspace apply-patch"), false, "commander summary must stay compact and avoid dumping command bodies");
+    assertNoSecretLeak({ toolResult, assistantLive, inactiveLive, terminalResult, session, commanderSummary });
+
+    return { name: "timeout-liveness", tests: ["tool-in-flight-liveness", "assistant-progress-liveness", "stdio-inactive-timeout-budget", "terminal-timeout-recovery", "queue-commander-supervisor"] };
+  } finally {
+    await new Promise<void>((resolve) => server.server.close(() => resolve()));
+  }
+};
+
+async function createActiveRun(client: ManagerClient, context: SelfTestContext, sessionSuffix: string, timeoutMs: number): Promise<{ runId: string; commandId: string; sessionId: string }> {
+  const sessionId = `selftest-${sessionSuffix}`;
+  const run = await client.post("/api/v1/runs", runBody(context, sessionId, timeoutMs)) as JsonRecord;
+  const command = await client.post(`/api/v1/runs/${run.id}/commands`, { type: "turn", payload: { prompt: sessionSuffix }, idempotencyKey: sessionSuffix }) as JsonRecord;
+  await client.post(`/api/v1/runs/${run.id}/claim`, { runnerId: `runner_${sessionSuffix}`, leaseMs: 60_000 });
+  await client.post(`/api/v1/commands/${command.id}/ack`, {});
+  return { runId: String(run.id), commandId: String(command.id), sessionId };
+}
+
+async function commandResult(client: ManagerClient, item: { runId: string; commandId: string }): Promise<JsonRecord> {
+  return await client.get(`/api/v1/runs/${item.runId}/commands/${item.commandId}/result`) as JsonRecord;
+}
+
+function runBody(context: SelfTestContext, sessionId: string, timeoutMs: number): JsonRecord {
+  return {
+    tenantId: "unidesk",
+    projectId: "pikasTech/agentrun",
+    workspaceRef: { kind: "host-path", path: context.workspace },
+    sessionRef: { sessionId, conversationId: sessionId },
+    providerId: "G14",
+    backendProfile: "codex",
+    executionPolicy: executionPolicy(timeoutMs, context.codexHome),
+    traceSink: null,
+  };
+}
+
+function queueTask(context: SelfTestContext, sessionId: string, timeoutMs: number): JsonRecord {
+  return {
+    tenantId: "unidesk",
+    projectId: "pikasTech/agentrun",
+    queue: "timeout-liveness",
+    lane: "selftest",
+    title: "timeout liveness commander",
+    priority: 1,
+    backendProfile: "codex",
+    providerId: "G14",
+    workspaceRef: { kind: "host-path", path: context.workspace },
+    sessionRef: { sessionId, conversationId: sessionId },
+    executionPolicy: executionPolicy(timeoutMs, context.codexHome),
+    resourceBundleRef: null,
+    payload: { prompt: "timeout liveness commander" },
+    references: [],
+    metadata: {},
+  };
+}
+
+function executionPolicy(timeoutMs: number, codexHome: string): JsonRecord {
+  return {
+    sandbox: "workspace-write",
+    approval: "never",
+    timeoutMs,
+    network: "default",
+    secretScope: { allowCredentialEcho: false, providerCredentials: [{ profile: "codex", secretRef: { name: "agentrun-v01-provider-codex", keys: ["auth.json", "config.toml"], mountPath: codexHome } }] },
+  };
+}
+
+export default selfTest;