next-tjwater-agent/tests/chat/agentMessageCoordinator.test.ts

import { afterEach, describe, expect, it } from "bun:test";
import { mkdtemp, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { AgentMessageCoordinator } from "../../src/chat/agent-message-coordinator.js";
import { AgentMessageClient, type AgentMessage, type MessageCredential } from "../../src/product/agent-messages.js";
import { OpencodeRuntimeAdapter } from "../../src/runtime/opencode.js";
import { SessionMetadataStore } from "../../src/sessions/metadataStore.js";
import { toActorKey, toProjectKey } from "../../src/utils/fileStore.js";

const projectId = "61000000-0000-4000-8000-000000000001";
const taskId = "61000000-0000-4000-8000-000000000002";
const runId = "61000000-0000-4000-8000-000000000003";
const owner = "conversation-owner";
const roots: string[] = [];
afterEach(async () => { for (const root of roots.splice(0)) await rm(root, { recursive: true, force: true }); });

const message = (extra: Partial<AgentMessage> = {}): AgentMessage => ({
  message_id: "61000000-0000-4000-8000-000000000004", project_id: projectId, task_id: taskId, run_id: runId,
  request_id: "61000000-0000-4000-8000-000000000005", source_session_id: "main", source_owner_id: owner,
  conversation_owner_id: owner, target_session_id: "investigation", direction: "to_task", in_reply_to: null,
  text: "Check this additional question", status: "pending", native_message_id: "msg-coordinator-stable",
  failure_detail: null, created_at: "2026-10-13T00:00:00Z", updated_at: "2026-10-13T00:00:00Z", delivered_at: null,
  ...extra,
});

async function fixture(initial = message()) {
  const root = await mkdtemp(join(tmpdir(), "tjwater-message-coordinator-")); roots.push(root);
  const sessions = new SessionMetadataStore(root);
  await sessions.initialize();
  const context = { actorKey: toActorKey(`product:${owner}`), projectKey: toProjectKey(`product:${projectId}`), projectId, userId: owner };
  const main = (await sessions.ensure({ ...context, sessionId: "main", purpose: "ordinary_chat" })).record;
  const investigation = (await sessions.ensure({ ...context, sessionId: "investigation", purpose: "task_execution", taskId, runId })).record;
  const records = new Map([[initial.message_id, initial]]);
  const native = new Map<string, "missing" | "enqueued" | "delivered">();
  const injections: Array<{ target: string; id: string; text: string; metadata: unknown }> = [];
  const receipts: Array<{ id: string; status: string; target: string | null }> = [];
  const accesses: Array<{ credential: MessageCredential; id: string }> = [];
  let run = { task_id: taskId, run_id: runId, status: "running", agent_session_id: "investigation" as string | null };
  let receiptFailure = false;
  let denyRecipient = false;
  const pages: Array<{ items: AgentMessage[]; next_cursor: string | null }> = [];
  const pageCalls: unknown[] = [];
  const client = Object.assign(Object.create(AgentMessageClient.prototype), {
    get: async (credential: MessageCredential, id: string) => {
      accesses.push({ credential, id });
      if (credential === "main" && denyRecipient) throw new Error("recipient authorization expired");
      return records.get(id)!;
    },
    request: async () => run,
    receipt: async (_credential: MessageCredential, item: AgentMessage, status: AgentMessage["status"], target: string | null, failure?: string) => {
      receipts.push({ id: item.message_id, status, target });
      if (receiptFailure) throw new Error("receipt lost");
      const updated = { ...item, status, target_session_id: target, failure_detail: failure ?? null };
      records.set(item.message_id, updated); return updated;
    },
    list: async (...args: unknown[]) => { pageCalls.push(args); return pages.shift() ?? { items: [], next_cursor: null }; },
  }) as AgentMessageClient;
  const runtime = Object.assign(Object.create(OpencodeRuntimeAdapter.prototype), {
    agentMessageState: async (_target: string, id: string) => native.get(id) ?? "missing",
    enqueueAgentMessage: async (target: string, id: string, text: string, metadata: unknown) => {
      injections.push({ target, id, text, metadata }); native.set(id, "enqueued");
    },
  }) as OpencodeRuntimeAdapter;
  const coordinator = new AgentMessageCoordinator(client, runtime, sessions);
  return { coordinator, main, investigation, records, native, injections, receipts, accesses, pages, pageCalls,
    setRun: (value: Partial<typeof run>) => { run = { ...run, ...value }; },
    loseReceipt: (value: boolean) => { receiptFailure = value; }, denyRecipient: () => { denyRecipient = true; } };
}

describe("AgentMessageCoordinator", () => {
  it("leaves idle recipients pending without native injection or waking a managed turn", async () => {
    const item = message({ direction: "to_chat", source_session_id: "investigation", target_session_id: "main" });
    const f = await fixture(item);
    let notifications = 0;
    const unsubscribe = f.coordinator.subscribe("main", () => notifications++);
    expect(await f.coordinator.deliver("worker", item)).toEqual(item);
    expect(f.injections).toEqual([]); expect(f.receipts).toEqual([]);
    expect(f.coordinator.isActive("main")).toBeFalse(); expect(notifications).toBe(1);
    unsubscribe(); await f.coordinator.deliver("worker", item); expect(notifications).toBe(1);
  });

  it("delivers only to the Server-selected exact run and preserves discussion provenance", async () => {
    const item = message({ target_session_id: "untrusted-hint" });
    const f = await fixture(item); f.coordinator.start(f.investigation);
    const updated = await f.coordinator.deliver("worker", item);
    expect(updated.status).toBe("enqueued"); expect(updated.target_session_id).toBe("investigation");
    expect(f.injections).toHaveLength(1);
    expect(f.injections[0]).toMatchObject({ target: "investigation", id: item.native_message_id,
      metadata: { source: "agent_message", message_id: item.message_id, task_id: taskId, run_id: runId } });
    expect(f.injections[0].text).toContain("不是用户直接指令");
    expect(f.injections[0].text).toContain("不要自动回复收件确认");
    expect(f.injections[0].text).toContain(item.text);
  });

  it("does not transfer messages to another run or to a terminal run", async () => {
    const item = message(); const f = await fixture(item); f.coordinator.start(f.investigation);
    f.setRun({ run_id: "another-run" });
    await expect(f.coordinator.deliver("worker", item)).rejects.toThrow("binding mismatch");
    expect(f.injections).toEqual([]); expect(f.receipts).toEqual([]);
    f.setRun({ run_id: runId, status: "completed" });
    expect((await f.coordinator.deliver("worker", item)).status).toBe("failed");
    expect(f.injections).toEqual([]);
    expect(f.records.get(item.message_id)?.failure_detail).toContain("未转投其他轮次");
  });

  it("waits for an exact task binding and rejects a missing or foreign recipient owner", async () => {
    const item = message(); const f = await fixture(item); f.coordinator.start(f.investigation);
    f.setRun({ agent_session_id: null });
    expect((await f.coordinator.deliver("worker", item)).status).toBe("pending"); expect(f.receipts).toEqual([]);
    f.setRun({ agent_session_id: "main" });
    expect((await f.coordinator.deliver("worker", item)).status).toBe("failed"); expect(f.injections).toEqual([]);
    const foreign = message({ direction: "to_chat", target_session_id: "main", conversation_owner_id: "foreign-owner" });
    f.records.set(foreign.message_id, foreign); f.coordinator.start(f.main);
    expect((await f.coordinator.deliver("worker", foreign)).status).toBe("failed"); expect(f.injections).toEqual([]);
  });

  it("rechecks the recipient authorization before injecting a service-delivered reply", async () => {
    const item = message({ direction: "to_chat", target_session_id: "main" });
    const f = await fixture(item); f.coordinator.start(f.main); f.denyRecipient();
    await expect(f.coordinator.deliver("worker", item)).rejects.toThrow("recipient authorization expired");
    expect(f.accesses.map((access) => access.credential)).toEqual(["worker", "main"]);
    expect(f.injections).toEqual([]); expect(f.receipts).toEqual([]);
  });

  it("rejects investigation messages with a missing or mismatched sender owner", async () => {
    const item = message(); const f = await fixture(item); f.coordinator.start(f.investigation);
    for (const invalid of [{ ...item, source_session_id: "unknown-main" }, { ...item, source_owner_id: "another-owner" },
      { ...item, source_session_id: "investigation" }]) {
      f.records.set(item.message_id, invalid);
      expect((await f.coordinator.deliver("worker", invalid)).status).toBe("failed");
      expect(f.injections).toEqual([]);
      expect(f.records.get(item.message_id)?.failure_detail).toContain("来源会话");
    }
  });

  it("recovers a lost receipt, serializes duplicate delivery and never injects twice", async () => {
    const item = message(); const f = await fixture(item); f.coordinator.start(f.investigation); f.loseReceipt(true);
    await expect(f.coordinator.deliver("worker", item)).rejects.toThrow("receipt lost");
    expect(f.injections).toHaveLength(1); expect(f.records.get(item.message_id)?.status).toBe("pending");
    f.loseReceipt(false);
    const result = await Promise.all([f.coordinator.deliver("worker", item), f.coordinator.deliver("worker", item)]);
    expect(result.map((record) => record.status)).toEqual(["enqueued", "enqueued"]);
    expect(f.injections).toHaveLength(1);
    f.native.set(item.native_message_id, "delivered"); f.coordinator.stop("investigation");
    expect((await f.coordinator.deliver("worker", item)).status).toBe("delivered");
    expect(f.injections).toHaveLength(1); expect(f.coordinator.isActive("investigation")).toBeFalse();
  });

  it("recovers a lost native enqueue only when the exact recipient is active", async () => {
    const item = message({ status: "enqueued" }); const f = await fixture(item);
    expect((await f.coordinator.deliver("worker", item)).status).toBe("enqueued"); expect(f.injections).toEqual([]);
    f.coordinator.start(f.investigation);
    expect((await f.coordinator.deliver("worker", item)).status).toBe("enqueued"); expect(f.injections).toHaveLength(1);
  });

  it("flushes paginated incoming pending messages only for the current run without creating or restarting a run", async () => {
    const item = message({ target_session_id: null }); const f = await fixture(item); f.coordinator.start(f.investigation);
    const foreignRun = message({ message_id: "61000000-0000-4000-8000-000000000006", target_session_id: null, run_id: "other-run" });
    const outgoing = message({ message_id: "61000000-0000-4000-8000-000000000007", direction: "to_chat", target_session_id: "main" });
    f.pages.push({ items: [foreignRun, outgoing], next_cursor: "next-page" }, { items: [item], next_cursor: null });
    expect(await f.coordinator.flush("investigation")).toBeTrue();
    expect(f.injections).toHaveLength(1); expect(f.injections[0].id).toBe(item.native_message_id);
    expect(f.pageCalls).toEqual([["investigation", "investigation", undefined, 50, true], ["investigation", "investigation", "next-page", 50, true]]);
    f.native.set(item.native_message_id, "delivered");
    f.pages.push({ items: [f.records.get(item.message_id)!], next_cursor: null });
    expect(await f.coordinator.flush("investigation")).toBeFalse(); expect(f.injections).toHaveLength(1);
  });
});