feat: Implement Worker Service for long-running HTTP service with PM2 management
- Introduced WorkerService class to handle HTTP requests and manage sessions. - Added endpoints for health check, session management, and data retrieval. - Integrated ChromaSync for background data synchronization. - Implemented SSE for real-time updates to connected clients. - Added error handling and logging throughout the service. - Cached Claude executable path for improved performance. - Included settings management for user configuration. - Established database interactions for session and observation management.
This commit is contained in:
Alex Newman
@@ -1209,22 +1209,10 @@ export class SessionStore {
|
||||
stmt.run(now.toISOString(), nowEpoch, id);
|
||||
}
|
||||
|
||||
/**
|
||||
* Clean up orphaned active sessions (called on worker startup)
|
||||
*/
|
||||
cleanupOrphanedSessions(): number {
|
||||
const now = new Date();
|
||||
const nowEpoch = now.getTime();
|
||||
|
||||
const stmt = this.db.prepare(`
|
||||
UPDATE sdk_sessions
|
||||
SET status = 'failed', completed_at = ?, completed_at_epoch = ?
|
||||
WHERE status = 'active'
|
||||
`);
|
||||
|
||||
const result = stmt.run(now.toISOString(), nowEpoch);
|
||||
return result.changes;
|
||||
}
|
||||
// REMOVED: cleanupOrphanedSessions - violates "EVERYTHING SHOULD SAVE ALWAYS"
|
||||
// There's no such thing as an "orphaned" session. Sessions are created by hooks
|
||||
// and managed by Claude Code's lifecycle. Worker restarts don't invalidate them.
|
||||
// Marking all active sessions as 'failed' on startup destroys the user's current work.
|
||||
|
||||
/**
|
||||
* Get session summaries by IDs (for hybrid Chroma search)
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -1,487 +0,0 @@
|
||||
/**
|
||||
* Worker Service v2: Clean Object-Oriented Architecture
|
||||
*
|
||||
* This is a complete rewrite following the architecture document.
|
||||
* Key improvements:
|
||||
* - Single database connection (no open/close churn)
|
||||
* - Event-driven queues (zero polling)
|
||||
* - DRY utilities for pagination and settings
|
||||
* - Clean separation of concerns
|
||||
* - ~600-700 lines (down from 1173)
|
||||
*/
|
||||
|
||||
import express, { Request, Response } from 'express';
|
||||
import cors from 'cors';
|
||||
import http from 'http';
|
||||
import path from 'path';
|
||||
import { readFileSync } from 'fs';
|
||||
import { getPackageRoot } from '../shared/paths.js';
|
||||
import { getWorkerPort } from '../shared/worker-utils.js';
|
||||
import { logger } from '../utils/logger.js';
|
||||
|
||||
// Import composed services
|
||||
import { DatabaseManager } from './worker/DatabaseManager.js';
|
||||
import { SessionManager } from './worker/SessionManager.js';
|
||||
import { SSEBroadcaster } from './worker/SSEBroadcaster.js';
|
||||
import { SDKAgent } from './worker/SDKAgent.js';
|
||||
import { PaginationHelper } from './worker/PaginationHelper.js';
|
||||
import { SettingsManager } from './worker/SettingsManager.js';
|
||||
|
||||
export class WorkerService {
|
||||
private app: express.Application;
|
||||
private server: http.Server | null = null;
|
||||
|
||||
// Composed services
|
||||
private dbManager: DatabaseManager;
|
||||
private sessionManager: SessionManager;
|
||||
private sseBroadcaster: SSEBroadcaster;
|
||||
private sdkAgent: SDKAgent;
|
||||
private paginationHelper: PaginationHelper;
|
||||
private settingsManager: SettingsManager;
|
||||
|
||||
constructor() {
|
||||
this.app = express();
|
||||
|
||||
// Initialize services (dependency injection)
|
||||
this.dbManager = new DatabaseManager();
|
||||
this.sessionManager = new SessionManager(this.dbManager);
|
||||
this.sseBroadcaster = new SSEBroadcaster();
|
||||
this.sdkAgent = new SDKAgent(this.dbManager, this.sessionManager);
|
||||
this.paginationHelper = new PaginationHelper(this.dbManager);
|
||||
this.settingsManager = new SettingsManager(this.dbManager);
|
||||
|
||||
this.setupMiddleware();
|
||||
this.setupRoutes();
|
||||
}
|
||||
|
||||
/**
|
||||
* Setup Express middleware
|
||||
*/
|
||||
private setupMiddleware(): void {
|
||||
this.app.use(express.json({ limit: '50mb' }));
|
||||
this.app.use(cors());
|
||||
}
|
||||
|
||||
/**
|
||||
* Setup HTTP routes
|
||||
*/
|
||||
private setupRoutes(): void {
|
||||
// Health & Viewer
|
||||
this.app.get('/health', this.handleHealth.bind(this));
|
||||
this.app.get('/', this.handleViewerUI.bind(this));
|
||||
this.app.get('/stream', this.handleSSEStream.bind(this));
|
||||
|
||||
// Session endpoints
|
||||
this.app.post('/sessions/:sessionDbId/init', this.handleSessionInit.bind(this));
|
||||
this.app.post('/sessions/:sessionDbId/observations', this.handleObservations.bind(this));
|
||||
this.app.post('/sessions/:sessionDbId/summarize', this.handleSummarize.bind(this));
|
||||
this.app.get('/sessions/:sessionDbId/status', this.handleSessionStatus.bind(this));
|
||||
this.app.delete('/sessions/:sessionDbId', this.handleSessionDelete.bind(this));
|
||||
this.app.post('/sessions/:sessionDbId/complete', this.handleSessionComplete.bind(this));
|
||||
|
||||
// Data retrieval
|
||||
this.app.get('/api/observations', this.handleGetObservations.bind(this));
|
||||
this.app.get('/api/summaries', this.handleGetSummaries.bind(this));
|
||||
this.app.get('/api/prompts', this.handleGetPrompts.bind(this));
|
||||
this.app.get('/api/stats', this.handleGetStats.bind(this));
|
||||
|
||||
// Settings
|
||||
this.app.get('/api/settings', this.handleGetSettings.bind(this));
|
||||
this.app.post('/api/settings', this.handleUpdateSettings.bind(this));
|
||||
}
|
||||
|
||||
/**
|
||||
* Start the worker service
|
||||
*/
|
||||
async start(): Promise<void> {
|
||||
// Initialize database (once, stays open)
|
||||
await this.dbManager.initialize();
|
||||
|
||||
// Cleanup orphaned sessions from previous runs
|
||||
const cleaned = this.dbManager.cleanupOrphanedSessions();
|
||||
if (cleaned > 0) {
|
||||
logger.info('SYSTEM', `Cleaned ${cleaned} orphaned sessions`);
|
||||
}
|
||||
|
||||
// Start HTTP server
|
||||
const port = getWorkerPort();
|
||||
this.server = await new Promise<http.Server>((resolve, reject) => {
|
||||
const srv = this.app.listen(port, () => resolve(srv));
|
||||
srv.on('error', reject);
|
||||
});
|
||||
|
||||
logger.info('SYSTEM', 'Worker started', { port, pid: process.pid });
|
||||
}
|
||||
|
||||
/**
|
||||
* Shutdown the worker service
|
||||
*/
|
||||
async shutdown(): Promise<void> {
|
||||
// Shutdown all active sessions
|
||||
await this.sessionManager.shutdownAll();
|
||||
|
||||
// Close HTTP server
|
||||
if (this.server) {
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
this.server!.close(err => err ? reject(err) : resolve());
|
||||
});
|
||||
}
|
||||
|
||||
// Close database connection
|
||||
await this.dbManager.close();
|
||||
|
||||
logger.info('SYSTEM', 'Worker shutdown complete');
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Route Handlers
|
||||
// ============================================================================
|
||||
|
||||
/**
|
||||
* Health check endpoint
|
||||
*/
|
||||
private handleHealth(req: Request, res: Response): void {
|
||||
res.json({ status: 'ok', timestamp: Date.now() });
|
||||
}
|
||||
|
||||
/**
|
||||
* Serve viewer UI
|
||||
*/
|
||||
private handleViewerUI(req: Request, res: Response): void {
|
||||
try {
|
||||
const packageRoot = getPackageRoot();
|
||||
const viewerPath = path.join(packageRoot, 'plugin', 'ui', 'viewer.html');
|
||||
const html = readFileSync(viewerPath, 'utf-8');
|
||||
res.setHeader('Content-Type', 'text/html');
|
||||
res.send(html);
|
||||
} catch (error) {
|
||||
logger.failure('WORKER', 'Viewer UI error', {}, error as Error);
|
||||
res.status(500).json({ error: 'Failed to load viewer UI' });
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* SSE stream endpoint
|
||||
*/
|
||||
private handleSSEStream(req: Request, res: Response): void {
|
||||
// Setup SSE headers
|
||||
res.setHeader('Content-Type', 'text/event-stream');
|
||||
res.setHeader('Cache-Control', 'no-cache');
|
||||
res.setHeader('Connection', 'keep-alive');
|
||||
|
||||
// Add client to broadcaster
|
||||
this.sseBroadcaster.addClient(res);
|
||||
}
|
||||
|
||||
/**
|
||||
* Initialize a new session
|
||||
*/
|
||||
private handleSessionInit(req: Request, res: Response): void {
|
||||
try {
|
||||
const sessionDbId = parseInt(req.params.sessionDbId, 10);
|
||||
const session = this.sessionManager.initializeSession(sessionDbId);
|
||||
|
||||
// Start SDK agent in background
|
||||
this.sdkAgent.startSession(session).catch(err => {
|
||||
logger.failure('WORKER', 'SDK agent error', { sessionId: sessionDbId }, err);
|
||||
});
|
||||
|
||||
// Broadcast SSE event
|
||||
this.sseBroadcaster.broadcast({
|
||||
type: 'session_started',
|
||||
sessionDbId,
|
||||
project: session.project
|
||||
});
|
||||
|
||||
res.json({ status: 'initialized', sessionDbId, port: getWorkerPort() });
|
||||
} catch (error) {
|
||||
logger.failure('WORKER', 'Session init failed', {}, error as Error);
|
||||
res.status(500).json({ error: (error as Error).message });
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Queue observations for processing
|
||||
*/
|
||||
private handleObservations(req: Request, res: Response): void {
|
||||
try {
|
||||
const sessionDbId = parseInt(req.params.sessionDbId, 10);
|
||||
const { tool_name, tool_input, tool_response, prompt_number } = req.body;
|
||||
|
||||
this.sessionManager.queueObservation(sessionDbId, {
|
||||
tool_name,
|
||||
tool_input,
|
||||
tool_response,
|
||||
prompt_number
|
||||
});
|
||||
|
||||
// Broadcast SSE event
|
||||
this.sseBroadcaster.broadcast({
|
||||
type: 'observation_queued',
|
||||
sessionDbId
|
||||
});
|
||||
|
||||
res.json({ status: 'queued' });
|
||||
} catch (error) {
|
||||
logger.failure('WORKER', 'Observation queuing failed', {}, error as Error);
|
||||
res.status(500).json({ error: (error as Error).message });
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Queue summarize request
|
||||
*/
|
||||
private handleSummarize(req: Request, res: Response): void {
|
||||
try {
|
||||
const sessionDbId = parseInt(req.params.sessionDbId, 10);
|
||||
this.sessionManager.queueSummarize(sessionDbId);
|
||||
|
||||
res.json({ status: 'queued' });
|
||||
} catch (error) {
|
||||
logger.failure('WORKER', 'Summarize queuing failed', {}, error as Error);
|
||||
res.status(500).json({ error: (error as Error).message });
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get session status
|
||||
*/
|
||||
private handleSessionStatus(req: Request, res: Response): void {
|
||||
try {
|
||||
const sessionDbId = parseInt(req.params.sessionDbId, 10);
|
||||
const session = this.sessionManager.getSession(sessionDbId);
|
||||
|
||||
if (!session) {
|
||||
res.json({ status: 'not_found' });
|
||||
return;
|
||||
}
|
||||
|
||||
res.json({
|
||||
status: 'active',
|
||||
sessionDbId,
|
||||
project: session.project,
|
||||
queueLength: session.pendingMessages.length,
|
||||
uptime: Date.now() - session.startTime
|
||||
});
|
||||
} catch (error) {
|
||||
logger.failure('WORKER', 'Session status failed', {}, error as Error);
|
||||
res.status(500).json({ error: (error as Error).message });
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Delete a session
|
||||
*/
|
||||
private async handleSessionDelete(req: Request, res: Response): Promise<void> {
|
||||
try {
|
||||
const sessionDbId = parseInt(req.params.sessionDbId, 10);
|
||||
await this.sessionManager.deleteSession(sessionDbId);
|
||||
|
||||
// Mark session complete in database
|
||||
this.dbManager.markSessionComplete(sessionDbId);
|
||||
|
||||
// Broadcast SSE event
|
||||
this.sseBroadcaster.broadcast({
|
||||
type: 'session_completed',
|
||||
sessionDbId
|
||||
});
|
||||
|
||||
res.json({ status: 'deleted' });
|
||||
} catch (error) {
|
||||
logger.failure('WORKER', 'Session delete failed', {}, error as Error);
|
||||
res.status(500).json({ error: (error as Error).message });
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Complete a session (backward compatibility for cleanup-hook)
|
||||
* cleanup-hook expects POST /sessions/:sessionDbId/complete instead of DELETE
|
||||
*/
|
||||
private async handleSessionComplete(req: Request, res: Response): Promise<void> {
|
||||
try {
|
||||
const sessionDbId = parseInt(req.params.sessionDbId, 10);
|
||||
if (isNaN(sessionDbId)) {
|
||||
res.status(400).json({ success: false, error: 'Invalid session ID' });
|
||||
return;
|
||||
}
|
||||
|
||||
await this.sessionManager.deleteSession(sessionDbId);
|
||||
|
||||
// Mark session complete in database
|
||||
this.dbManager.markSessionComplete(sessionDbId);
|
||||
|
||||
// Broadcast SSE event
|
||||
this.sseBroadcaster.broadcast({
|
||||
type: 'session_completed',
|
||||
timestamp: Date.now(),
|
||||
sessionDbId
|
||||
});
|
||||
|
||||
res.json({ success: true });
|
||||
} catch (error) {
|
||||
logger.failure('WORKER', 'Session complete failed', {}, error as Error);
|
||||
res.status(500).json({ success: false, error: String(error) });
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get paginated observations
|
||||
*/
|
||||
private handleGetObservations(req: Request, res: Response): void {
|
||||
try {
|
||||
const { offset, limit, project } = parsePaginationParams(req);
|
||||
const result = this.paginationHelper.getObservations(offset, limit, project);
|
||||
res.json(result);
|
||||
} catch (error) {
|
||||
logger.failure('WORKER', 'Get observations failed', {}, error as Error);
|
||||
res.status(500).json({ error: (error as Error).message });
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get paginated summaries
|
||||
*/
|
||||
private handleGetSummaries(req: Request, res: Response): void {
|
||||
try {
|
||||
const { offset, limit, project } = parsePaginationParams(req);
|
||||
const result = this.paginationHelper.getSummaries(offset, limit, project);
|
||||
res.json(result);
|
||||
} catch (error) {
|
||||
logger.failure('WORKER', 'Get summaries failed', {}, error as Error);
|
||||
res.status(500).json({ error: (error as Error).message });
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get paginated user prompts
|
||||
*/
|
||||
private handleGetPrompts(req: Request, res: Response): void {
|
||||
try {
|
||||
const { offset, limit, project } = parsePaginationParams(req);
|
||||
const result = this.paginationHelper.getPrompts(offset, limit, project);
|
||||
res.json(result);
|
||||
} catch (error) {
|
||||
logger.failure('WORKER', 'Get prompts failed', {}, error as Error);
|
||||
res.status(500).json({ error: (error as Error).message });
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get database statistics
|
||||
*/
|
||||
private handleGetStats(req: Request, res: Response): void {
|
||||
try {
|
||||
const db = this.dbManager.getSessionStore().db;
|
||||
|
||||
// Get total counts
|
||||
const totalObservations = db.prepare('SELECT COUNT(*) as count FROM observations').get() as { count: number };
|
||||
const totalSessions = db.prepare('SELECT COUNT(*) as count FROM sessions').get() as { count: number };
|
||||
const totalPrompts = db.prepare('SELECT COUNT(*) as count FROM user_prompts').get() as { count: number };
|
||||
const totalSummaries = db.prepare('SELECT COUNT(*) as count FROM summaries').get() as { count: number };
|
||||
|
||||
// Get project counts
|
||||
const projectCounts: Record<string, any> = {};
|
||||
|
||||
const projects = db.prepare('SELECT DISTINCT project FROM observations').all() as Array<{ project: string }>;
|
||||
|
||||
for (const { project } of projects) {
|
||||
const obsCount = db.prepare('SELECT COUNT(*) as count FROM observations WHERE project = ?').get(project) as { count: number };
|
||||
const sessCount = db.prepare('SELECT COUNT(*) as count FROM sessions WHERE project = ?').get(project) as { count: number };
|
||||
const promptCount = db.prepare('SELECT COUNT(*) as count FROM user_prompts WHERE project = ?').get(project) as { count: number };
|
||||
const summCount = db.prepare('SELECT COUNT(*) as count FROM summaries WHERE project = ?').get(project) as { count: number };
|
||||
|
||||
projectCounts[project] = {
|
||||
observations: obsCount.count,
|
||||
sessions: sessCount.count,
|
||||
prompts: promptCount.count,
|
||||
summaries: summCount.count
|
||||
};
|
||||
}
|
||||
|
||||
res.json({
|
||||
totalObservations: totalObservations.count,
|
||||
totalSessions: totalSessions.count,
|
||||
totalPrompts: totalPrompts.count,
|
||||
totalSummaries: totalSummaries.count,
|
||||
projectCounts
|
||||
});
|
||||
} catch (error) {
|
||||
logger.failure('WORKER', 'Get stats failed', {}, error as Error);
|
||||
res.status(500).json({ error: (error as Error).message });
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get viewer settings
|
||||
*/
|
||||
private handleGetSettings(req: Request, res: Response): void {
|
||||
try {
|
||||
const settings = this.settingsManager.getSettings();
|
||||
res.json(settings);
|
||||
} catch (error) {
|
||||
logger.failure('WORKER', 'Get settings failed', {}, error as Error);
|
||||
res.status(500).json({ error: (error as Error).message });
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Update viewer settings
|
||||
*/
|
||||
private handleUpdateSettings(req: Request, res: Response): void {
|
||||
try {
|
||||
const updates = req.body;
|
||||
const settings = this.settingsManager.updateSettings(updates);
|
||||
res.json(settings);
|
||||
} catch (error) {
|
||||
logger.failure('WORKER', 'Update settings failed', {}, error as Error);
|
||||
res.status(500).json({ error: (error as Error).message });
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Utilities
|
||||
// ============================================================================
|
||||
|
||||
/**
|
||||
* Parse pagination parameters from request
|
||||
*/
|
||||
function parsePaginationParams(req: Request): { offset: number; limit: number; project?: string } {
|
||||
const offset = parseInt(req.query.offset as string, 10) || 0;
|
||||
const limit = Math.min(parseInt(req.query.limit as string, 10) || 20, 100); // Max 100
|
||||
const project = req.query.project as string | undefined;
|
||||
|
||||
return { offset, limit, project };
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Main Entry Point
|
||||
// ============================================================================
|
||||
|
||||
/**
|
||||
* Start the worker service (if running as main module)
|
||||
*/
|
||||
if (import.meta.url === `file://${process.argv[1]}`) {
|
||||
const worker = new WorkerService();
|
||||
|
||||
// Graceful shutdown
|
||||
process.on('SIGTERM', async () => {
|
||||
logger.info('SYSTEM', 'Received SIGTERM, shutting down gracefully');
|
||||
await worker.shutdown();
|
||||
process.exit(0);
|
||||
});
|
||||
|
||||
process.on('SIGINT', async () => {
|
||||
logger.info('SYSTEM', 'Received SIGINT, shutting down gracefully');
|
||||
await worker.shutdown();
|
||||
process.exit(0);
|
||||
});
|
||||
|
||||
// Start the worker
|
||||
worker.start().catch(error => {
|
||||
logger.failure('SYSTEM', 'Worker startup failed', {}, error);
|
||||
process.exit(1);
|
||||
});
|
||||
}
|
||||
|
||||
export default WorkerService;
|
||||
+474
-970
File diff suppressed because it is too large
Load Diff
@@ -81,15 +81,17 @@ export interface ViewerSettings {
|
||||
|
||||
export interface Observation {
|
||||
id: number;
|
||||
session_db_id: number;
|
||||
claude_session_id: string;
|
||||
sdk_session_id: string;
|
||||
project: string;
|
||||
type: string;
|
||||
title: string;
|
||||
subtitle: string | null;
|
||||
text: string;
|
||||
text: string | null;
|
||||
narrative: string | null;
|
||||
facts: string | null;
|
||||
concepts: string | null;
|
||||
files: string | null;
|
||||
files_read: string | null;
|
||||
files_modified: string | null;
|
||||
prompt_number: number;
|
||||
created_at: string;
|
||||
created_at_epoch: number;
|
||||
@@ -97,13 +99,12 @@ export interface Observation {
|
||||
|
||||
export interface Summary {
|
||||
id: number;
|
||||
session_db_id: number;
|
||||
claude_session_id: string;
|
||||
session_id: string; // claude_session_id (from JOIN)
|
||||
project: string;
|
||||
request: string | null;
|
||||
completion: string | null;
|
||||
summary: string;
|
||||
learnings: string | null;
|
||||
learned: string | null;
|
||||
completed: string | null;
|
||||
next_steps: string | null;
|
||||
notes: string | null;
|
||||
created_at: string;
|
||||
created_at_epoch: number;
|
||||
@@ -111,10 +112,10 @@ export interface Summary {
|
||||
|
||||
export interface UserPrompt {
|
||||
id: number;
|
||||
session_db_id: number;
|
||||
claude_session_id: string;
|
||||
project: string;
|
||||
prompt: string;
|
||||
project: string; // From JOIN with sdk_sessions
|
||||
prompt_number: number;
|
||||
prompt_text: string;
|
||||
created_at: string;
|
||||
created_at_epoch: number;
|
||||
}
|
||||
@@ -150,9 +151,10 @@ export interface ParsedObservation {
|
||||
|
||||
export interface ParsedSummary {
|
||||
request: string | null;
|
||||
completion: string | null;
|
||||
summary: string;
|
||||
learnings: string | null;
|
||||
investigated: string | null;
|
||||
learned: string | null;
|
||||
completed: string | null;
|
||||
next_steps: string | null;
|
||||
notes: string | null;
|
||||
}
|
||||
|
||||
|
||||
@@ -81,13 +81,9 @@ export class DatabaseManager {
|
||||
return this.chromaSync;
|
||||
}
|
||||
|
||||
/**
|
||||
* Cleanup orphaned sessions from previous runs
|
||||
* @returns Number of sessions cleaned
|
||||
*/
|
||||
cleanupOrphanedSessions(): number {
|
||||
return this.getSessionStore().cleanupOrphanedSessions();
|
||||
}
|
||||
// REMOVED: cleanupOrphanedSessions - violates "EVERYTHING SHOULD SAVE ALWAYS"
|
||||
// Worker restarts don't make sessions orphaned. Sessions are managed by hooks
|
||||
// and exist independently of worker state.
|
||||
|
||||
/**
|
||||
* Get session by ID (throws if not found)
|
||||
|
||||
@@ -23,7 +23,7 @@ export class PaginationHelper {
|
||||
getObservations(offset: number, limit: number, project?: string): PaginatedResult<Observation> {
|
||||
return this.paginate<Observation>(
|
||||
'observations',
|
||||
'id, session_db_id, claude_session_id, project, type, title, subtitle, text, concepts, files, prompt_number, created_at, created_at_epoch',
|
||||
'id, sdk_session_id, project, type, title, subtitle, narrative, text, facts, concepts, files_read, files_modified, prompt_number, created_at, created_at_epoch',
|
||||
offset,
|
||||
limit,
|
||||
project
|
||||
@@ -34,26 +34,73 @@ export class PaginationHelper {
|
||||
* Get paginated summaries
|
||||
*/
|
||||
getSummaries(offset: number, limit: number, project?: string): PaginatedResult<Summary> {
|
||||
return this.paginate<Summary>(
|
||||
'summaries',
|
||||
'id, session_db_id, claude_session_id, project, request, completion, summary, learnings, notes, created_at, created_at_epoch',
|
||||
const db = this.dbManager.getSessionStore().db;
|
||||
|
||||
let query = `
|
||||
SELECT
|
||||
ss.id,
|
||||
s.claude_session_id as session_id,
|
||||
ss.request,
|
||||
ss.learned,
|
||||
ss.completed,
|
||||
ss.next_steps,
|
||||
ss.project,
|
||||
ss.created_at,
|
||||
ss.created_at_epoch
|
||||
FROM session_summaries ss
|
||||
JOIN sdk_sessions s ON ss.sdk_session_id = s.sdk_session_id
|
||||
`;
|
||||
const params: any[] = [];
|
||||
|
||||
if (project) {
|
||||
query += ' WHERE ss.project = ?';
|
||||
params.push(project);
|
||||
}
|
||||
|
||||
query += ' ORDER BY ss.created_at_epoch DESC LIMIT ? OFFSET ?';
|
||||
params.push(limit + 1, offset);
|
||||
|
||||
const stmt = db.prepare(query);
|
||||
const results = stmt.all(...params) as Summary[];
|
||||
|
||||
return {
|
||||
items: results.slice(0, limit),
|
||||
hasMore: results.length > limit,
|
||||
offset,
|
||||
limit,
|
||||
project
|
||||
);
|
||||
limit
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Get paginated user prompts
|
||||
*/
|
||||
getPrompts(offset: number, limit: number, project?: string): PaginatedResult<UserPrompt> {
|
||||
return this.paginate<UserPrompt>(
|
||||
'user_prompts',
|
||||
'id, session_db_id, claude_session_id, project, prompt, created_at, created_at_epoch',
|
||||
const db = this.dbManager.getSessionStore().db;
|
||||
|
||||
let query = `
|
||||
SELECT up.id, up.claude_session_id, s.project, up.prompt_number, up.prompt_text, up.created_at, up.created_at_epoch
|
||||
FROM user_prompts up
|
||||
JOIN sdk_sessions s ON up.claude_session_id = s.claude_session_id
|
||||
`;
|
||||
const params: any[] = [];
|
||||
|
||||
if (project) {
|
||||
query += ' WHERE s.project = ?';
|
||||
params.push(project);
|
||||
}
|
||||
|
||||
query += ' ORDER BY up.created_at_epoch DESC LIMIT ? OFFSET ?';
|
||||
params.push(limit + 1, offset);
|
||||
|
||||
const stmt = db.prepare(query);
|
||||
const results = stmt.all(...params) as UserPrompt[];
|
||||
|
||||
return {
|
||||
items: results.slice(0, limit),
|
||||
hasMore: results.length > limit,
|
||||
offset,
|
||||
limit,
|
||||
project
|
||||
);
|
||||
limit
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -34,8 +34,9 @@ export class SDKAgent {
|
||||
|
||||
/**
|
||||
* Start SDK agent for a session (event-driven, no polling)
|
||||
* @param worker WorkerService reference for spinner control (optional)
|
||||
*/
|
||||
async startSession(session: ActiveSession): Promise<void> {
|
||||
async startSession(session: ActiveSession, worker?: any): Promise<void> {
|
||||
try {
|
||||
// Find Claude executable
|
||||
const claudePath = this.findClaudeExecutable();
|
||||
@@ -74,7 +75,7 @@ export class SDKAgent {
|
||||
});
|
||||
|
||||
// Parse and process response
|
||||
await this.processSDKResponse(session, textContent);
|
||||
await this.processSDKResponse(session, textContent, worker);
|
||||
}
|
||||
|
||||
// Log result messages
|
||||
@@ -168,7 +169,7 @@ export class SDKAgent {
|
||||
/**
|
||||
* Process SDK response text (parse XML, save to database, sync to Chroma)
|
||||
*/
|
||||
private async processSDKResponse(session: ActiveSession, text: string): Promise<void> {
|
||||
private async processSDKResponse(session: ActiveSession, text: string, worker?: any): Promise<void> {
|
||||
// Parse observations
|
||||
const observations = parseObservations(text, session.claudeSessionId);
|
||||
|
||||
@@ -191,6 +192,23 @@ export class SDKAgent {
|
||||
createdAtEpoch
|
||||
).catch(() => {});
|
||||
|
||||
// Broadcast to SSE clients (for web UI)
|
||||
if (worker && worker.sseBroadcaster) {
|
||||
worker.sseBroadcaster.broadcast({
|
||||
type: 'new_observation',
|
||||
observation: {
|
||||
id: obsId,
|
||||
session_id: session.claudeSessionId,
|
||||
type: obs.type,
|
||||
title: obs.title,
|
||||
subtitle: obs.subtitle,
|
||||
project: session.project,
|
||||
prompt_number: session.lastPromptNumber,
|
||||
created_at_epoch: createdAtEpoch
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
logger.info('SDK', 'Observation saved', { obsId, type: obs.type });
|
||||
}
|
||||
|
||||
@@ -216,8 +234,33 @@ export class SDKAgent {
|
||||
createdAtEpoch
|
||||
).catch(() => {});
|
||||
|
||||
// Broadcast to SSE clients (for web UI)
|
||||
if (worker && worker.sseBroadcaster) {
|
||||
worker.sseBroadcaster.broadcast({
|
||||
type: 'new_summary',
|
||||
summary: {
|
||||
id: summaryId,
|
||||
session_id: session.claudeSessionId,
|
||||
request: summary.request,
|
||||
investigated: summary.investigated,
|
||||
learned: summary.learned,
|
||||
completed: summary.completed,
|
||||
next_steps: summary.next_steps,
|
||||
notes: summary.notes,
|
||||
project: session.project,
|
||||
prompt_number: session.lastPromptNumber,
|
||||
created_at_epoch: createdAtEpoch
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
logger.info('SDK', 'Summary saved', { summaryId });
|
||||
}
|
||||
|
||||
// Check and stop spinner after processing (debounced)
|
||||
if (worker && typeof worker.checkAndStopSpinner === 'function') {
|
||||
worker.checkAndStopSpinner();
|
||||
}
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
|
||||
@@ -44,12 +44,15 @@ export class SSEBroadcaster {
|
||||
*/
|
||||
broadcast(event: SSEEvent): void {
|
||||
if (this.sseClients.size === 0) {
|
||||
logger.debug('WORKER', 'SSE broadcast skipped (no clients)', { eventType: event.type });
|
||||
return; // Short-circuit if no clients
|
||||
}
|
||||
|
||||
const eventWithTimestamp = { ...event, timestamp: Date.now() };
|
||||
const data = `data: ${JSON.stringify(eventWithTimestamp)}\n\n`;
|
||||
|
||||
logger.debug('WORKER', 'SSE broadcast sent', { eventType: event.type, clients: this.sseClients.size });
|
||||
|
||||
// Single-pass write + cleanup
|
||||
for (const client of this.sseClients) {
|
||||
try {
|
||||
|
||||
@@ -69,11 +69,13 @@ export class SessionManager {
|
||||
|
||||
/**
|
||||
* Queue an observation for processing (zero-latency notification)
|
||||
* Auto-initializes session if not in memory but exists in database
|
||||
*/
|
||||
queueObservation(sessionDbId: number, data: ObservationData): void {
|
||||
const session = this.sessions.get(sessionDbId);
|
||||
// Auto-initialize from database if needed (handles worker restarts)
|
||||
let session = this.sessions.get(sessionDbId);
|
||||
if (!session) {
|
||||
throw new Error(`Session ${sessionDbId} not active`);
|
||||
session = this.initializeSession(sessionDbId);
|
||||
}
|
||||
|
||||
session.pendingMessages.push({
|
||||
@@ -96,11 +98,13 @@ export class SessionManager {
|
||||
|
||||
/**
|
||||
* Queue a summarize request (zero-latency notification)
|
||||
* Auto-initializes session if not in memory but exists in database
|
||||
*/
|
||||
queueSummarize(sessionDbId: number): void {
|
||||
const session = this.sessions.get(sessionDbId);
|
||||
// Auto-initialize from database if needed (handles worker restarts)
|
||||
let session = this.sessions.get(sessionDbId);
|
||||
if (!session) {
|
||||
throw new Error(`Session ${sessionDbId} not active`);
|
||||
session = this.initializeSession(sessionDbId);
|
||||
}
|
||||
|
||||
session.pendingMessages.push({ type: 'summarize' });
|
||||
@@ -143,13 +147,31 @@ export class SessionManager {
|
||||
await Promise.all(sessionIds.map(id => this.deleteSession(id)));
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if any session has pending messages (for spinner tracking)
|
||||
*/
|
||||
hasPendingMessages(): boolean {
|
||||
return Array.from(this.sessions.values()).some(
|
||||
session => session.pendingMessages.length > 0
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Get number of active sessions (for stats)
|
||||
*/
|
||||
getActiveSessionCount(): number {
|
||||
return this.sessions.size;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get message iterator for SDKAgent to consume (event-driven, no polling)
|
||||
* Auto-initializes session if not in memory but exists in database
|
||||
*/
|
||||
async *getMessageIterator(sessionDbId: number): AsyncIterableIterator<PendingMessage> {
|
||||
const session = this.sessions.get(sessionDbId);
|
||||
// Auto-initialize from database if needed (handles worker restarts)
|
||||
let session = this.sessions.get(sessionDbId);
|
||||
if (!session) {
|
||||
throw new Error(`Session ${sessionDbId} not active`);
|
||||
session = this.initializeSession(sessionDbId);
|
||||
}
|
||||
|
||||
const emitter = this.sessionQueues.get(sessionDbId);
|
||||
|
||||
Reference in New Issue
Block a user