Andreas/LibreChat
1
0
Fork 0
mirror of https://github.com/danny-avila/LibreChat.git synced 2025-12-20 10:20:15 +01:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

WIP: resuming

Browse source
This commit is contained in:
Danny Avila 2025-12-04 08:57:13 -05:00
parent cae8c06c9c
commit b1714a7529
No known key found for this signature in database
GPG key ID: BF31EEB2C5CA0956
8 changed files with 478 additions and 85 deletions
Download patch file Download diff file Expand all files Collapse all files

85
api/server/routes/agents/chat.js
View file

@ -18,8 +18,6 @@ const { getRoleByName } = require('~/models/Role');
const router = express.Router();
router.use(moderateText);
const checkAgentAccess = generateCheckAccess({
permissionType: PermissionTypes.AGENTS,
permissions: [Permissions.USE],
@ -30,77 +28,12 @@ const checkAgentResourceAccess = canAccessAgentFromBody({
requiredPermission: PermissionBits.VIEW,
});
/**
* @route GET /stream/:streamId
* @desc Subscribe to an ongoing generation job's SSE stream
* @access Private
*/
router.get('/stream/:streamId', requireJwtAuth, (req, res) => {
const { streamId } = req.params;
const job = GenerationJobManager.getJob(streamId);
if (!job) {
return res.status(404).json({
error: 'Stream not found',
message: 'The generation job does not exist or has expired.',
});
}
// Disable compression for SSE
res.setHeader('Content-Encoding', 'identity');
res.setHeader('Content-Type', 'text/event-stream');
res.setHeader('Cache-Control', 'no-cache, no-transform');
res.setHeader('Connection', 'keep-alive');
res.setHeader('X-Accel-Buffering', 'no');
res.flushHeaders();
logger.debug(`[AgentStream] Client subscribed to ${streamId}`);
const unsubscribe = GenerationJobManager.subscribe(
streamId,
(event) => {
if (!res.writableEnded) {
res.write(`event: message\ndata: ${JSON.stringify(event)}\n\n`);
if (typeof res.flush === 'function') {
res.flush();
}
}
},
(event) => {
if (!res.writableEnded) {
res.write(`event: message\ndata: ${JSON.stringify(event)}\n\n`);
if (typeof res.flush === 'function') {
res.flush();
}
res.end();
}
},
(error) => {
if (!res.writableEnded) {
res.write(`event: error\ndata: ${JSON.stringify({ error })}\n\n`);
if (typeof res.flush === 'function') {
res.flush();
}
res.end();
}
},
);
if (!unsubscribe) {
return res.status(404).json({ error: 'Failed to subscribe to stream' });
}
if (job.status === 'complete' || job.status === 'error' || job.status === 'aborted') {
res.write(`event: message\ndata: ${JSON.stringify({ final: true, status: job.status })}\n\n`);
res.end();
return;
}
req.on('close', () => {
logger.debug(`[AgentStream] Client disconnected from ${streamId}`);
unsubscribe();
});
});
router.use(moderateText);
router.use(checkAgentAccess);
router.use(checkAgentResourceAccess);
router.use(validateConvoAccess);
router.use(buildEndpointOption);
router.use(setHeaders);
/**
* @route POST /abort
@ -121,12 +54,6 @@ router.post('/abort', (req, res) => {
res.status(404).json({ error: 'Job not found' });
});
router.use(checkAgentAccess);
router.use(checkAgentResourceAccess);
router.use(validateConvoAccess);
router.use(buildEndpointOption);
router.use(setHeaders);
const controller = async (req, res, next) => {
await AgentController(req, res, next, initializeClient, addTitle);
};

105
api/server/routes/agents/index.js
View file

@ -1,5 +1,6 @@
const express = require('express');
const { isEnabled } = require('@librechat/api');
const { isEnabled, GenerationJobManager } = require('@librechat/api');
const { logger } = require('@librechat/data-schemas');
const {
uaParser,
checkBan,
@ -22,6 +23,108 @@ router.use(uaParser);
router.use('/', v1);
/**
* Stream endpoints - mounted before chatRouter to bypass rate limiters
* These are GET requests and don't need message body validation or rate limiting
*/
/**
* @route GET /chat/stream/:streamId
* @desc Subscribe to an ongoing generation job's SSE stream with replay support
* @access Private
* @description Replays any chunks missed during disconnect, then streams live
*/
router.get('/chat/stream/:streamId', (req, res) => {
const { streamId } = req.params;
const job = GenerationJobManager.getJob(streamId);
if (!job) {
return res.status(404).json({
error: 'Stream not found',
message: 'The generation job does not exist or has expired.',
});
}
res.setHeader('Content-Encoding', 'identity');
res.setHeader('Content-Type', 'text/event-stream');
res.setHeader('Cache-Control', 'no-cache, no-transform');
res.setHeader('Connection', 'keep-alive');
res.setHeader('X-Accel-Buffering', 'no');
res.flushHeaders();
logger.debug(`[AgentStream] Client subscribed to ${streamId}`);
const result = GenerationJobManager.subscribe(
streamId,
(event) => {
if (!res.writableEnded) {
res.write(`event: message\ndata: ${JSON.stringify(event)}\n\n`);
if (typeof res.flush === 'function') {
res.flush();
}
}
},
(event) => {
if (!res.writableEnded) {
res.write(`event: message\ndata: ${JSON.stringify(event)}\n\n`);
if (typeof res.flush === 'function') {
res.flush();
}
res.end();
}
},
(error) => {
if (!res.writableEnded) {
res.write(`event: error\ndata: ${JSON.stringify({ error })}\n\n`);
if (typeof res.flush === 'function') {
res.flush();
}
res.end();
}
},
);
if (!result) {
return res.status(404).json({ error: 'Failed to subscribe to stream' });
}
req.on('close', () => {
logger.debug(`[AgentStream] Client disconnected from ${streamId}`);
result.unsubscribe();
});
});
/**
* @route GET /chat/status/:conversationId
* @desc Check if there's an active generation job for a conversation
* @access Private
* @returns { active, streamId, status, chunkCount, aggregatedContent, createdAt }
*/
router.get('/chat/status/:conversationId', (req, res) => {
const { conversationId } = req.params;
const job = GenerationJobManager.getJobByConversation(conversationId);
if (!job) {
return res.json({ active: false });
}
if (job.metadata.userId !== req.user.id) {
return res.status(403).json({ error: 'Unauthorized' });
}
const info = GenerationJobManager.getStreamInfo(job.streamId);
res.json({
active: info?.active ?? false,
streamId: job.streamId,
status: info?.status ?? job.status,
chunkCount: info?.chunkCount ?? 0,
aggregatedContent: info?.aggregatedContent,
createdAt: info?.createdAt ?? job.createdAt,
});
});
const chatRouter = express.Router();
chatRouter.use(configMiddleware);