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