memind/agent-run-routes.test.mjs

import assert from 'node:assert/strict';
import test from 'node:test';
import {
  createAgentRunEventsHandler,
  createGetAgentRunHandler,
  createPostAgentRunsHandler,
} from './agent-run-routes.mjs';

function createResponseRecorder() {
  return {
    statusCode: 200,
    body: null,
    status(code) {
      this.statusCode = code;
      return this;
    },
    json(payload) {
      this.body = payload;
      return this;
    },
  };
}

function createSseResponseRecorder() {
  return {
    statusCode: 200,
    headers: {},
    chunks: [],
    ended: false,
    status(code) {
      this.statusCode = code;
      return this;
    },
    json(payload) {
      this.body = payload;
      return this;
    },
    setHeader(key, value) {
      this.headers[key] = value;
    },
    write(chunk) {
      this.chunks.push(chunk);
    },
    end() {
      this.ended = true;
    },
  };
}

async function waitFor(predicate) {
  for (let i = 0; i < 50; i += 1) {
    if (predicate()) return;
    await new Promise((resolve) => setTimeout(resolve, 5));
  }
  assert.fail('condition was not met');
}

test('POST /agent/runs creates a run and returns 202', async () => {
  const created = [];
  const handler = createPostAgentRunsHandler({
    userAuth: {
      async ownsSession() {
        return true;
      },
    },
    agentRunGateway: {
      async createRun(userId, payload) {
        created.push({ userId, payload });
        return { id: 'run-1', status: 'queued' };
      },
    },
  });
  const res = createResponseRecorder();

  await handler(
    {
      currentUser: { id: 'user-1' },
      body: {
        session_id: 'session-1',
        request_id: 'req-1',
        user_message: { role: 'user', content: [] },
      },
    },
    res,
  );

  assert.equal(res.statusCode, 202);
  assert.deepEqual(res.body, { run: { id: 'run-1', status: 'queued' } });
  assert.deepEqual(created, [
    {
      userId: 'user-1',
      payload: {
        sessionId: 'session-1',
        requestId: 'req-1',
        userMessage: { role: 'user', content: [] },
      },
    },
  ]);
});

test('POST /agent/runs rejects a session the user does not own', async () => {
  const handler = createPostAgentRunsHandler({
    userAuth: {
      async ownsSession() {
        return false;
      },
    },
    agentRunGateway: {
      async createRun() {
        throw new Error('should not be called');
      },
    },
  });
  const res = createResponseRecorder();

  await handler(
    {
      currentUser: { id: 'user-1' },
      body: {
        session_id: 'session-2',
        request_id: 'req-1',
        user_message: { role: 'user', content: [] },
      },
    },
    res,
  );

  assert.equal(res.statusCode, 403);
  assert.deepEqual(res.body, { message: '无权访问该会话' });
});

test('GET /agent/runs/:runId dispatches unfinished runs', async () => {
  const dispatched = [];
  const handler = createGetAgentRunHandler({
    agentRunGateway: {
      async getRunForUser(userId, runId) {
        assert.equal(userId, 'user-1');
        assert.equal(runId, 'run-1');
        return { id: 'run-1', status: 'running' };
      },
      dispatchRun(runId) {
        dispatched.push(runId);
      },
    },
  });
  const res = createResponseRecorder();

  await handler(
    {
      currentUser: { id: 'user-1' },
      params: { runId: 'run-1' },
    },
    res,
  );

  assert.equal(res.statusCode, 200);
  assert.deepEqual(res.body, { run: { id: 'run-1', status: 'running' } });
  assert.deepEqual(dispatched, ['run-1']);
});

test('GET /agent/runs/:runId returns 404 for missing runs', async () => {
  const handler = createGetAgentRunHandler({
    agentRunGateway: {
      async getRunForUser() {
        return null;
      },
      dispatchRun() {},
    },
  });
  const res = createResponseRecorder();

  await handler(
    {
      currentUser: { id: 'user-1' },
      params: { runId: 'missing' },
    },
    res,
  );

  assert.equal(res.statusCode, 404);
  assert.deepEqual(res.body, { message: '任务不存在' });
});

test('GET /agent/runs/:runId/events streams the current run and closes on terminal state', async () => {
  const handler = createAgentRunEventsHandler({
    agentRunGateway: {
      async getRunForUser() {
        return { id: 'run-1', status: 'succeeded', sessionId: 'session-1' };
      },
      dispatchRun() {
        throw new Error('should not dispatch terminal run');
      },
    },
    pollIntervalMs: 5,
    keepaliveIntervalMs: 50,
  });
  const res = createSseResponseRecorder();
  const listeners = new Map();

  await handler(
    {
      currentUser: { id: 'user-1' },
      params: { runId: 'run-1' },
      on(event, cb) {
        listeners.set(event, cb);
      },
    },
    res,
  );

  assert.equal(res.statusCode, 200);
  assert.equal(res.headers['Content-Type'], 'text/event-stream');
  assert.match(res.chunks.join(''), /event: run/);
  assert.match(res.chunks.join(''), /"status":"succeeded"/);
  assert.equal(res.ended, true);
});

test('GET /agent/runs/:runId/events republishes changed run state before terminal close', async () => {
  let readCount = 0;
  const dispatched = [];
  const handler = createAgentRunEventsHandler({
    agentRunGateway: {
      async getRunForUser() {
        readCount += 1;
        if (readCount < 2) {
          return { id: 'run-2', status: 'running', sessionId: null };
        }
        return { id: 'run-2', status: 'succeeded', sessionId: 'session-2' };
      },
      dispatchRun(runId) {
        dispatched.push(runId);
      },
    },
    pollIntervalMs: 5,
    keepaliveIntervalMs: 50,
  });
  const res = createSseResponseRecorder();

  await handler(
    {
      currentUser: { id: 'user-1' },
      params: { runId: 'run-2' },
      on() {},
    },
    res,
  );

  await waitFor(() => res.ended);

  const output = res.chunks.join('');
  assert.match(output, /"status":"running"/);
  assert.match(output, /"status":"succeeded"/);
  assert.deepEqual(dispatched, ['run-2']);
});