gori-agent/test/gateway.test.ts

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

interface PendingTurn {
  request: ConversationRequest;
  resolve(text: string): void;
  reject(error: Error): void;
}

class FakeRuntime implements ConversationRuntime {
  readonly turns: PendingTurn[] = [];
  cancelled = 0;
  resets = 0;
  idleSeconds = 0;

  async prompt(request: ConversationRequest) {
    return new Promise<{ text: string; botId: string; agentId: string }>((resolve, reject) => {
      this.turns.push({
        request,
        resolve: (text) => resolve({ text, botId: "test-bot", agentId: "kimi" }),
        reject
      });
    });
  }
  async cancel() { this.cancelled++; this.turns.at(-1)?.resolve("cancelled"); return true; }
  async reset() { this.resets++; }
  status() {
    return {
      bot: "test-bot", agent: "kimi", workspace: "/tmp",
      running: this.turns.length > 0, phase: "processing", idleSeconds: this.idleSeconds
    };
  }
  stats() { return { activeWorkers: 0, inFlight: 0, crashes: 0, persistedBindings: 0 }; }
  async shutdown() {}
}

interface ScheduledTimer extends GatewayTimer {
  deadline: number;
  callback: () => void;
  cancelled: boolean;
  order: number;
}

class FakeClock {
  nowMs = 0;
  private order = 0;
  readonly timers: ScheduledTimer[] = [];

  readonly now = () => this.nowMs;
  readonly schedule = (callback: () => void, delayMs: number): GatewayTimer => {
    const timer: ScheduledTimer = {
      deadline: this.nowMs + delayMs,
      callback,
      cancelled: false,
      order: this.order++,
      cancel() { timer.cancelled = true; }
    };
    this.timers.push(timer);
    return timer;
  };

  set(timeMs: number) { this.nowMs = timeMs; }

  async advanceTo(timeMs: number) {
    while (true) {
      const next = this.timers
        .filter((timer) => !timer.cancelled && timer.deadline <= timeMs)
        .sort((left, right) => left.deadline - right.deadline || left.order - right.order)[0];
      if (!next) break;
      next.cancelled = true;
      this.nowMs = next.deadline;
      next.callback();
      await flush();
    }
    this.nowMs = timeMs;
  }

  async jumpTo(timeMs: number) {
    this.nowMs = timeMs;
    const due = this.timers
      .filter((timer) => !timer.cancelled && timer.deadline <= timeMs)
      .sort((left, right) => left.deadline - right.deadline || left.order - right.order);
    for (const timer of due) {
      if (timer.cancelled) continue;
      timer.cancelled = true;
      timer.callback();
      await flush();
    }
  }

  activeTimers() { return this.timers.filter((timer) => !timer.cancelled).length; }
}

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

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

function createGateway(runtime = new FakeRuntime(), clock = new FakeClock()) {
  return { runtime, clock, gateway: new Gateway(policy, runtime, { now: clock.now, schedule: clock.schedule }) };
}

function messagesFor(sent: OutgoingMessage[], replyTo: string) {
  return sent.filter((outgoing) => outgoing.replyTo === replyTo);
}

test("short asynchronous work sends only the real result with sequence one", async () => {
  const { runtime, gateway } = createGateway();
  const sent: OutgoingMessage[] = [];
  const turn = gateway.receive(message("work", "short"), recordingAdapter(sent));
  await waitFor(() => runtime.turns.length === 1);
  runtime.turns[0].resolve("done");
  assert.deepEqual(await turn, { ok: true, reply: "done" });
  assert.deepEqual(sent.map((outgoing) => [outgoing.text, outgoing.replySequence]), [["done", 1]]);
});

test("completion and timer callbacks are safe in both orders at 15 seconds", async () => {
  {
    const { runtime, clock, gateway } = createGateway();
    const sent: OutgoingMessage[] = [];
    const turn = gateway.receive(message("work", "completion-first"), recordingAdapter(sent));
    await waitFor(() => runtime.turns.length === 1);
    clock.set(15_000);
    runtime.turns[0].resolve("done");
    await turn;
    await clock.advanceTo(15_000);
    assert.deepEqual(sent.map((outgoing) => outgoing.text), ["done"]);
  }
  {
    const { runtime, clock, gateway } = createGateway();
    const sent: OutgoingMessage[] = [];
    const turn = gateway.receive(message("work", "timer-first"), recordingAdapter(sent));
    await waitFor(() => runtime.turns.length === 1);
    await clock.advanceTo(15_000);
    runtime.turns[0].resolve("done");
    await turn;
    assert.deepEqual(sent.map((outgoing) => [outgoing.text, outgoing.replySequence]), [
      ["这条还在处理，完成后我会直接回复。", 1],
      ["done", 2]
    ]);
  }
});

test("running and queued notices are independent and queued work announces start", async () => {
  const { runtime, clock, gateway } = createGateway();
  const sent: OutgoingMessage[] = [];
  const adapter = recordingAdapter(sent);
  const first = gateway.receive(message("one", "first"), adapter);
  const second = gateway.receive(message("two", "second"), adapter);
  await waitFor(() => runtime.turns.length === 1);
  await clock.advanceTo(15_000);
  assert.deepEqual(messagesFor(sent, "first").map((outgoing) => outgoing.text), ["这条还在处理，完成后我会直接回复。"]);
  assert.deepEqual(messagesFor(sent, "second").map((outgoing) => outgoing.text), ["前面还有任务，这条还在排队，轮到后我马上处理。"]);

  runtime.turns[0].resolve("first done");
  await waitFor(() => runtime.turns.length === 2);
  assert.deepEqual(messagesFor(sent, "second").map((outgoing) => [outgoing.text, outgoing.replySequence]), [
    ["前面还有任务，这条还在排队，轮到后我马上处理。", 1],
    ["轮到这条了，我开始处理。", 2]
  ]);
  runtime.turns[1].resolve("second done");
  await Promise.all([first, second]);
  assert.deepEqual(messagesFor(sent, "first").map((outgoing) => outgoing.replySequence), [1, 2]);
  assert.deepEqual(messagesFor(sent, "second").map((outgoing) => outgoing.replySequence), [1, 2, 3]);
});

test("running reminders fire at 60, 180, and 480 seconds with dynamic activity text", async () => {
  const { runtime, clock, gateway } = createGateway();
  const sent: OutgoingMessage[] = [];
  const turn = gateway.receive(message("work", "reminders"), recordingAdapter(sent));
  await waitFor(() => runtime.turns.length === 1);
  await clock.advanceTo(15_000);
  runtime.idleSeconds = 10;
  await clock.advanceTo(60_000);
  runtime.idleSeconds = 60;
  await clock.advanceTo(180_000);
  runtime.idleSeconds = 180;
  await clock.advanceTo(480_000);
  assert.deepEqual(sent.map((outgoing) => outgoing.text), [
    "这条还在处理，完成后我会直接回复。",
    "这条还在处理中，刚刚还有进展；完成后我会直接回复。",
    "这条还在处理中，最近 60 秒没新进展；完成后我会直接回复。",
    "这条还在处理中，最近 3 分钟没新进展；完成后我会直接回复。"
  ]);
  runtime.turns[0].resolve("done");
  await turn;
});

test("queued reminders use queue wording and sample the active ACP status", async () => {
  const { runtime, clock, gateway } = createGateway();
  const sent: OutgoingMessage[] = [];
  const adapter = recordingAdapter(sent);
  const first = gateway.receive(message("one", "active"), adapter);
  const second = gateway.receive(message("two", "queued"), adapter);
  await waitFor(() => runtime.turns.length === 1);
  await clock.advanceTo(15_000);
  runtime.idleSeconds = 45;
  await clock.advanceTo(60_000);
  assert.equal(messagesFor(sent, "queued").at(-1)?.text,
    "前面的任务还在处理，最近 45 秒没新进展；这条仍在排队，轮到后我马上处理。");
  runtime.turns[0].resolve("one done");
  await waitFor(() => runtime.turns.length === 2);
  runtime.turns[1].resolve("two done");
  await Promise.all([first, second]);
});

test("a delayed timer jump emits one reminder instead of backfilling every deadline", async () => {
  const { runtime, clock, gateway } = createGateway();
  const sent: OutgoingMessage[] = [];
  const turn = gateway.receive(message("work", "jump"), recordingAdapter(sent));
  await waitFor(() => runtime.turns.length === 1);
  await clock.advanceTo(15_000);
  await clock.jumpTo(1_000_000);
  assert.equal(sent.length, 2);
  assert.equal(clock.activeTimers(), 1);
  runtime.turns[0].resolve("done");
  await turn;
});

test("completion and shutdown cancel all inbound timers", async () => {
  const { runtime, clock, gateway } = createGateway();
  const sent: OutgoingMessage[] = [];
  const first = gateway.receive(message("one", "complete"), recordingAdapter(sent));
  await waitFor(() => runtime.turns.length === 1);
  assert.equal(clock.activeTimers(), 1);
  runtime.turns[0].resolve("done");
  await first;
  assert.equal(clock.activeTimers(), 0);

  const second = gateway.receive(message("two", "shutdown"), recordingAdapter(sent));
  await waitFor(() => runtime.turns.length === 2);
  assert.equal(clock.activeTimers(), 1);
  gateway.shutdown();
  assert.equal(clock.activeTimers(), 0);
  await clock.advanceTo(600_000);
  assert.deepEqual(messagesFor(sent, "shutdown"), []);
  runtime.turns[1].resolve("done after shutdown");
  await second;
});

test("prompt does not wait for a notice send and final delivery delay releases the chat lock", async () => {
  const { runtime, clock, gateway } = createGateway();
  const sent: OutgoingMessage[] = [];
  let releaseFirstSend!: () => void;
  const firstSend = new Promise<void>((resolve) => { releaseFirstSend = resolve; });
  const adapter: PlatformAdapter = {
    name: "test", async handleWebhook() { return {}; },
    async sendMessage(outgoing) {
      sent.push(outgoing);
      if (outgoing.replyTo === "first" && outgoing.replySequence === 1) await firstSend;
    }
  };
  const first = gateway.receive(message("one", "first"), adapter);
  await waitFor(() => runtime.turns.length === 1);
  await clock.advanceTo(15_000);
  assert.equal(runtime.turns.length, 1);
  runtime.turns[0].resolve("first done");
  const second = gateway.receive(message("two", "second"), adapter);
  await waitFor(() => runtime.turns.length === 2);
  assert.equal(messagesFor(sent, "first").length, 1);
  releaseFirstSend();
  runtime.turns[1].resolve("second done");
  await Promise.all([first, second]);
  assert.deepEqual(messagesFor(sent, "first").map((outgoing) => outgoing.text), [
    "这条还在处理，完成后我会直接回复。", "first done"
  ]);
});

test("delivery failures are logged safely and do not block later sends", async () => {
  const { runtime, clock, gateway } = createGateway();
  const sent: OutgoingMessage[] = [];
  const errors: string[] = [];
  let sends = 0;
  const adapter: PlatformAdapter = {
    name: "test", async handleWebhook() { return {}; },
    async sendMessage(outgoing) {
      sent.push(outgoing);
      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", "failure"), adapter);
    await waitFor(() => runtime.turns.length === 1);
    await clock.advanceTo(15_000);
    runtime.turns[0].resolve("done");
    assert.deepEqual(await turn, { ok: true, reply: "done" });
  } finally {
    console.error = originalError;
  }
  assert.deepEqual(sent.map((outgoing) => outgoing.text), ["这条还在处理，完成后我会直接回复。", "done"]);
  assert.deepEqual(errors, ["Gateway delivery send failed (platform=qq)"]);
});

test("a failed final send does not turn a successful runtime result into Agent error", async () => {
  const { runtime, gateway } = createGateway();
  const sent: OutgoingMessage[] = [];
  const errors: string[] = [];
  const adapter: PlatformAdapter = {
    name: "test", async handleWebhook() { return {}; },
    async sendMessage(outgoing) { sent.push(outgoing); throw new Error("provider detail"); }
  };
  const originalError = console.error;
  console.error = (...args: unknown[]) => { errors.push(args.map(String).join(" ")); };
  try {
    const turn = gateway.receive(message("work", "final-failure"), adapter);
    await waitFor(() => runtime.turns.length === 1);
    runtime.turns[0].resolve("real result");
    assert.deepEqual(await turn, { ok: true, reply: "real result" });
  } finally {
    console.error = originalError;
  }
  assert.deepEqual(sent.map((outgoing) => outgoing.text), ["real result"]);
  assert.deepEqual(errors, ["Gateway delivery send failed (platform=qq)"]);
});

test("runtime errors remain runtime errors and are delivered through the same stream", async () => {
  const { runtime, gateway } = createGateway();
  const sent: OutgoingMessage[] = [];
  const turn = gateway.receive(message("work", "runtime-error"), recordingAdapter(sent));
  await waitFor(() => runtime.turns.length === 1);
  runtime.turns[0].reject(new Error("failed"));
  assert.deepEqual(await turn, { ok: false, error: "failed", reply: "Agent error: failed" });
  assert.deepEqual(sent.map((outgoing) => [outgoing.text, outgoing.replySequence]), [["Agent error: failed", 1]]);
});

test("status, cancel, and help bypass the chat lock while new remains serial", async () => {
  const { runtime, clock, gateway } = createGateway();
  const adapter = recordingAdapter();
  const turn = gateway.receive(message("work"), adapter, { synchronous: true });
  await waitFor(() => runtime.turns.length === 1);
  clock.set(5_000);

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

  const second = gateway.receive(message("work"), adapter, { synchronous: true });
  await waitFor(() => runtime.turns.length === 2);
  const reset = gateway.receive(message("/new"), adapter, { synchronous: true });
  await flush();
  assert.equal(runtime.resets, 0);
  runtime.turns[1].resolve("done");
  await second;
  await reset;
  assert.equal(runtime.resets, 1);
});

test("synchronous work sends no notices or delivery messages", async () => {
  const { runtime, clock, gateway } = createGateway();
  const sent: OutgoingMessage[] = [];
  const turn = gateway.receive(message("work"), recordingAdapter(sent), { synchronous: true });
  await waitFor(() => runtime.turns.length === 1);
  await clock.advanceTo(600_000);
  assert.deepEqual(sent, []);
  runtime.turns[0].resolve("done");
  assert.deepEqual(await turn, { ok: true, reply: "done" });
  assert.deepEqual(sent, []);
});

test("asynchronous help replies with sequence one", async () => {
  const { runtime, gateway } = createGateway();
  const sent: OutgoingMessage[] = [];
  await gateway.receive(message("/help", "help"), recordingAdapter(sent));
  assert.deepEqual(messagesFor(sent, "help").map((outgoing) => outgoing.replySequence), [1]);
  assert.equal(runtime.turns.length, 0);
});

test("asynchronous new waits for prior work and then replies with sequence one", async () => {
  const { runtime, gateway } = createGateway();
  const sent: OutgoingMessage[] = [];
  const adapter = recordingAdapter(sent);
  const work = gateway.receive(message("work", "work"), adapter);
  await waitFor(() => runtime.turns.length === 1);

  const reset = gateway.receive(message("/new", "new"), adapter);
  await flush();
  assert.equal(runtime.resets, 0);
  assert.deepEqual(messagesFor(sent, "new"), []);

  runtime.turns[0].resolve("done");
  await work;
  assert.deepEqual(await reset, { ok: true, reply: "Started a new native ACP session for this chat." });
  assert.equal(runtime.resets, 1);
  assert.deepEqual(messagesFor(sent, "new").map((outgoing) => [outgoing.text, outgoing.replySequence]), [
    ["Started a new native ACP session for this chat.", 1]
  ]);
});

test("asynchronous retired-role command replies with sequence one without reaching ACP", async () => {
  const { runtime, gateway } = createGateway();
  const sent: OutgoingMessage[] = [];
  const result = await gateway.receive(message("/roles", "retired"), recordingAdapter(sent));
  assert.match(result.reply || "", /removed in Config v3/);
  assert.equal(runtime.turns.length, 0);
  assert.deepEqual(messagesFor(sent, "retired").map((outgoing) => outgoing.replySequence), [1]);
});