jiang
a7ddff7b64
Auto-approve sandbox Shell only for host-bound task execution, stage native results within their owning session, and preserve active lease state on response stream loss. The repeated failures came from interactive approval in unattended runs and transport cleanup invalidating execution state; focused regressions cover both boundaries.
395 lines
16 KiB
TypeScript
395 lines
16 KiB
TypeScript
import { afterEach, describe, expect, it } from "bun:test";
|
|
|
|
import {
|
|
bindOperationalTaskCredential,
|
|
humanTaskEvidenceCommandSchema,
|
|
hasOperationalTaskRun,
|
|
OperationalTaskClient,
|
|
proposedTaskActionSchema,
|
|
releaseOperationalTaskCredential,
|
|
} from "../../src/product/operational-task-client.js";
|
|
|
|
import { runBackgroundTaskExecution } from "../../src/routes/chatSessionRoutes.js";
|
|
|
|
const originalFetch = globalThis.fetch;
|
|
const taskId = "50000000-0000-4000-8000-000000000001";
|
|
const runId = "55000000-0000-4000-8000-000000000001";
|
|
const executionRequestId = "60000000-0000-4000-8000-000000000001";
|
|
const run = {
|
|
run_id: runId,
|
|
task_id: taskId,
|
|
generation: 1,
|
|
execution_request_id: executionRequestId,
|
|
status: "running",
|
|
revision: 2,
|
|
agent_session_id: "session-1",
|
|
lease_expires_at: "2026-09-23T01:01:00Z",
|
|
heartbeat_at: "2026-09-23T01:00:00Z",
|
|
started_at: "2026-09-23T01:00:00Z",
|
|
finished_at: null,
|
|
failure_code: null,
|
|
failure_detail: null,
|
|
dispatch_attempts: 0,
|
|
next_dispatch_at: "2026-09-23T01:00:00Z",
|
|
created_at: "2026-09-23T01:00:00Z",
|
|
updated_at: "2026-09-23T01:00:00Z",
|
|
};
|
|
|
|
const task = {
|
|
task_id: taskId,
|
|
task_type: "event_investigation",
|
|
workflow: "water_quality_complaint",
|
|
lifecycle_status: "open",
|
|
active_run: run,
|
|
status: "running",
|
|
revision: 2,
|
|
execution_request_id: executionRequestId,
|
|
title: "水质投诉调查",
|
|
summary: "调查投诉",
|
|
};
|
|
|
|
afterEach(() => {
|
|
globalThis.fetch = originalFetch;
|
|
releaseOperationalTaskCredential("runtime-1");
|
|
});
|
|
|
|
describe("OperationalTaskClient", () => {
|
|
it("requests a selected report while retaining the current task snapshot", async () => {
|
|
let requestedUrl = "";
|
|
globalThis.fetch = (async (input: Parameters<typeof fetch>[0]) => {
|
|
requestedUrl = String(input);
|
|
return Response.json({ task: { ...task, execution_generation: 2, latest_plan: null,
|
|
latest_report: { revision: 8, execution_generation: 2 }, evidence: [], actions: [], permissions: {} },
|
|
approved_data_scopes: [], selected_report: { revision: 3, execution_generation: 1, report: { title: "上一轮报告" } } });
|
|
}) as unknown as typeof fetch;
|
|
bindOperationalTaskCredential("runtime-1", { authorization: "Bearer transient-secret", projectId: "project-1" });
|
|
const client = new OperationalTaskClient("http://server.test");
|
|
const context = await client.getAgentContext("runtime-1", taskId, 3);
|
|
expect(new URL(requestedUrl).searchParams.get("report_revision")).toBe("3");
|
|
expect(context.selected_report?.execution_generation).toBe(1);
|
|
expect(context.task.execution_generation).toBe(2);
|
|
});
|
|
it("matches proposed action payloads to the generated Server contract", () => {
|
|
const common = {
|
|
request_id: "60000000-0000-4000-8000-000000000001",
|
|
kind: "create_work_order" as const,
|
|
reason: "需要现场采样",
|
|
evidence_refs: [],
|
|
node_id: "field_sampling",
|
|
};
|
|
expect(proposedTaskActionSchema.safeParse({
|
|
...common,
|
|
payload: {
|
|
title: "现场采样",
|
|
description: "采集二次供水出水和末端水样",
|
|
work_order_type: "water_quality_sampling",
|
|
priority: "urgent",
|
|
location: "联调演示小区",
|
|
},
|
|
}).success).toBe(true);
|
|
expect(proposedTaskActionSchema.safeParse({
|
|
...common,
|
|
payload: {
|
|
title: "现场采样",
|
|
summary: "字段名不属于后端合同",
|
|
priority: "high",
|
|
},
|
|
}).success).toBe(false);
|
|
expect(proposedTaskActionSchema.safeParse({
|
|
...common,
|
|
kind: "issue_command",
|
|
payload: {
|
|
title: "现场采样",
|
|
description: "错误动作类型不得复用工单载荷",
|
|
work_order_type: "water_quality_sampling",
|
|
},
|
|
}).success).toBe(false);
|
|
});
|
|
|
|
it("accepts only explicit human input on the collaboration evidence path", () => {
|
|
const evidence = {
|
|
request_id: "60000000-0000-4000-8000-000000000001",
|
|
evidence: {
|
|
evidence_id: "human-confirmation-1",
|
|
kind: "human_input",
|
|
summary: "现场人员确认已完成采样。",
|
|
source_label: "调度员补充",
|
|
observed_at: "2026-09-23T01:00:00Z",
|
|
},
|
|
};
|
|
expect(humanTaskEvidenceCommandSchema.safeParse(evidence).success).toBe(true);
|
|
expect(humanTaskEvidenceCommandSchema.safeParse({
|
|
...evidence,
|
|
evidence: { ...evidence.evidence, kind: "system_data" },
|
|
}).success).toBe(false);
|
|
});
|
|
|
|
it("uses an in-memory credential and converts plans to the server contract", async () => {
|
|
let request: Request | undefined;
|
|
globalThis.fetch = (async (input, init) => {
|
|
request = new Request(input, init);
|
|
if (request.url.endsWith(`/runs/${runId}/claim`)) return Response.json(run);
|
|
return Response.json({
|
|
task,
|
|
plan_revision: 1,
|
|
report_revision: null,
|
|
});
|
|
}) as typeof fetch;
|
|
bindOperationalTaskCredential("runtime-1", {
|
|
authorization: "Bearer transient-secret",
|
|
projectId: "project-1",
|
|
});
|
|
const client = new OperationalTaskClient("http://server.test");
|
|
await client.claimTask("runtime-1", taskId, runId, {
|
|
requestId: crypto.randomUUID(), expectedRevision: 1, sourceSessionId: "session-1",
|
|
});
|
|
await client.publishArtifacts("runtime-1", taskId, {
|
|
requestId: executionRequestId,
|
|
sourceSessionId: "session-1",
|
|
executionPlan: {
|
|
title: "调查",
|
|
reason: "更新",
|
|
nodes: [{ id: "READ", title: "读取", owner: "agent", status: "completed", detail: "完成", evidence: ["evidence-1"] }],
|
|
edges: [],
|
|
},
|
|
});
|
|
|
|
expect(request?.headers.get("authorization")).toBe("Bearer transient-secret");
|
|
expect(request?.headers.get("x-project-id")).toBe("project-1");
|
|
const body = await request?.json() as Record<string, unknown>;
|
|
expect(body).toMatchObject({
|
|
request_id: "60000000-0000-4000-8000-000000000001",
|
|
source_session_id: "session-1",
|
|
run_id: runId,
|
|
plan: {
|
|
schema: "execution-plan@1",
|
|
nodes: [{ evidence_refs: ["evidence-1"] }],
|
|
},
|
|
});
|
|
expect(JSON.stringify(body)).not.toContain("transient-secret");
|
|
});
|
|
|
|
it("publishes complete progress and presentation with the claimed run without browser receipts", async () => {
|
|
const requests: Request[] = [];
|
|
globalThis.fetch = (async (input, init) => {
|
|
const request = new Request(input, init);
|
|
requests.push(request);
|
|
if (request.url.endsWith("/claim")) return Response.json(run);
|
|
if (request.url.endsWith("/presentation-events")) return Response.json({
|
|
recorded: true, event_seq: 12, run_id: runId, execution_generation: 1,
|
|
});
|
|
return Response.json({ task, plan_revision: null, report_revision: null, progress_revision: 3 });
|
|
}) as typeof fetch;
|
|
bindOperationalTaskCredential("runtime-1", { authorization: "Bearer transient-secret", projectId: "project-1" });
|
|
const client = new OperationalTaskClient("http://server.test");
|
|
await client.claimTask("runtime-1", taskId, runId, {
|
|
requestId: executionRequestId, expectedRevision: 1, sourceSessionId: "session-1",
|
|
});
|
|
const progress = { summary: "正在查询投诉时段", diagram: null };
|
|
expect(await client.publishArtifacts("runtime-1", taskId, {
|
|
requestId: executionRequestId, sourceSessionId: "session-1", progress,
|
|
})).toMatchObject({ progress_revision: 3 });
|
|
expect(await requests[1]!.json()).toMatchObject({ progress, run_id: runId, source_session_id: "session-1" });
|
|
const command = { action: "set_overlay" as const, overlay_id: "complaints", title: "投诉位置", mode: "geometry" as const, evidence_id: "query-1" };
|
|
expect(await client.publishPresentation("runtime-1", taskId, {
|
|
requestId: executionRequestId, sourceSessionId: "session-1", command,
|
|
})).toEqual({ recorded: true, event_seq: 12, run_id: runId, execution_generation: 1 });
|
|
expect(await requests[2]!.json()).toEqual({ request_id: executionRequestId, run_id: runId, source_session_id: "session-1", command });
|
|
expect(requests).toHaveLength(3);
|
|
expect(requests[2]!.headers.get("authorization")).toBe("Bearer transient-secret");
|
|
});
|
|
|
|
it("injects the claimed run into task questions without changing its lease revision", async () => {
|
|
const requests: Request[] = [];
|
|
globalThis.fetch = (async (input, init) => {
|
|
const request = new Request(input, init); requests.push(request);
|
|
if (request.url.endsWith("/questions")) return Response.json({ question_request_id: crypto.randomUUID(), status: "pending", version: 1 });
|
|
return Response.json(run);
|
|
}) as typeof fetch;
|
|
bindOperationalTaskCredential("runtime-1", { authorization: "Bearer transient-secret", projectId: "project-1" });
|
|
const client = new OperationalTaskClient("http://server.test");
|
|
const questions = [{ question_id: "flow", title: "流量", body: "是否有计量记录？", options: [],
|
|
mode: "single" as const, allow_custom: true, required: true }];
|
|
await expect(client.askTaskUser("runtime-1", taskId, { request_id: executionRequestId,
|
|
sourceSessionId: "session-1", questions })).rejects.toThrow("task run context unavailable");
|
|
await client.claimTask("runtime-1", taskId, runId, { requestId: executionRequestId, expectedRevision: 1, sourceSessionId: "session-1" });
|
|
await client.askTaskUser("runtime-1", taskId, { request_id: executionRequestId, sourceSessionId: "session-1", questions });
|
|
expect(await requests[1]!.json()).toEqual({ request_id: executionRequestId, source_session_id: "session-1", run_id: runId, questions });
|
|
await client.heartbeatTask("runtime-1", taskId, runId, "session-1");
|
|
expect(await requests[2]!.json()).toMatchObject({ expected_revision: run.revision });
|
|
});
|
|
|
|
it("fails closed without a current product credential", async () => {
|
|
const client = new OperationalTaskClient("http://server.test");
|
|
expect(client.getAgentContext("runtime-1", "50000000-0000-4000-8000-000000000001"))
|
|
.rejects.toThrow("current product credential");
|
|
});
|
|
|
|
it("preserves the Server problem code for lease authority failures", async () => {
|
|
let call = 0;
|
|
globalThis.fetch = (async (_input, _init) => {
|
|
call += 1;
|
|
if (call === 1) return Response.json(run);
|
|
return Response.json({
|
|
type: "https://tjwater.example/problems/lease-expired",
|
|
title: "Conflict",
|
|
status: 409,
|
|
detail: "Agent lease has expired",
|
|
code: "LEASE_EXPIRED",
|
|
}, { status: 409 });
|
|
}) as typeof fetch;
|
|
bindOperationalTaskCredential("runtime-1", {
|
|
authorization: "Bearer transient-secret",
|
|
projectId: "project-1",
|
|
});
|
|
const client = new OperationalTaskClient("http://server.test");
|
|
await client.claimTask("runtime-1", taskId, runId, {
|
|
requestId: crypto.randomUUID(), expectedRevision: 1, sourceSessionId: "session-1",
|
|
});
|
|
|
|
expect(client.publishArtifacts(
|
|
"runtime-1",
|
|
taskId,
|
|
{
|
|
requestId: "60000000-0000-4000-8000-000000000001",
|
|
sourceSessionId: "session-1",
|
|
report: {
|
|
title: "失败",
|
|
stage: "final", outcome: "concluded", stop_reason: null, data_gaps: [],
|
|
current_conclusion: "租约已失效",
|
|
conclusion_boundary: "未继续写回",
|
|
evidence_refs: [],
|
|
recommendations: [],
|
|
pending_items: [],
|
|
},
|
|
},
|
|
)).rejects.toMatchObject({
|
|
status: 409,
|
|
code: "LEASE_EXPIRED",
|
|
message: "Agent lease has expired",
|
|
});
|
|
});
|
|
|
|
it("keeps the current task revision for heartbeats after an action-only response", async () => {
|
|
const requestBodies: Record<string, unknown>[] = [];
|
|
let call = 0;
|
|
globalThis.fetch = (async (_input, init) => {
|
|
if (init?.body) requestBodies.push(JSON.parse(String(init.body)) as Record<string, unknown>);
|
|
call += 1;
|
|
if (call === 1) return Response.json(run);
|
|
if (call === 2) return Response.json({
|
|
action_id: "70000000-0000-4000-8000-000000000001",
|
|
approval_status: "pending",
|
|
execution_status: "not_submitted",
|
|
});
|
|
return Response.json({ ...run, revision: 3 });
|
|
}) as typeof fetch;
|
|
bindOperationalTaskCredential("runtime-1", {
|
|
authorization: "Bearer transient-secret",
|
|
projectId: "project-1",
|
|
});
|
|
const client = new OperationalTaskClient("http://server.test");
|
|
|
|
await client.claimTask("runtime-1", taskId, runId, {
|
|
requestId: "60000000-0000-4000-8000-000000000002",
|
|
expectedRevision: 1,
|
|
sourceSessionId: "session-1",
|
|
});
|
|
await client.proposeAction("runtime-1", "50000000-0000-4000-8000-000000000001", {
|
|
request_id: "60000000-0000-4000-8000-000000000003",
|
|
kind: "create_work_order",
|
|
reason: "需要现场采样",
|
|
payload: {
|
|
title: "现场采样",
|
|
description: "采集水样",
|
|
work_order_type: "water_quality_sampling",
|
|
priority: "urgent",
|
|
},
|
|
evidence_refs: [],
|
|
});
|
|
await client.heartbeatTask(
|
|
"runtime-1",
|
|
taskId,
|
|
runId,
|
|
"session-1",
|
|
);
|
|
|
|
expect(requestBodies[2]).toMatchObject({
|
|
expected_revision: 2,
|
|
source_session_id: "session-1",
|
|
});
|
|
});
|
|
|
|
it("serializes artifact publication and heartbeat revision updates", async () => {
|
|
const requestBodies: Record<string, unknown>[] = [];
|
|
let releaseArtifact: (() => void) | undefined;
|
|
const artifactGate = new Promise<void>((resolve) => { releaseArtifact = resolve; });
|
|
let call = 0;
|
|
globalThis.fetch = (async (_input, init) => {
|
|
if (init?.body) requestBodies.push(JSON.parse(String(init.body)) as Record<string, unknown>);
|
|
call += 1;
|
|
if (call === 1) return Response.json(run);
|
|
if (call === 2) {
|
|
await artifactGate;
|
|
return Response.json({
|
|
task: { ...task, active_run: { ...run, revision: 3 } },
|
|
plan_revision: 1,
|
|
report_revision: null,
|
|
});
|
|
}
|
|
return Response.json({ ...run, revision: 4 });
|
|
}) as typeof fetch;
|
|
bindOperationalTaskCredential("runtime-1", {
|
|
authorization: "Bearer transient-secret",
|
|
projectId: "project-1",
|
|
});
|
|
const client = new OperationalTaskClient("http://server.test");
|
|
await client.claimTask("runtime-1", taskId, runId, {
|
|
requestId: executionRequestId,
|
|
expectedRevision: 1,
|
|
sourceSessionId: "session-1",
|
|
});
|
|
|
|
const publication = client.publishArtifacts("runtime-1", taskId, {
|
|
requestId: "60000000-0000-4000-8000-000000000002",
|
|
sourceSessionId: "session-1",
|
|
executionPlan: {
|
|
title: "调查计划",
|
|
reason: "核实告警",
|
|
nodes: [],
|
|
edges: [],
|
|
},
|
|
});
|
|
const heartbeat = client.heartbeatTask("runtime-1", taskId, runId, "session-1");
|
|
releaseArtifact?.();
|
|
await Promise.all([publication, heartbeat]);
|
|
|
|
expect(requestBodies[2]).toMatchObject({ expected_revision: 3 });
|
|
});
|
|
});
|
|
|
|
|
|
it.each([true, false])("retains lease state on background stream loss only while running=%s", async (running) => {
|
|
const credential = { authorization: "Bearer fixture", projectId: "project-1" };
|
|
bindOperationalTaskCredential("runtime-1", credential);
|
|
globalThis.fetch = (async () => Response.json(run)) as unknown as typeof fetch;
|
|
const client = new OperationalTaskClient("http://server.test");
|
|
await client.claimTask("runtime-1", taskId, runId, {
|
|
requestId: executionRequestId, expectedRevision: 1, sourceSessionId: "session-1",
|
|
});
|
|
expect(hasOperationalTaskRun("runtime-1")).toBe(true);
|
|
globalThis.fetch = (async () => new Response(new ReadableStream({
|
|
start(controller) { controller.error(new Error("fixture stream disconnected")); },
|
|
}))) as unknown as typeof fetch;
|
|
await runBackgroundTaskExecution({ sessionId: "runtime-1", credential, model: "fixture/model",
|
|
isRunning: () => running });
|
|
expect(hasOperationalTaskRun("runtime-1")).toBe(running);
|
|
if (running) {
|
|
globalThis.fetch = (async (_input: Parameters<typeof fetch>[0], init?: RequestInit) => {
|
|
expect(JSON.parse(String(init?.body)).expected_revision).toBe(2);
|
|
return Response.json({ ...run, revision: 3 });
|
|
}) as unknown as typeof fetch;
|
|
await client.heartbeatTask("runtime-1", taskId, runId, "session-1");
|
|
}
|
|
});
|