memind/server/portal-agent-job-routes.mjs

import fs from 'node:fs';

const TERMINAL_AGENT_JOB_STATUSES = new Set([
  'completed',
  'failed',
  'cancelled',
  'timed_out',
]);

function assertRouter(api) {
  if (
    !api ||
    typeof api.get !== 'function' ||
    typeof api.post !== 'function'
  ) {
    throw new Error(
      'attachPortalAgentJobRoutes requires an Express-compatible router',
    );
  }
}

export function attachPortalAgentJobRoutes(
  api,
  {
    getMindSpaceAgentJobs = () => null,
    getMindSpaceAgentRunner = () => null,
    getMindSpaceAudit = () => null,
    ensureMindSpaceEnabled = () => false,
    requireInternalAgentSecret = () => false,
    bearerToken = () => null,
    sendData = (res, _req, data, status = 200) =>
      res.status(status).json({ data }),
    handleMindSpaceError = (_res, _req, error) => {
      throw error;
    },
    readFile = fs.promises.readFile,
    getSsePollMs = () => 1_000,
    setIntervalFn = setInterval,
    clearIntervalFn = clearInterval,
    logger = console,
  } = {},
) {
  assertRouter(api);

  api.post('/mindspace/v1/agent/jobs', async (req, res) => {
    const agentJobs = getMindSpaceAgentJobs();
    if (!agentJobs || !ensureMindSpaceEnabled(res, req, { agent: true })) return;
    try {
      const job = await agentJobs.createJob(req.currentUser.id, {
        jobType: req.body?.job_type,
        instruction: req.body?.instruction,
        allowedAssetIds: req.body?.allowed_asset_ids,
        outputCategoryId: req.body?.output_category_id,
        outputType: req.body?.output_type,
        idempotencyKey: req.body?.idempotency_key,
        locale: req.body?.locale,
        timezone: req.body?.timezone,
        capabilities: req.body?.capabilities,
      });
      await getMindSpaceAudit()?.write({
        userId: req.currentUser.id,
        action: 'agent_access',
        objectType: 'agent_job',
        objectId: job.id,
        ip: req.ip,
        detail: {
          jobType: job.jobType,
          assetIds: job.assets.map((asset) => asset.assetId),
        },
      });
      return sendData(res, req, job, 201);
    } catch (error) {
      return handleMindSpaceError(res, req, error);
    }
  });

  api.get('/mindspace/v1/agent/jobs/:jobId', async (req, res) => {
    const agentJobs = getMindSpaceAgentJobs();
    if (!agentJobs || !ensureMindSpaceEnabled(res, req, { agent: true })) return;
    try {
      return sendData(
        res,
        req,
        await agentJobs.getJob(req.currentUser.id, req.params.jobId),
      );
    } catch (error) {
      return handleMindSpaceError(res, req, error);
    }
  });

  // Long-running jobs are dispatched asynchronously and observed through this
  // SSE stream. Ownership remains enforced by the injected getJob service.
  api.get('/mindspace/v1/agent/jobs/:jobId/stream', async (req, res) => {
    const agentJobs = getMindSpaceAgentJobs();
    if (!agentJobs || !ensureMindSpaceEnabled(res, req, { agent: true })) return;
    let job;
    try {
      job = await agentJobs.getJob(req.currentUser.id, req.params.jobId);
    } catch (error) {
      return handleMindSpaceError(res, req, error);
    }
    res.writeHead(200, {
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache, no-transform',
      Connection: 'keep-alive',
      'X-Accel-Buffering': 'no',
    });
    const send = (event, payload) => {
      res.write(`event: ${event}\n`);
      res.write(`data: ${JSON.stringify(payload)}\n\n`);
    };
    let lastSignature = '';
    const emitIfChanged = (current) => {
      const signature = `${current.status}:${JSON.stringify(current.progress ?? {})}`;
      if (signature !== lastSignature) {
        lastSignature = signature;
        send('progress', current);
      }
      return signature;
    };
    emitIfChanged(job);
    if (TERMINAL_AGENT_JOB_STATUSES.has(job.status)) {
      send('done', job);
      return res.end();
    }
    let closed = false;
    const cleanup = () => {
      if (closed) return;
      closed = true;
      clearIntervalFn(pollTimer);
      clearIntervalFn(keepAliveTimer);
    };
    const pollTimer = setIntervalFn(async () => {
      if (closed) return;
      try {
        const current = await agentJobs.getJob(
          req.currentUser.id,
          req.params.jobId,
        );
        emitIfChanged(current);
        if (TERMINAL_AGENT_JOB_STATUSES.has(current.status)) {
          send('done', current);
          cleanup();
          res.end();
        }
      } catch {
        cleanup();
        res.end();
      }
    }, getSsePollMs());
    const keepAliveTimer = setIntervalFn(() => {
      if (!closed) res.write(': keep-alive\n\n');
    }, 15_000);
    pollTimer.unref?.();
    keepAliveTimer.unref?.();
    req.on('close', cleanup);
  });

  api.get('/mindspace/v1/agent/jobs', async (req, res) => {
    const agentJobs = getMindSpaceAgentJobs();
    if (!agentJobs || !ensureMindSpaceEnabled(res, req, { agent: true })) return;
    try {
      const result = await agentJobs.listJobs(req.currentUser.id, {
        limit: Number(req.query.limit ?? 10),
        offset: Number(req.query.offset ?? 0),
      });
      return res.json({
        data: result.items,
        page: {
          total: result.total,
          offset: result.offset,
          limit: result.limit,
          has_more: result.hasMore,
        },
        request_id: req.requestId,
      });
    } catch (error) {
      return handleMindSpaceError(res, req, error);
    }
  });

  for (const [route, operation] of [
    ['/mindspace/v1/agent/jobs/:jobId/cancel', 'cancelJob'],
    ['/mindspace/v1/agent/jobs/:jobId/retry', 'retryJob'],
  ]) {
    api.post(route, async (req, res) => {
      const agentJobs = getMindSpaceAgentJobs();
      if (!agentJobs || !ensureMindSpaceEnabled(res, req, { agent: true })) return;
      try {
        return sendData(
          res,
          req,
          await agentJobs[operation](req.currentUser.id, req.params.jobId),
        );
      } catch (error) {
        return handleMindSpaceError(res, req, error);
      }
    });
  }

  api.post('/mindspace/v1/agent/jobs/:jobId/run', async (req, res) => {
    const agentJobs = getMindSpaceAgentJobs();
    const agentRunner = getMindSpaceAgentRunner();
    if (
      !agentJobs ||
      !agentRunner ||
      !ensureMindSpaceEnabled(res, req, { agent: true })
    ) {
      return;
    }
    try {
      const job = await agentJobs.getJob(req.currentUser.id, req.params.jobId);
      if (job.status !== 'queued') {
        return sendData(res, req, job);
      }
      void agentRunner.runJob(req.params.jobId).catch((error) => {
        logger.error('MindSpace agent job run failed:', error);
      });
      return sendData(
        res,
        req,
        { started: true, jobId: req.params.jobId },
        202,
      );
    } catch (error) {
      return handleMindSpaceError(res, req, error);
    }
  });

  api.post('/internal/agent/jobs/:jobId/claim', async (req, res) => {
    const agentJobs = getMindSpaceAgentJobs();
    if (!agentJobs || !ensureMindSpaceEnabled(res, req, { agent: true })) return;
    if (!requireInternalAgentSecret(req, res)) return;
    try {
      return sendData(res, req, await agentJobs.claimJob(req.params.jobId));
    } catch (error) {
      return handleMindSpaceError(res, req, error);
    }
  });

  api.get('/internal/agent/jobs/:jobId/assets/:assetId', async (req, res) => {
    const agentJobs = getMindSpaceAgentJobs();
    if (!agentJobs || !ensureMindSpaceEnabled(res, req, { agent: true })) return;
    try {
      const asset = await agentJobs.getAssetForJob(
        req.params.jobId,
        bearerToken(req),
        req.params.assetId,
      );
      res.set('Content-Type', asset.mimeType);
      res.set(
        'Content-Disposition',
        `inline; filename="${encodeURIComponent(asset.displayName)}"`,
      );
      res.set('Cache-Control', 'private, no-store');
      res.setHeader('X-Request-Id', req.requestId);
      return res.send(await readFile(asset.path));
    } catch (error) {
      return handleMindSpaceError(res, req, error);
    }
  });

  api.post('/internal/agent/jobs/:jobId/heartbeat', async (req, res) => {
    const agentJobs = getMindSpaceAgentJobs();
    if (!agentJobs || !ensureMindSpaceEnabled(res, req, { agent: true })) return;
    try {
      return sendData(
        res,
        req,
        await agentJobs.heartbeat(req.params.jobId, bearerToken(req), {
          stage: req.body?.stage,
          message: req.body?.message,
        }),
      );
    } catch (error) {
      return handleMindSpaceError(res, req, error);
    }
  });

  api.post('/internal/agent/jobs/:jobId/complete', async (req, res) => {
    const agentJobs = getMindSpaceAgentJobs();
    if (!agentJobs || !ensureMindSpaceEnabled(res, req, { agent: true })) return;
    try {
      const job = await agentJobs.completeJob(
        req.params.jobId,
        bearerToken(req),
        {
          status: req.body?.status,
          errorCode: req.body?.error_code,
          errorMessage: req.body?.error_message,
          outputType: req.body?.output_type,
          title: req.body?.title,
          summary: req.body?.summary,
          content: req.body?.content,
          contentFormat: req.body?.content_format,
          pageType: req.body?.page_type,
          templateId: req.body?.template_id,
          sourceAssetIds: req.body?.source_asset_ids,
        },
      );
      return sendData(res, req, job);
    } catch (error) {
      return handleMindSpaceError(res, req, error);
    }
  });
}