Assistent: Claude-Proxy auf Host statt Volume-Mount
- Claude CLI Volumes aus docker-compose.yml entfernt - Eigener Proxy-Service auf dem Host (Port 3100, systemd) - assistant.js nutzt HTTP-Request an Proxy statt child_process.spawn - Token-Auth zwischen Container und Proxy - Saubere Trennung: Claude laeuft nur auf dem Host Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Dieser Commit ist enthalten in:
Server Deploy
@@ -2,11 +2,14 @@
|
||||
* TASKMATE - Claude Assistant Routes
|
||||
* ====================================
|
||||
* REST-Endpunkte und Session-Manager fuer den Claude-Assistenten
|
||||
*
|
||||
* Kommuniziert ueber HTTP mit dem Claude-Proxy (SSE-Streaming)
|
||||
* Proxy-URL: http://172.20.0.1:3100/api/chat
|
||||
*/
|
||||
|
||||
const express = require('express');
|
||||
const router = express.Router();
|
||||
const { spawn } = require('child_process');
|
||||
const http = require('http');
|
||||
const { getDb } = require('../database');
|
||||
const logger = require('../utils/logger');
|
||||
|
||||
@@ -14,18 +17,30 @@ const logger = require('../utils/logger');
|
||||
// SESSION MANAGER
|
||||
// ============================================================================
|
||||
|
||||
// Aktive Prozesse: userId -> { process, sessionId, timeout, outputBuffer, saveTimer }
|
||||
// Aktive Sessions: userId -> { claudeSessionId, busy, sessionId, socket, currentRequest, timeout }
|
||||
const activeSessions = new Map();
|
||||
|
||||
const SESSION_TIMEOUT_MS = 30 * 60 * 1000; // 30 Minuten
|
||||
const DB_SAVE_INTERVAL_MS = 2000; // Output alle 2 Sekunden in DB speichern
|
||||
|
||||
/**
|
||||
* Erlaubte Benutzer fuer den Assistenten (lowercase)
|
||||
*/
|
||||
const ALLOWED_ASSISTANT_USERS = new Set([
|
||||
'hendrik', 'monami',
|
||||
'hendrik_gebhardt@gmx.de', 'momohomma@googlemail.com'
|
||||
]);
|
||||
|
||||
function hasAssistantAccess(user) {
|
||||
const username = (user.username || '').toLowerCase();
|
||||
const displayName = (user.displayName || '').toLowerCase();
|
||||
return ALLOWED_ASSISTANT_USERS.has(username) || ALLOWED_ASSISTANT_USERS.has(displayName);
|
||||
}
|
||||
|
||||
/**
|
||||
* Berechtigungs-Middleware: Nur Hendrik und Monami duerfen den Assistenten nutzen
|
||||
*/
|
||||
function requireAssistantAccess(req, res, next) {
|
||||
const username = (req.user.username || '').toLowerCase();
|
||||
if (username !== 'hendrik' && username !== 'monami') {
|
||||
if (!hasAssistantAccess(req.user)) {
|
||||
return res.status(403).json({ error: 'Kein Zugriff auf den Assistenten' });
|
||||
}
|
||||
next();
|
||||
@@ -35,8 +50,7 @@ function requireAssistantAccess(req, res, next) {
|
||||
* Socket-Berechtigung pruefen
|
||||
*/
|
||||
function checkSocketAccess(socket) {
|
||||
const username = (socket.user.username || '').toLowerCase();
|
||||
return username === 'hendrik' || username === 'monami';
|
||||
return hasAssistantAccess(socket.user);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -51,155 +65,81 @@ function resetTimeout(userId) {
|
||||
}
|
||||
|
||||
session.timeout = setTimeout(() => {
|
||||
logger.info(`[Assistant] Session-Timeout fuer User ${userId} - Prozess wird beendet`);
|
||||
logger.info(`[Assistant] Session-Timeout fuer User ${userId} - wird beendet`);
|
||||
stopSession(userId);
|
||||
}, SESSION_TIMEOUT_MS);
|
||||
}
|
||||
|
||||
/**
|
||||
* Gepufferten Output in DB speichern (Debounce)
|
||||
* Proxy-URL und Token
|
||||
*/
|
||||
function flushOutputBuffer(userId) {
|
||||
const session = activeSessions.get(userId);
|
||||
if (!session || !session.outputBuffer || session.outputBuffer.length === 0) return;
|
||||
|
||||
try {
|
||||
const db = getDb();
|
||||
const content = session.outputBuffer.join('');
|
||||
session.outputBuffer = [];
|
||||
|
||||
db.prepare(`
|
||||
INSERT INTO assistant_messages (session_id, role, content)
|
||||
VALUES (?, 'assistant', ?)
|
||||
`).run(session.sessionId, content);
|
||||
} catch (err) {
|
||||
logger.error(`[Assistant] Fehler beim Speichern des Outputs: ${err.message}`);
|
||||
}
|
||||
}
|
||||
const PROXY_URL = 'http://172.20.0.1:3100/api/chat';
|
||||
const PROXY_TOKEN = process.env.PROXY_TOKEN || '';
|
||||
|
||||
/**
|
||||
* Claude-Prozess starten
|
||||
* Claude-Session aktivieren (kein Prozess - der wird erst bei sendMessage gestartet)
|
||||
*/
|
||||
function startSession(userId, sessionId, socket) {
|
||||
if (activeSessions.has(userId)) {
|
||||
throw new Error('Es laeuft bereits eine aktive Session');
|
||||
const existing = activeSessions.get(userId);
|
||||
|
||||
// Gleiche Session: Nur Socket aktualisieren
|
||||
if (existing && existing.sessionId === sessionId) {
|
||||
existing.socket = socket;
|
||||
resetTimeout(userId);
|
||||
socket.emit('assistant:status', {
|
||||
sessionId,
|
||||
status: existing.busy ? 'thinking' : 'active'
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
const proc = spawn('claude', [], {
|
||||
cwd: '/home/claude-dev/TaskMate',
|
||||
env: { ...process.env, HOME: '/home/claude-dev' },
|
||||
stdio: ['pipe', 'pipe', 'pipe']
|
||||
});
|
||||
// Andere Session: bestehende aufraemen
|
||||
if (existing) {
|
||||
if (existing.currentRequest) {
|
||||
try { existing.currentRequest.destroy(); } catch (e) {}
|
||||
}
|
||||
if (existing.timeout) clearTimeout(existing.timeout);
|
||||
}
|
||||
|
||||
const sessionData = {
|
||||
process: proc,
|
||||
claudeSessionId: null,
|
||||
busy: false,
|
||||
sessionId,
|
||||
timeout: null,
|
||||
outputBuffer: [],
|
||||
saveTimer: null
|
||||
socket,
|
||||
currentRequest: null,
|
||||
timeout: null
|
||||
};
|
||||
|
||||
activeSessions.set(userId, sessionData);
|
||||
resetTimeout(userId);
|
||||
|
||||
logger.info(`[Assistant] Session ${sessionId} gestartet fuer User ${userId} (PID: ${proc.pid})`);
|
||||
logger.info(`[Assistant] Session ${sessionId} aktiviert fuer User ${userId}`);
|
||||
socket.emit('assistant:status', { sessionId, status: 'active' });
|
||||
|
||||
// stdout verarbeiten
|
||||
proc.stdout.on('data', (data) => {
|
||||
const text = data.toString();
|
||||
sessionData.outputBuffer.push(text);
|
||||
|
||||
// An Socket senden
|
||||
if (socket && socket.connected) {
|
||||
socket.emit('assistant:output', { sessionId, content: text, stream: 'stdout' });
|
||||
// TaskContext als erste Nachricht senden
|
||||
try {
|
||||
const db = getDb();
|
||||
const dbSession = db.prepare('SELECT task_context FROM assistant_sessions WHERE id = ?').get(sessionId);
|
||||
if (dbSession && dbSession.task_context) {
|
||||
sendMessage(userId, dbSession.task_context, socket);
|
||||
}
|
||||
|
||||
// Debounced DB-Save
|
||||
if (!sessionData.saveTimer) {
|
||||
sessionData.saveTimer = setTimeout(() => {
|
||||
sessionData.saveTimer = null;
|
||||
flushOutputBuffer(userId);
|
||||
}, DB_SAVE_INTERVAL_MS);
|
||||
}
|
||||
});
|
||||
|
||||
// stderr verarbeiten (gleich wie stdout)
|
||||
proc.stderr.on('data', (data) => {
|
||||
const text = data.toString();
|
||||
sessionData.outputBuffer.push(text);
|
||||
|
||||
if (socket && socket.connected) {
|
||||
socket.emit('assistant:output', { sessionId, content: text, stream: 'stderr' });
|
||||
}
|
||||
|
||||
if (!sessionData.saveTimer) {
|
||||
sessionData.saveTimer = setTimeout(() => {
|
||||
sessionData.saveTimer = null;
|
||||
flushOutputBuffer(userId);
|
||||
}, DB_SAVE_INTERVAL_MS);
|
||||
}
|
||||
});
|
||||
|
||||
// Prozess beendet
|
||||
proc.on('close', (code) => {
|
||||
logger.info(`[Assistant] Prozess beendet fuer User ${userId} (Code: ${code})`);
|
||||
|
||||
// Restlichen Buffer speichern
|
||||
flushOutputBuffer(userId);
|
||||
|
||||
// Aufraumen
|
||||
const session = activeSessions.get(userId);
|
||||
if (session) {
|
||||
if (session.timeout) clearTimeout(session.timeout);
|
||||
if (session.saveTimer) clearTimeout(session.saveTimer);
|
||||
activeSessions.delete(userId);
|
||||
}
|
||||
|
||||
// Session in DB als beendet markieren
|
||||
try {
|
||||
const db = getDb();
|
||||
db.prepare(`
|
||||
UPDATE assistant_sessions SET status = 'ended', ended_at = CURRENT_TIMESTAMP
|
||||
WHERE id = ?
|
||||
`).run(sessionId);
|
||||
} catch (err) {
|
||||
logger.error(`[Assistant] Fehler beim Beenden der Session: ${err.message}`);
|
||||
}
|
||||
|
||||
// Socket benachrichtigen
|
||||
if (socket && socket.connected) {
|
||||
socket.emit('assistant:status', { sessionId, status: 'ended', code });
|
||||
}
|
||||
});
|
||||
|
||||
proc.on('error', (err) => {
|
||||
logger.error(`[Assistant] Prozess-Fehler fuer User ${userId}: ${err.message}`);
|
||||
|
||||
const session = activeSessions.get(userId);
|
||||
if (session) {
|
||||
if (session.timeout) clearTimeout(session.timeout);
|
||||
if (session.saveTimer) clearTimeout(session.saveTimer);
|
||||
activeSessions.delete(userId);
|
||||
}
|
||||
|
||||
if (socket && socket.connected) {
|
||||
socket.emit('assistant:status', { sessionId, status: 'error', error: err.message });
|
||||
}
|
||||
});
|
||||
|
||||
return proc;
|
||||
} catch (err) {
|
||||
logger.error(`[Assistant] TaskContext-Fehler: ${err.message}`);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Nachricht an Claude-Prozess senden
|
||||
* Nachricht an Claude senden (HTTP-Request an Proxy mit SSE-Streaming)
|
||||
*/
|
||||
function sendMessage(userId, message) {
|
||||
function sendMessage(userId, message, socket) {
|
||||
const session = activeSessions.get(userId);
|
||||
if (!session) {
|
||||
throw new Error('Keine aktive Session');
|
||||
}
|
||||
if (!session) throw new Error('Keine aktive Session');
|
||||
if (session.busy) throw new Error('Assistent verarbeitet noch eine Nachricht');
|
||||
|
||||
// In DB speichern
|
||||
session.busy = true;
|
||||
session.socket = socket;
|
||||
|
||||
// User-Nachricht in DB speichern
|
||||
try {
|
||||
const db = getDb();
|
||||
db.prepare(`
|
||||
@@ -207,12 +147,184 @@ function sendMessage(userId, message) {
|
||||
VALUES (?, 'user', ?)
|
||||
`).run(session.sessionId, message);
|
||||
} catch (err) {
|
||||
logger.error(`[Assistant] Fehler beim Speichern der Nachricht: ${err.message}`);
|
||||
logger.error(`[Assistant] DB-Fehler (user msg): ${err.message}`);
|
||||
}
|
||||
|
||||
// An Prozess senden
|
||||
session.process.stdin.write(message + '\n');
|
||||
resetTimeout(userId);
|
||||
socket.emit('assistant:status', { sessionId: session.sessionId, status: 'thinking' });
|
||||
|
||||
logger.info(`[Assistant] Proxy-Aufruf fuer Session ${session.sessionId} (resume: ${session.claudeSessionId || 'nein'})`);
|
||||
|
||||
// HTTP-Request an Proxy
|
||||
const postData = JSON.stringify({
|
||||
message,
|
||||
resumeSessionId: session.claudeSessionId || null
|
||||
});
|
||||
|
||||
const url = new URL(PROXY_URL);
|
||||
const options = {
|
||||
hostname: url.hostname,
|
||||
port: url.port,
|
||||
path: url.pathname,
|
||||
method: 'POST',
|
||||
headers: {
|
||||
'Content-Type': 'application/json',
|
||||
'X-Proxy-Token': PROXY_TOKEN,
|
||||
'Content-Length': Buffer.byteLength(postData)
|
||||
}
|
||||
};
|
||||
|
||||
const req = http.request(options, (res) => {
|
||||
let fullOutput = '';
|
||||
let buffer = '';
|
||||
let currentEvent = null;
|
||||
|
||||
res.setEncoding('utf8');
|
||||
|
||||
res.on('data', (chunk) => {
|
||||
buffer += chunk;
|
||||
const lines = buffer.split('\n');
|
||||
buffer = lines.pop(); // Unvollstaendige Zeile behalten
|
||||
|
||||
for (const line of lines) {
|
||||
// SSE event-Zeile
|
||||
if (line.startsWith('event: ')) {
|
||||
currentEvent = line.substring(7).trim();
|
||||
continue;
|
||||
}
|
||||
|
||||
// SSE data-Zeile
|
||||
if (line.startsWith('data: ')) {
|
||||
const dataStr = line.substring(6);
|
||||
|
||||
if (currentEvent === 'done') {
|
||||
// Done-Event: Enthaelt claudeSessionId
|
||||
try {
|
||||
const json = JSON.parse(dataStr);
|
||||
if (json.sessionId) {
|
||||
session.claudeSessionId = json.sessionId;
|
||||
logger.info(`[Assistant] Claude-SessionId gesetzt: ${json.sessionId}`);
|
||||
}
|
||||
} catch (e) {
|
||||
logger.warn(`[Assistant] Done-Event Parse-Fehler: ${e.message}`);
|
||||
}
|
||||
currentEvent = null;
|
||||
continue;
|
||||
}
|
||||
|
||||
if (currentEvent === 'error') {
|
||||
// Error-Event
|
||||
try {
|
||||
const json = JSON.parse(dataStr);
|
||||
logger.error(`[Assistant] Proxy-Fehler: ${json.error || dataStr}`);
|
||||
if (socket.connected) {
|
||||
socket.emit('assistant:status', {
|
||||
sessionId: session.sessionId,
|
||||
status: 'error',
|
||||
error: json.error || 'Proxy-Fehler'
|
||||
});
|
||||
}
|
||||
} catch (e) {
|
||||
logger.error(`[Assistant] Proxy-Fehler (raw): ${dataStr}`);
|
||||
}
|
||||
currentEvent = null;
|
||||
continue;
|
||||
}
|
||||
|
||||
// Normales Text-Event
|
||||
try {
|
||||
const json = JSON.parse(dataStr);
|
||||
const text = json.text || json.content || '';
|
||||
if (text) {
|
||||
fullOutput += text;
|
||||
if (socket.connected) {
|
||||
socket.emit('assistant:output', { sessionId: session.sessionId, content: text });
|
||||
}
|
||||
}
|
||||
} catch (e) {
|
||||
// Kein valides JSON - Rohtext verwenden
|
||||
if (dataStr.trim()) {
|
||||
fullOutput += dataStr;
|
||||
if (socket.connected) {
|
||||
socket.emit('assistant:output', { sessionId: session.sessionId, content: dataStr });
|
||||
}
|
||||
}
|
||||
}
|
||||
currentEvent = null;
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
res.on('end', () => {
|
||||
// Restlichen Buffer verarbeiten
|
||||
if (buffer.trim()) {
|
||||
const lines = buffer.split('\n');
|
||||
for (const line of lines) {
|
||||
if (line.startsWith('data: ')) {
|
||||
const dataStr = line.substring(6);
|
||||
try {
|
||||
const json = JSON.parse(dataStr);
|
||||
const text = json.text || json.content || '';
|
||||
if (text) {
|
||||
fullOutput += text;
|
||||
if (socket.connected) {
|
||||
socket.emit('assistant:output', { sessionId: session.sessionId, content: text });
|
||||
}
|
||||
}
|
||||
} catch (e) {}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Komplette Antwort in DB speichern
|
||||
if (fullOutput) {
|
||||
try {
|
||||
const db = getDb();
|
||||
db.prepare(`
|
||||
INSERT INTO assistant_messages (session_id, role, content)
|
||||
VALUES (?, 'assistant', ?)
|
||||
`).run(session.sessionId, fullOutput);
|
||||
} catch (err) {
|
||||
logger.error(`[Assistant] DB-Fehler (assistant msg): ${err.message}`);
|
||||
}
|
||||
}
|
||||
|
||||
logger.info(`[Assistant] Proxy-Antwort abgeschlossen fuer Session ${session.sessionId}`);
|
||||
|
||||
session.busy = false;
|
||||
session.currentRequest = null;
|
||||
resetTimeout(userId);
|
||||
|
||||
if (socket.connected) {
|
||||
if (res.statusCode !== 200 && !fullOutput) {
|
||||
socket.emit('assistant:status', {
|
||||
sessionId: session.sessionId,
|
||||
status: 'error',
|
||||
error: `Proxy-Fehler (HTTP ${res.statusCode})`
|
||||
});
|
||||
} else {
|
||||
socket.emit('assistant:status', { sessionId: session.sessionId, status: 'active' });
|
||||
}
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
req.on('error', (err) => {
|
||||
logger.error(`[Assistant] HTTP-Fehler: ${err.message}`);
|
||||
session.busy = false;
|
||||
session.currentRequest = null;
|
||||
|
||||
if (socket.connected) {
|
||||
socket.emit('assistant:status', {
|
||||
sessionId: session.sessionId,
|
||||
status: 'error',
|
||||
error: `Verbindung zum Assistenten fehlgeschlagen: ${err.message}`
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
session.currentRequest = req;
|
||||
req.write(postData);
|
||||
req.end();
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -224,19 +336,10 @@ function stopSession(userId) {
|
||||
|
||||
logger.info(`[Assistant] Session ${session.sessionId} wird beendet fuer User ${userId}`);
|
||||
|
||||
// Restlichen Buffer speichern
|
||||
flushOutputBuffer(userId);
|
||||
|
||||
// Timer aufraemen
|
||||
if (session.timeout) clearTimeout(session.timeout);
|
||||
if (session.saveTimer) clearTimeout(session.saveTimer);
|
||||
|
||||
// Prozess beenden
|
||||
try {
|
||||
session.process.kill('SIGTERM');
|
||||
} catch (err) {
|
||||
logger.error(`[Assistant] Fehler beim Killen des Prozesses: ${err.message}`);
|
||||
if (session.currentRequest) {
|
||||
try { session.currentRequest.destroy(); } catch (e) {}
|
||||
}
|
||||
if (session.timeout) clearTimeout(session.timeout);
|
||||
|
||||
activeSessions.delete(userId);
|
||||
|
||||
@@ -248,7 +351,7 @@ function stopSession(userId) {
|
||||
WHERE id = ?
|
||||
`).run(session.sessionId);
|
||||
} catch (err) {
|
||||
logger.error(`[Assistant] Fehler beim DB-Update: ${err.message}`);
|
||||
logger.error(`[Assistant] DB-Fehler beim Beenden: ${err.message}`);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -380,7 +483,7 @@ router.delete('/sessions/:id', (req, res) => {
|
||||
function registerSocketEvents(socket) {
|
||||
const userId = socket.user.id;
|
||||
|
||||
// assistant:start - Session starten
|
||||
// assistant:start - Session aktivieren
|
||||
socket.on('assistant:start', (data) => {
|
||||
try {
|
||||
if (!checkSocketAccess(socket)) {
|
||||
@@ -405,8 +508,12 @@ function registerSocketEvents(socket) {
|
||||
return;
|
||||
}
|
||||
|
||||
// Beendete Session reaktivieren
|
||||
if (session.status === 'ended') {
|
||||
db.prepare(`UPDATE assistant_sessions SET status = 'active', ended_at = NULL WHERE id = ?`).run(sessionId);
|
||||
}
|
||||
|
||||
startSession(userId, sessionId, socket);
|
||||
socket.emit('assistant:status', { sessionId, status: 'running' });
|
||||
} catch (err) {
|
||||
logger.error(`[Assistant] Fehler beim Starten: ${err.message}`);
|
||||
socket.emit('assistant:status', { status: 'error', error: err.message });
|
||||
@@ -427,7 +534,7 @@ function registerSocketEvents(socket) {
|
||||
return;
|
||||
}
|
||||
|
||||
sendMessage(userId, message);
|
||||
sendMessage(userId, message, socket);
|
||||
} catch (err) {
|
||||
logger.error(`[Assistant] Fehler beim Senden: ${err.message}`);
|
||||
socket.emit('assistant:status', { status: 'error', error: err.message });
|
||||
|
||||
In neuem Issue referenzieren
Einen Benutzer sperren