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: [] },
        toolMode: 'chat',
        taskType: null,
      },
    },
  ]);
});

test('POST /agent/runs forwards deep reasoning flag to the run gateway', async () => {
  const created = [];
  const handler = createPostAgentRunsHandler({
    userAuth: {
      async ownsSession() {
        return true;
      },
    },
    agentRunGateway: {
      async createRun(userId, payload) {
        created.push({ userId, payload });
        return { id: 'run-deep', status: 'queued' };
      },
    },
  });
  const res = createResponseRecorder();

  await handler(
    {
      currentUser: { id: 'user-1' },
      body: {
        request_id: 'req-deep',
        user_message: { role: 'user', content: [{ type: 'text', text: 'hello' }] },
        force_deep_reasoning: true,
      },
    },
    res,
  );

  assert.equal(res.statusCode, 202);
  assert.equal(created[0].payload.forceDeepReasoning, true);
});

test('POST /agent/runs accepts explicit code tool mode and task type', async () => {
  const created = [];
  const handler = createPostAgentRunsHandler({
    userAuth: {
      async ownsSession() {
        return true;
      },
    },
    agentRunGateway: {
      async createRun(userId, payload) {
        created.push({ userId, payload });
        return { id: 'run-code', status: 'queued' };
      },
    },
    codeRunsEnabled: true,
  });
  const res = createResponseRecorder();

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

  assert.equal(res.statusCode, 202);
  assert.deepEqual(created[0].payload, {
    sessionId: 'session-1',
    requestId: 'req-code',
    userMessage: { role: 'user', content: [] },
    toolMode: 'code',
    taskType: 'repo_refactor',
  });
});

test('POST /agent/runs accepts code tool mode for whitelisted user', async () => {
  const created = [];
  const handler = createPostAgentRunsHandler({
    userAuth: {
      async ownsSession() {
        return true;
      },
    },
    agentRunGateway: {
      async createRun(userId, payload) {
        created.push({ userId, payload });
        return { id: 'run-code', status: 'queued' };
      },
    },
    codeRunsEnabled: true,
    codeRunUserIds: new Set(['user-1']),
  });
  const res = createResponseRecorder();

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

  assert.equal(res.statusCode, 202);
  assert.equal(created[0].userId, 'user-1');
  assert.equal(created[0].payload.toolMode, 'code');
});

test('POST /agent/runs accepts allowlisted code task type with validation policy', async () => {
  const created = [];
  const handler = createPostAgentRunsHandler({
    userAuth: {
      async ownsSession() {
        return true;
      },
    },
    agentRunGateway: {
      async createRun(userId, payload) {
        created.push({ userId, payload });
        return { id: 'run-code', status: 'queued' };
      },
    },
    codeRunsEnabled: true,
    codeRunUserIds: new Set(['user-1']),
    codeRunTaskTypes: new Set(['small_patch']),
    requireCodeRunValidation: true,
  });
  const userMessage = {
    role: 'user',
    content: [{ type: 'text', text: 'create artifact' }],
    metadata: {
      memindRun: {
        validation: {
          expectedFile: {
            path: 'RESULT.md',
            contains: 'ok',
          },
        },
      },
    },
  };
  const res = createResponseRecorder();

  await handler(
    {
      currentUser: { id: 'user-1' },
      body: {
        request_id: 'req-code-small-patch',
        user_message: userMessage,
        tool_mode: 'code',
        task_type: 'small_patch',
      },
    },
    res,
  );

  assert.equal(res.statusCode, 202);
  assert.equal(created[0].payload.taskType, 'small_patch');
  assert.deepEqual(created[0].payload.userMessage, userMessage);
});

test('POST /agent/runs rejects non-allowlisted code task type', async () => {
  const handler = createPostAgentRunsHandler({
    userAuth: {
      async ownsSession() {
        throw new Error('should not check ownership after task type gate rejects code mode');
      },
    },
    agentRunGateway: {
      async createRun() {
        throw new Error('should not be called');
      },
    },
    codeRunsEnabled: true,
    codeRunUserIds: new Set(['user-1']),
    codeRunTaskTypes: new Set(['small_patch']),
  });
  const res = createResponseRecorder();

  await handler(
    {
      currentUser: { id: 'user-1' },
      body: {
        request_id: 'req-code-repo-refactor',
        user_message: { role: 'user', content: [] },
        tool_mode: 'code',
        task_type: 'repo_refactor',
      },
    },
    res,
  );

  assert.equal(res.statusCode, 403);
  assert.deepEqual(res.body, { message: '当前代码任务类型未开启灰度' });
});

test('POST /agent/runs requires validation metadata for guarded code tasks', async () => {
  const handler = createPostAgentRunsHandler({
    userAuth: {
      async ownsSession() {
        throw new Error('should not check ownership after validation gate rejects code mode');
      },
    },
    agentRunGateway: {
      async createRun() {
        throw new Error('should not be called');
      },
    },
    codeRunsEnabled: true,
    codeRunUserIds: new Set(['user-1']),
    requireCodeRunValidation: true,
  });
  const res = createResponseRecorder();

  await handler(
    {
      currentUser: { id: 'user-1' },
      body: {
        request_id: 'req-code-no-validation',
        user_message: { role: 'user', content: [] },
        tool_mode: 'code',
        task_type: 'small_patch',
      },
    },
    res,
  );

  assert.equal(res.statusCode, 400);
  assert.deepEqual(res.body, { message: '代码任务必须声明产物校验规则' });
});

test('POST /agent/runs rejects code tool mode for non-whitelisted user', async () => {
  const handler = createPostAgentRunsHandler({
    userAuth: {
      async ownsSession() {
        throw new Error('should not check ownership after user gate rejects code mode');
      },
    },
    agentRunGateway: {
      async createRun() {
        throw new Error('should not be called');
      },
    },
    codeRunsEnabled: true,
    codeRunUserIds: new Set(['user-2']),
  });
  const res = createResponseRecorder();

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

  assert.equal(res.statusCode, 403);
  assert.deepEqual(res.body, { message: '当前用户未开启代码任务灰度' });
});

test('POST /agent/runs rejects code tool mode when server gate is disabled', async () => {
  const handler = createPostAgentRunsHandler({
    userAuth: {
      async ownsSession() {
        throw new Error('should not check ownership after disabled code mode');
      },
    },
    agentRunGateway: {
      async createRun() {
        throw new Error('should not be called');
      },
    },
    codeRunsEnabled: false,
  });
  const res = createResponseRecorder();

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

  assert.equal(res.statusCode, 403);
  assert.deepEqual(res.body, { message: '代码任务灰度未开启' });
});

test('POST /agent/runs rejects unsupported tool mode', async () => {
  const handler = createPostAgentRunsHandler({
    userAuth: {
      async ownsSession() {
        throw new Error('should not check ownership after invalid mode');
      },
    },
    agentRunGateway: {
      async createRun() {
        throw new Error('should not be called');
      },
    },
  });
  const res = createResponseRecorder();

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

  assert.equal(res.statusCode, 400);
  assert.match(res.body.message, /tool_mode/);
});

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('POST /agent/runs returns 409 when session already has active run', async () => {
  const handler = createPostAgentRunsHandler({
    userAuth: {
      async ownsSession() {
        return true;
      },
    },
    agentRunGateway: {
      async createRun() {
        const err = new Error('该会话有正在处理的任务，请等待完成后再发送');
        err.code = 'SESSION_RUN_CONFLICT';
        err.status = 409;
        throw err;
      },
    },
  });
  const res = createResponseRecorder();

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

  assert.equal(res.statusCode, 409);
  assert.deepEqual(res.body, {
    message: '该会话有正在处理的任务，请等待完成后再发送',
    code: 'SESSION_RUN_CONFLICT',
  });
});

test('POST /agent/runs validates ownership through sessionAccess when provided', async () => {
  const validated = [];
  const handler = createPostAgentRunsHandler({
    userAuth: {},
    sessionAccess: {
      async validateOwnership(userId, sessionId) {
        validated.push({ userId, sessionId });
        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-broker',
        request_id: 'req-broker',
        user_message: { role: 'user', content: [] },
      },
    },
    res,
  );

  assert.deepEqual(validated, [{ userId: 'user-1', sessionId: 'session-broker' }]);
  assert.equal(res.statusCode, 403);
});

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 attaches run taxonomy when flag enabled', async () => {
  const previous = process.env.MEMIND_SSE_EVENT_TAXONOMY;
  process.env.MEMIND_SSE_EVENT_TAXONOMY = '1';
  try {
    const handler = createAgentRunEventsHandler({
      agentRunGateway: {
        async getRunForUser() {
          return { id: 'run-tax', status: 'succeeded', sessionId: 'session-1' };
        },
        dispatchRun() {},
      },
      pollIntervalMs: 5,
      keepaliveIntervalMs: 50,
    });
    const res = createSseResponseRecorder();
    await handler(
      {
        currentUser: { id: 'user-1' },
        params: { runId: 'run-tax' },
        on() {},
      },
      res,
    );
    assert.match(res.chunks.join(''), /"taxonomy":"terminal"/);
  } finally {
    if (previous == null) delete process.env.MEMIND_SSE_EVENT_TAXONOMY;
    else process.env.MEMIND_SSE_EVENT_TAXONOMY = previous;
  }
});

test('GET /agent/runs/:runId/events replay mode emits SSE ids and replays after Last-Event-ID', async () => {
  const events = [
    {
      id: 'evt-1',
      eventType: 'run_snapshot',
      data: { run: { id: 'run-replay', status: 'running', sessionId: null } },
      createdAt: 1,
    },
    {
      id: 'evt-2',
      eventType: 'run_snapshot',
      data: { run: { id: 'run-replay', status: 'succeeded', sessionId: 'session-9' } },
      createdAt: 2,
    },
  ];
  const handler = createAgentRunEventsHandler({
    replayEnabled: true,
    agentRunGateway: {
      async getRunForUser() {
        return { id: 'run-replay', status: 'succeeded', sessionId: 'session-9' };
      },
      async listRunEventsForUser(_userId, _runId, { afterEventId } = {}) {
        if (afterEventId === 'evt-1') {
          return {
            run: { id: 'run-replay', status: 'succeeded', sessionId: 'session-9' },
            events: [events[1]],
            cursorMiss: false,
          };
        }
        return {
          run: { id: 'run-replay', status: 'running', sessionId: null },
          events,
          cursorMiss: false,
        };
      },
      dispatchRun() {},
    },
    pollIntervalMs: 1000,
    keepaliveIntervalMs: 5000,
  });
  const res = createSseResponseRecorder();
  await handler(
    {
      currentUser: { id: 'user-1' },
      params: { runId: 'run-replay' },
      get(name) {
        return name.toLowerCase() === 'last-event-id' ? 'evt-1' : undefined;
      },
      on() {},
    },
    res,
  );
  assert.equal(res.ended, true);
  const output = res.chunks.join('');
  assert.match(output, /id: evt-2/);
  assert.match(output, /"sessionId":"session-9"/);
});

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']);
});

test('POST /agent/runs materializes selected assets before creating run', async () => {
  const materialized = [];
  const handler = createPostAgentRunsHandler({
    userAuth: {
      async ownsSession() {
        return true;
      },
    },
    mindSpaceAssetAgent: {
      async materializeAssetsForSession(payload) {
        materialized.push(payload);
        return { materialized: [{ relativePath: 'oa/sales.xlsx' }] };
      },
    },
    agentRunGateway: {
      async createRun(userId, payload) {
        return { id: 'run-1', status: 'queued', userId, payload };
      },
    },
  });
  const res = createResponseRecorder();

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

  assert.equal(res.statusCode, 202);
  assert.deepEqual(materialized, [{
    sessionId: 'session-1',
    userId: 'user-1',
    assetIds: ['asset-9'],
  }]);
});