f89e074dd1
Governed substrate for autonomous agents: scoped identity (passports), audited actions, MCP workspace. Infra IPs and secrets redacted for public release.
225 lines
11 KiB
JavaScript
225 lines
11 KiB
JavaScript
/**
|
|
* SSE stream, reset-processing, and stop agent routes.
|
|
*/
|
|
|
|
import {
|
|
dbRun, dbGet, dbAll, isPostgres, safeJsonParse, apiLogger,
|
|
success, error, badRequest,
|
|
requireAuth, getJobsForConversation, cancelJob,
|
|
} from './chatShared.js';
|
|
import { getContextUsage } from '../../../services/agent-job/liveContextUsage.js';
|
|
// T-148528 (WP-B): /stop must also reach dispatcher-managed runs (ADR-0030
|
|
// path) — the legacy `cancelJob` only covers `agent_jobs` rows.
|
|
import { abortRunByConversation as abortDispatcherRunByConversation } from '../../../services/agent-run-dispatcher/index.js';
|
|
// ADR-0057-A WP-B: live push of `_inflight_runs` deltas (pause/resume/done)
|
|
// to subscribed chat streams. Falls back silently to the per-poll snapshot
|
|
// from messageController.js when the bus is unavailable.
|
|
import { subscribeInflight } from '../../../services/inflight/notifyBus.js';
|
|
|
|
export default function registerStreamRoutes(router) {
|
|
|
|
// GET /conversations/:id/stream - Server-Sent Events stream
|
|
router.get('/conversations/:id/stream', requireAuth, (req, res) => {
|
|
const { id } = req.params;
|
|
let lastMessageId = parseInt(req.query.after) || 0;
|
|
|
|
res.setHeader('Content-Type', 'text/event-stream');
|
|
res.setHeader('Cache-Control', 'no-cache');
|
|
res.setHeader('Connection', 'keep-alive');
|
|
res.setHeader('X-Accel-Buffering', 'no');
|
|
|
|
res.write(`event: connected\ndata: {"conversationId": ${id}}\n\n`);
|
|
|
|
const heartbeat = setInterval(() => {
|
|
try { res.write(': heartbeat\n\n'); } catch (_) {}
|
|
}, 15000);
|
|
|
|
// ADR-0057-A WP-B — subscribe to chat_inflight pg_notify bus and fan
|
|
// payloads scoped to this conversation out to the client. Idempotent
|
|
// unsubscribe on close prevents EventEmitter listener leaks.
|
|
const convIdNum = Number(id);
|
|
const unsubscribeInflight = subscribeInflight((payload) => {
|
|
if (!payload || Number(payload.conversation_id) !== convIdNum) return;
|
|
try {
|
|
res.write(`event: inflight\ndata: ${JSON.stringify(payload)}\n\n`);
|
|
} catch (err) {
|
|
apiLogger.debug({ err: err.message, conversationId: id }, 'SSE inflight write failed (client gone)');
|
|
}
|
|
});
|
|
|
|
let lastPollTime = new Date().toISOString();
|
|
|
|
const poller = setInterval(async () => {
|
|
try {
|
|
const pollStart = new Date().toISOString();
|
|
|
|
const newMessages = await dbAll(
|
|
isPostgres()
|
|
? `SELECT m.*, u.name as sender_name, u.avatar as sender_avatar FROM messages m LEFT JOIN users u ON m.sender_id = u.id WHERE m.conversation_id = $1 AND m.id > $2 ORDER BY m.id ASC LIMIT 50`
|
|
: `SELECT m.*, u.name as sender_name, u.avatar as sender_avatar FROM messages m LEFT JOIN users u ON m.sender_id = u.id WHERE m.conversation_id = ? AND m.id > ? ORDER BY m.id ASC LIMIT 50`,
|
|
[id, lastMessageId]
|
|
);
|
|
|
|
const newMessageIds = new Set(newMessages.map(m => m.id));
|
|
const updatedStatusMessages = await dbAll(
|
|
isPostgres()
|
|
? `SELECT m.*, u.name as sender_name, u.avatar as sender_avatar FROM messages m LEFT JOIN users u ON m.sender_id = u.id WHERE m.conversation_id = $1 AND m.content_type = 'agent_status' AND m.updated_at > $2 AND m.id <= $3 ORDER BY m.id ASC LIMIT 10`
|
|
: `SELECT m.*, u.name as sender_name, u.avatar as sender_avatar FROM messages m LEFT JOIN users u ON m.sender_id = u.id WHERE m.conversation_id = ? AND m.content_type = 'agent_status' AND m.updated_at > ? AND m.id <= ? ORDER BY m.id ASC LIMIT 10`,
|
|
[id, lastPollTime, lastMessageId]
|
|
);
|
|
|
|
for (const msg of updatedStatusMessages) {
|
|
if (newMessageIds.has(msg.id)) continue;
|
|
const parsed = {
|
|
...msg, contentType: msg.content_type || 'text', senderType: msg.sender_type || 'human',
|
|
toolResults: msg.tool_results ? safeJsonParse(msg.tool_results) : null,
|
|
mentions: safeJsonParse(msg.mentions) || [], attachments: safeJsonParse(msg.attachments) || [],
|
|
timestamp: msg.created_at, metadata: safeJsonParse(msg.metadata) || {},
|
|
};
|
|
res.write(`event: message_updated\ndata: ${JSON.stringify(parsed)}\n\n`);
|
|
}
|
|
|
|
for (const msg of newMessages) {
|
|
const parsed = {
|
|
...msg, contentType: msg.content_type || 'text', senderType: msg.sender_type || 'human',
|
|
toolResults: msg.tool_results ? safeJsonParse(msg.tool_results) : null,
|
|
mentions: safeJsonParse(msg.mentions) || [], attachments: safeJsonParse(msg.attachments) || [],
|
|
metadata: safeJsonParse(msg.metadata) || {}, timestamp: msg.created_at,
|
|
};
|
|
res.write(`event: message\ndata: ${JSON.stringify(parsed)}\n\n`);
|
|
lastMessageId = msg.id;
|
|
}
|
|
|
|
lastPollTime = pollStart;
|
|
|
|
const conv = await dbGet(
|
|
isPostgres()
|
|
? `SELECT is_processing, processing_agent_name, processing_agent_id, processing_started_at FROM conversations WHERE id = $1`
|
|
: `SELECT is_processing, processing_agent_name, processing_agent_id, processing_started_at FROM conversations WHERE id = ?`,
|
|
[id]
|
|
);
|
|
|
|
if (conv) {
|
|
const statusPayload = {
|
|
is_processing: conv.is_processing ? true : false,
|
|
processing_agent_name: conv.processing_agent_name || null,
|
|
processing_agent_id: conv.processing_agent_id || null,
|
|
processing_started_at: conv.processing_started_at || null,
|
|
};
|
|
|
|
// Attach live context usage if available
|
|
const ctxUsage = getContextUsage(Number(id));
|
|
if (ctxUsage) {
|
|
statusPayload.context_usage = {
|
|
prompt_tokens: ctxUsage.prompt_tokens,
|
|
completion_tokens: ctxUsage.completion_tokens,
|
|
total_tokens: ctxUsage.total_tokens,
|
|
context_window: ctxUsage.context_window,
|
|
model: ctxUsage.model,
|
|
iteration: ctxUsage.iteration,
|
|
max_iterations: ctxUsage.max_iterations,
|
|
};
|
|
}
|
|
|
|
res.write(`event: status\ndata: ${JSON.stringify(statusPayload)}\n\n`);
|
|
}
|
|
} catch (err) {
|
|
apiLogger.error({ err, conversationId: id }, 'SSE stream error');
|
|
}
|
|
}, 500);
|
|
|
|
req.on('close', () => {
|
|
clearInterval(poller);
|
|
clearInterval(heartbeat);
|
|
try { unsubscribeInflight(); } catch (_) {}
|
|
apiLogger.debug({ conversationId: id }, 'SSE stream closed');
|
|
});
|
|
});
|
|
|
|
// POST /conversations/:id/reset-processing
|
|
router.post('/conversations/:id/reset-processing', requireAuth, async (req, res) => {
|
|
try {
|
|
const { id } = req.params;
|
|
await dbRun(
|
|
isPostgres()
|
|
? `UPDATE conversations SET is_processing = false, processing_started_at = NULL, processing_agent_id = NULL, processing_agent_name = NULL, updated_at = NOW() WHERE id = $1`
|
|
: `UPDATE conversations SET is_processing = 0, processing_started_at = NULL, processing_agent_id = NULL, processing_agent_name = NULL, updated_at = datetime('now') WHERE id = ?`,
|
|
[id]
|
|
);
|
|
apiLogger.info({ conversationId: id, userId: req.user?.id }, 'Ticket #36708: Processing state manually reset by user');
|
|
return success(res, { reset: true });
|
|
} catch (err) {
|
|
apiLogger.error({ err }, 'Error resetting processing state');
|
|
return error(res, 'RESET_PROCESSING_ERROR', err.message, 500);
|
|
}
|
|
});
|
|
|
|
// POST /conversations/:id/stop - Stop active agent
|
|
//
|
|
// T-148528 (WP-B): two paths can hold a live agent for this conversation:
|
|
// 1. `agent_jobs` row (legacy chat / Claude Code via processJobLocally)
|
|
// → killed by `cancelJob(jobId)` (SIGTERM → 5s → SIGKILL on worker_pid).
|
|
// 2. ADR-0030 dispatcher run on a bound ticket → no agent_jobs row exists,
|
|
// so we ALSO call `abortDispatcherRunByConversation(convId)` which
|
|
// looks up `conversations.bound_row_id` and signals the live child
|
|
// tracked in `_activeAttempts`.
|
|
// Both legs run unconditionally and best-effort — either may be a no-op.
|
|
router.post('/conversations/:id/stop', requireAuth, async (req, res) => {
|
|
try {
|
|
const { id } = req.params;
|
|
|
|
// Always attempt the dispatcher leg first — it's idempotent and the
|
|
// common case for ticket-bound chats (where no agent_jobs row exists).
|
|
let dispatcherResult = { aborted: false, reason: 'not_invoked' };
|
|
try {
|
|
dispatcherResult = await abortDispatcherRunByConversation(Number(id), { reason: 'user_stop' });
|
|
} catch (dispErr) {
|
|
apiLogger.warn({ err: dispErr.message, conversationId: id }, 'T-148528: dispatcher abort threw (non-blocking)');
|
|
}
|
|
|
|
const jobs = await getJobsForConversation(Number(id));
|
|
const activeJob = jobs?.find(j => j.status === 'processing' || j.status === 'pending');
|
|
|
|
if (!activeJob) {
|
|
// No legacy job. Still flip processing state so the UI unblocks
|
|
// even when only the dispatcher leg fired (or neither did, in which
|
|
// case this is the original behaviour).
|
|
await dbRun(
|
|
isPostgres()
|
|
? `UPDATE conversations SET is_processing = false, processing_started_at = NULL, processing_agent_id = NULL, processing_agent_name = NULL, updated_at = NOW() WHERE id = $1`
|
|
: `UPDATE conversations SET is_processing = 0, processing_started_at = NULL, processing_agent_id = NULL, processing_agent_name = NULL, updated_at = datetime('now') WHERE id = ?`,
|
|
[id]
|
|
);
|
|
apiLogger.info({ conversationId: id, dispatcherAborted: dispatcherResult.aborted, dispatcherTicketId: dispatcherResult.ticket_id ?? null, userId: req.user?.id }, 'T-148528: /stop — no legacy job, dispatcher leg reported');
|
|
return success(res, {
|
|
stopped: true,
|
|
job_id: null,
|
|
dispatcher_aborted: dispatcherResult.aborted,
|
|
dispatcher_ticket_id: dispatcherResult.ticket_id ?? null,
|
|
message: dispatcherResult.aborted
|
|
? 'Dispatcher run aborted, processing state reset'
|
|
: 'No active job, processing state reset',
|
|
});
|
|
}
|
|
|
|
const result = await cancelJob(activeJob.id);
|
|
if (!result.success) return badRequest(res, result.error);
|
|
|
|
apiLogger.info({
|
|
conversationId: id, jobId: activeJob.id, agentName: activeJob.agent_name,
|
|
dispatcherAborted: dispatcherResult.aborted, dispatcherTicketId: dispatcherResult.ticket_id ?? null,
|
|
userId: req.user?.id,
|
|
}, 'T-148528: /stop — agent stopped by user');
|
|
return success(res, {
|
|
stopped: true,
|
|
job_id: activeJob.id,
|
|
agent_name: activeJob.agent_name,
|
|
dispatcher_aborted: dispatcherResult.aborted,
|
|
dispatcher_ticket_id: dispatcherResult.ticket_id ?? null,
|
|
});
|
|
} catch (err) {
|
|
apiLogger.error({ err }, 'Error stopping agent');
|
|
return error(res, 'STOP_AGENT_ERROR', err.message, 500);
|
|
}
|
|
});
|
|
}
|