gori-agent/test/gateway.test.ts

import assert from "node:assert/strict";
import test from "node:test";
import type { ConversationRuntime } from "../src/acp/types.js";
import type { PlatformAdapter } from "../src/core/adapter.js";
import { Gateway } from "../src/core/gateway.js";
import type { IncomingMessage, OutgoingMessage } from "../src/core/types.js";

class FakeRuntime implements ConversationRuntime {
  prompts = 0; cancelled = 0; resets = 0; promptError?: Error;
  private readonly releases: Array<() => void> = [];
  async prompt() {
    this.prompts++;
    if (this.promptError) throw this.promptError;
    await new Promise<void>((resolve) => { this.releases.push(resolve); });
    return { text: "done", botId: "test-bot", agentId: "kimi" };
  }
  async cancel() { this.cancelled++; this.releaseNext(); return true; }
  async reset() { this.resets++; }
  status() { return { bot: "test-bot", agent: "kimi", workspace: "/tmp", running: this.releases.length > 0, phase: "processing" }; }
  stats() { return { activeWorkers: 0, inFlight: 0, crashes: 0, persistedBindings: 0 }; }
  async shutdown() {}
  releaseNext() { this.releases.shift()?.(); }
}

const policy = { allowedUsers: [] as string[], allowedChats: [] as string[], requireMentionInGroup: false };
const message = (text: string): IncomingMessage => ({ platform: "qq", chatId: "chat", userId: "user", text });
const waitFor = async (condition: () => boolean) => {
  for (let count = 0; count < 50 && !condition(); count++) await new Promise((resolve) => setTimeout(resolve, 5));
  assert.equal(condition(), true);
};

function recordingAdapter(sent: string[] = []): PlatformAdapter {
  return { name: "test", async handleWebhook() { return {}; }, async sendMessage(outgoing) { sent.push(outgoing.text); } };
}

test("status, cancel, and help bypass the chat lock while new remains serial", async () => {
  const runtime = new FakeRuntime(); const gateway = new Gateway(policy, runtime); const adapter = recordingAdapter();
  const turn = gateway.receive(message("work"), adapter, { synchronous: true });
  await waitFor(() => runtime.prompts === 1);

  const status = await gateway.receive(message("/status"), adapter, { synchronous: true });
  assert.match(status.reply || "", /gatewayRunning=true/);
  assert.match(status.reply || "", /queued=0/);
  assert.match((await gateway.receive(message("/help"), adapter, { synchronous: true })).reply || "", /\/status/);
  const cancelled = await gateway.receive(message("/cancel"), adapter, { synchronous: true });
  assert.equal(cancelled.reply, "Cancellation requested.");
  await turn;

  const second = gateway.receive(message("work"), adapter, { synchronous: true });
  await waitFor(() => runtime.prompts === 2);
  const reset = gateway.receive(message("/new"), adapter, { synchronous: true });
  await new Promise((resolve) => setTimeout(resolve, 10));
  assert.equal(runtime.resets, 0);
  runtime.releaseNext();
  await second;
  await reset;
  assert.equal(runtime.resets, 1);
});

test("asynchronous messages acknowledge processing and queue state", async () => {
  const runtime = new FakeRuntime(); const gateway = new Gateway(policy, runtime); const sent: string[] = []; const adapter = recordingAdapter(sent);
  const first = gateway.receive(message("one"), adapter);
  await waitFor(() => runtime.prompts === 1);
  const second = gateway.receive(message("two"), adapter);
  await waitFor(() => sent.length >= 2);
  assert.deepEqual(sent.slice(0, 2), [
    "已收到，正在处理。可随时发送 /status 查看状态。",
    "已收到，已排队。可随时发送 /status 查看状态。"
  ]);
  const status = await gateway.receive(message("/status"), adapter, { synchronous: true });
  assert.match(status.reply || "", /queued=1/);
  assert.match(status.reply || "", /gatewayRunningSeconds=\d+/);
  runtime.releaseNext();
  await waitFor(() => runtime.prompts === 2);
  runtime.releaseNext();
  await Promise.all([first, second]);
});

test("prompt waits for acknowledgement and final reply uses sequence two", async () => {
  const runtime = new FakeRuntime(); const gateway = new Gateway(policy, runtime); const sent: OutgoingMessage[] = [];
  let releaseAcknowledgement!: () => void;
  const acknowledgement = new Promise<void>((resolve) => { releaseAcknowledgement = resolve; });
  const adapter: PlatformAdapter = {
    name: "test", async handleWebhook() { return {}; },
    async sendMessage(outgoing) {
      sent.push(outgoing);
      if (outgoing.replySequence === 1) await acknowledgement;
    }
  };
  const turn = gateway.receive({ ...message("work"), messageId: "message-1" }, adapter);
  await waitFor(() => sent.length === 1);
  assert.equal(runtime.prompts, 0);
  assert.equal(sent[0].replySequence, 1);
  releaseAcknowledgement();
  await waitFor(() => runtime.prompts === 1);
  runtime.releaseNext();
  await turn;
  assert.deepEqual(sent.map((outgoing) => [outgoing.text, outgoing.replyTo, outgoing.replySequence]), [
    ["已收到，正在处理。可随时发送 /status 查看状态。", "message-1", 1],
    ["done", "message-1", 2]
  ]);
});

test("acknowledgement failure is logged safely and does not block work", async () => {
  const runtime = new FakeRuntime(); const gateway = new Gateway(policy, runtime); const errors: string[] = [];
  let sends = 0;
  const adapter: PlatformAdapter = {
    name: "test", async handleWebhook() { return {}; },
    async sendMessage() { if (++sends === 1) throw new Error("sensitive provider response"); }
  };
  const originalError = console.error;
  console.error = (...args: unknown[]) => { errors.push(args.map(String).join(" ")); };
  try {
    const turn = gateway.receive(message("work"), adapter);
    await waitFor(() => runtime.prompts === 1);
    runtime.releaseNext();
    assert.equal((await turn).reply, "done");
  } finally {
    console.error = originalError;
  }
  assert.deepEqual(errors, ["Gateway acknowledgement send failed (platform=qq)"]);
});

test("asynchronous prompt errors reply with sequence two", async () => {
  const runtime = new FakeRuntime(); runtime.promptError = new Error("failed");
  const gateway = new Gateway(policy, runtime); const sent: OutgoingMessage[] = [];
  const adapter: PlatformAdapter = {
    name: "test", async handleWebhook() { return {}; }, async sendMessage(outgoing) { sent.push(outgoing); }
  };
  const result = await gateway.receive(message("work"), adapter);
  assert.equal(result.ok, false);
  assert.deepEqual(sent.map((outgoing) => outgoing.replySequence), [1, 2]);
});

test("single asynchronous command replies with sequence one", async () => {
  const runtime = new FakeRuntime(); const gateway = new Gateway(policy, runtime); const sent: OutgoingMessage[] = [];
  const adapter: PlatformAdapter = {
    name: "test", async handleWebhook() { return {}; }, async sendMessage(outgoing) { sent.push(outgoing); }
  };
  await gateway.receive(message("/help"), adapter);
  assert.equal(sent.length, 1);
  assert.equal(sent[0].replySequence, 1);
});

test("synchronous messages do not send acknowledgements", async () => {
  const runtime = new FakeRuntime(); const gateway = new Gateway(policy, runtime); const sent: string[] = []; const adapter = recordingAdapter(sent);
  const turn = gateway.receive(message("work"), adapter, { synchronous: true });
  await waitFor(() => runtime.prompts === 1);
  assert.deepEqual(sent, []);
  runtime.releaseNext();
  await turn;
  assert.deepEqual(sent, []);
});

test("fixed Bot status and retired role commands never reach ACP", async () => {
  const runtime = new FakeRuntime(); const gateway = new Gateway(policy, runtime); const adapter = recordingAdapter();
  for (const command of ["/roles", "/role ops", "/agents", "/agent ops"]) {
    assert.match((await gateway.receive(message(command), adapter, { synchronous: true })).reply || "", /removed in Config v3/);
  }
  assert.match((await gateway.receive(message("/status"), adapter, { synchronous: true })).reply || "", /bot=test-bot/);
  assert.equal(runtime.prompts, 0);
});