next-tjwater-agent/tests/product/operationalTaskClient.test.ts

import { afterEach, describe, expect, it } from "bun:test";

import {
  bindOperationalTaskCredential,
  humanTaskEvidenceCommandSchema,
  OperationalTaskClient,
  proposedTaskActionSchema,
  releaseOperationalTaskCredential,
} from "../../src/product/operational-task-client.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("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 });
  });
});