95 lines
5.6 KiB
TypeScript
95 lines
5.6 KiB
TypeScript
import { expect, it, mock } from "bun:test";
|
|
import { ResponseInputTracker, WaitingResponseInputQueue, type ResponseInput } from "../../src/chat/response-inputs.js";
|
|
|
|
const input = (id: number): ResponseInput => ({ id: `msg_${String(id).padStart(32, "0")}`, text: `Input ${id}`, delivery: "steer", status: "pending" });
|
|
|
|
it("serializes waiting input admission, deduplicates retries and enforces capacity without resuming", async () => {
|
|
const pending: ResponseInput[] = [];
|
|
const enqueue = mock(async (_session: string, item: ResponseInput, options?: { resume?: boolean }) => {
|
|
expect(options?.resume).toBe(false);
|
|
pending.push(item);
|
|
});
|
|
const queue = new WaitingResponseInputQueue({ listPendingUserInputs: async () => pending,
|
|
agentMessageState: async () => "missing", enqueueUserInput: enqueue });
|
|
await Promise.all([queue.submit("s", input(1)), queue.submit("s", input(1))]);
|
|
expect(enqueue).toHaveBeenCalledTimes(1);
|
|
await expect(queue.submit("s", { ...input(1), text: "different" })).rejects.toMatchObject({ status: 409 });
|
|
const results = await Promise.allSettled(Array.from({ length: 20 }, (_, i) => queue.submit("s", input(i + 2))));
|
|
expect(results.filter((result) => result.status === "fulfilled")).toHaveLength(19);
|
|
expect(pending).toHaveLength(20);
|
|
expect(enqueue).toHaveBeenCalledTimes(20);
|
|
});
|
|
|
|
it("does not reinject a consumed waiting input or infer absence after a native read failure", async () => {
|
|
const enqueue = mock(async () => {});
|
|
const runtime = { listPendingUserInputs: async () => [], agentMessageState: async () => "delivered" as const, enqueueUserInput: enqueue };
|
|
await expect(new WaitingResponseInputQueue(runtime).submit("s", input(1))).resolves.toMatchObject({ status: "delivered" });
|
|
expect(enqueue).not.toHaveBeenCalled();
|
|
const unavailable = { ...runtime, agentMessageState: async () => { throw new Error("native unavailable"); } };
|
|
await expect(new WaitingResponseInputQueue(unavailable).submit("s", input(2))).rejects.toThrow("native unavailable");
|
|
expect(enqueue).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it("waits for admission at completion, deduplicates concurrent retries and closes only after native consumption", async () => {
|
|
let release = () => {};
|
|
const gate = new Promise<void>((resolve) => { release = resolve; });
|
|
const runtime = { listPendingUserInputs: async () => [], enqueueUserInput: mock(async () => { await gate; }) };
|
|
const changed = mock((_input: ResponseInput) => {});
|
|
const tracker = new ResponseInputTracker(runtime, "owned", changed);
|
|
tracker.markReady();
|
|
const first = tracker.submit(input(1));
|
|
const retry = tracker.submit(input(1));
|
|
const closing = tracker.canClose();
|
|
release();
|
|
expect((await first).status).toBe("pending");
|
|
expect((await retry).status).toBe("pending");
|
|
expect(runtime.enqueueUserInput).toHaveBeenCalledTimes(1);
|
|
expect(await closing).toBe(false);
|
|
tracker.observe("session.inbox.delivery.changed", { inboxID: input(1).id, delivery: "queue" });
|
|
expect(changed.mock.calls.at(-1)?.[0]).toMatchObject({ delivery: "queue" });
|
|
tracker.observe("session.inbox.delivered", { inboxID: input(1).id });
|
|
expect(changed.mock.calls.at(-1)?.[0]).toMatchObject({ status: "delivered" });
|
|
expect(await tracker.canClose()).toBe(true);
|
|
await expect(tracker.submit(input(2))).rejects.toMatchObject({ status: 409 });
|
|
});
|
|
|
|
it("rejects a failed admission and permits a later retry without acknowledging nonexistent input", async () => {
|
|
let attempts = 0;
|
|
const runtime = { listPendingUserInputs: async () => [], enqueueUserInput: async () => { if (attempts++ === 0) throw new Error("fixture failure"); } };
|
|
const tracker = new ResponseInputTracker(runtime, "owned", () => {});
|
|
tracker.markReady();
|
|
const first = tracker.submit(input(1));
|
|
const duplicate = tracker.submit(input(1));
|
|
const results = await Promise.allSettled([first, duplicate]);
|
|
expect(results.map((result) => result.status)).toEqual(["rejected", "rejected"]);
|
|
for (const result of results) if (result.status === "rejected") expect(result.reason.message).toBe("fixture failure");
|
|
await expect(tracker.submit(input(1))).resolves.toMatchObject({ status: "pending" });
|
|
});
|
|
|
|
it("bounds pending input and refuses admission after an explicit stop", async () => {
|
|
const runtime = { listPendingUserInputs: async () => [], enqueueUserInput: async () => {} };
|
|
const tracker = new ResponseInputTracker(runtime, "owned", () => {});
|
|
tracker.markReady();
|
|
for (let id = 0; id < 20; id++) await tracker.submit(input(id));
|
|
await expect(tracker.submit(input(20))).rejects.toMatchObject({ status: 429 });
|
|
tracker.close();
|
|
await expect(tracker.submit(input(21))).rejects.toMatchObject({ status: 409 });
|
|
});
|
|
|
|
it("allows close after withdrawal and keeps cancelled retries from enqueueing again", async () => {
|
|
const runtime = { listPendingUserInputs: async () => [], enqueueUserInput: mock(async () => {}) };
|
|
const changed = mock((_input: ResponseInput) => {});
|
|
const tracker = new ResponseInputTracker(runtime, "owned", changed);
|
|
tracker.markReady();
|
|
await tracker.submit(input(1));
|
|
tracker.observe("session.inbox.cancelled", { inboxID: input(1).id });
|
|
expect(changed.mock.calls.at(-1)?.[0].status).toBe("cancelled");
|
|
expect(tracker.hasAdditionalInput).toBe(false);
|
|
expect((await tracker.submit(input(1))).status).toBe("cancelled");
|
|
expect(runtime.enqueueUserInput).toHaveBeenCalledTimes(1);
|
|
tracker.observe("session.inbox.delivery.changed", { inboxID: input(1).id, delivery: "queue" });
|
|
tracker.observe("session.inbox.delivered", { inboxID: input(1).id });
|
|
expect(changed).toHaveBeenCalledTimes(2);
|
|
expect(await tracker.canClose()).toBe(true);
|
|
});
|