diff --git a/src/app/api/chat/permission/route.ts b/src/app/api/chat/permission/route.ts index 5382b4d..e7e6427 100644 --- a/src/app/api/chat/permission/route.ts +++ b/src/app/api/chat/permission/route.ts @@ -1,5 +1,6 @@ import { NextRequest } from 'next/server'; import { resolvePendingPermission } from '@/lib/permission-registry'; +import { getPermissionRequest } from '@/lib/db'; import type { PermissionResponseRequest } from '@/types'; import type { PermissionResult, PermissionUpdate } from '@anthropic-ai/claude-agent-sdk'; @@ -18,6 +19,21 @@ export async function POST(request: NextRequest) { ); } + // Validate against DB before resolving in-memory + const dbRecord = getPermissionRequest(permissionRequestId); + if (!dbRecord) { + return new Response( + JSON.stringify({ error: 'Permission request not found', code: 'NOT_FOUND' }), + { status: 404, headers: { 'Content-Type': 'application/json' } }, + ); + } + if (dbRecord.status !== 'pending') { + return new Response( + JSON.stringify({ error: `Permission request already resolved (status: ${dbRecord.status})`, code: 'ALREADY_RESOLVED' }), + { status: 409, headers: { 'Content-Type': 'application/json' } }, + ); + } + let result: PermissionResult; if (decision.behavior === 'allow') { result = { @@ -36,8 +52,11 @@ export async function POST(request: NextRequest) { if (!found) { return new Response( - JSON.stringify({ error: 'Permission request not found or already resolved' }), - { status: 404, headers: { 'Content-Type': 'application/json' } }, + JSON.stringify({ + error: 'Permission request exists in DB but the in-memory waiter is gone (process may have restarted)', + code: 'WAITER_GONE', + }), + { status: 410, headers: { 'Content-Type': 'application/json' } }, ); } diff --git a/src/app/api/chat/route.ts b/src/app/api/chat/route.ts index 76a5650..8b4ea43 100644 --- a/src/app/api/chat/route.ts +++ b/src/app/api/chat/route.ts @@ -1,7 +1,8 @@ import { NextRequest } from 'next/server'; import { streamClaude } from '@/lib/claude-client'; -import { addMessage, getMessages, getSession, updateSessionTitle, updateSdkSessionId, updateSessionModel, updateSessionProvider, updateSessionProviderId, getSetting, getProvider, getDefaultProviderId } from '@/lib/db'; +import { addMessage, getMessages, getSession, updateSessionTitle, updateSdkSessionId, updateSessionModel, updateSessionProvider, updateSessionProviderId, getSetting, getProvider, getDefaultProviderId, acquireSessionLock, releaseSessionLock, setSessionRuntimeStatus } from '@/lib/db'; import type { SendMessageRequest, SSEEvent, TokenUsage, MessageContentBlock, FileAttachment } from '@/types'; +import crypto from 'crypto'; import fs from 'fs'; import path from 'path'; @@ -9,6 +10,9 @@ export const runtime = 'nodejs'; export const dynamic = 'force-dynamic'; export async function POST(request: NextRequest) { + let activeSessionId: string | undefined; + let activeLockId: string | undefined; + try { const body: SendMessageRequest & { files?: FileAttachment[]; toolTimeout?: number; provider_id?: string; systemPromptAppend?: string } = await request.json(); const { session_id, content, model, mode, files, toolTimeout, provider_id, systemPromptAppend } = body; @@ -31,6 +35,19 @@ export async function POST(request: NextRequest) { }); } + // Acquire exclusive lock for this session to prevent concurrent requests + const lockId = crypto.randomBytes(8).toString('hex'); + const lockAcquired = acquireSessionLock(session_id, lockId, `chat-${process.pid}`, 600); + if (!lockAcquired) { + return new Response( + JSON.stringify({ error: 'Session is busy processing another request', code: 'SESSION_BUSY' }), + { status: 409, headers: { 'Content-Type': 'application/json' } }, + ); + } + activeSessionId = session_id; + activeLockId = lockId; + setSessionRuntimeStatus(session_id, 'running'); + // Save user message — persist file metadata so attachments survive page reload let savedContent = content; let fileMeta: Array<{ id: string; name: string; type: string; size: number; filePath: string }> | undefined; @@ -168,20 +185,26 @@ export async function POST(request: NextRequest) { sdkSessionId: session.sdk_session_id || undefined, model: effectiveModel, systemPrompt: finalSystemPrompt, - workingDirectory: session.working_directory || undefined, + workingDirectory: session.sdk_cwd || session.working_directory || undefined, abortController, permissionMode, files: fileAttachments, toolTimeoutSeconds: toolTimeout || 300, provider: resolvedProvider, conversationHistory: historyMsgs, + onRuntimeStatusChange: (status: string) => { + try { setSessionRuntimeStatus(session_id, status); } catch { /* best effort */ } + }, }); // Tee the stream: one for client, one for collecting the response const [streamForClient, streamForCollect] = stream.tee(); - // Save assistant message in background - collectStreamResponse(streamForCollect, session_id); + // Save assistant message in background, with cleanup callback to release lock + collectStreamResponse(streamForCollect, session_id, () => { + releaseSessionLock(session_id, lockId); + setSessionRuntimeStatus(session_id, 'idle'); + }); return new Response(streamForClient, { headers: { @@ -191,6 +214,14 @@ export async function POST(request: NextRequest) { }, }); } catch (error) { + // Release lock and reset status on error (only if lock was acquired) + if (activeSessionId && activeLockId) { + try { + releaseSessionLock(activeSessionId, activeLockId); + setSessionRuntimeStatus(activeSessionId, 'idle', error instanceof Error ? error.message : 'Unknown error'); + } catch { /* best effort */ } + } + const message = error instanceof Error ? error.message : 'Internal server error'; return new Response(JSON.stringify({ error: message }), { status: 500, @@ -199,7 +230,7 @@ export async function POST(request: NextRequest) { } } -async function collectStreamResponse(stream: ReadableStream, sessionId: string) { +async function collectStreamResponse(stream: ReadableStream, sessionId: string, onComplete?: () => void) { const reader = stream.getReader(); const contentBlocks: MessageContentBlock[] = []; let currentText = ''; @@ -332,5 +363,7 @@ async function collectStreamResponse(stream: ReadableStream, sessionId: addMessage(sessionId, 'assistant', content); } } + } finally { + onComplete?.(); } } diff --git a/src/lib/claude-client.ts b/src/lib/claude-client.ts index 19048e2..988e716 100644 --- a/src/lib/claude-client.ts +++ b/src/lib/claude-client.ts @@ -19,7 +19,7 @@ import type { ClaudeStreamOptions, SSEEvent, TokenUsage, MCPServerConfig, Permis import { isImageFile } from '@/types'; import { registerPendingPermission } from './permission-registry'; import { registerConversation, unregisterConversation } from './conversation-registry'; -import { getSetting, getActiveProvider, updateSdkSessionId } from './db'; +import { getSetting, getActiveProvider, updateSdkSessionId, createPermissionRequest } from './db'; import { findClaudeBinary, findGitBash, getExpandedPath } from './platform'; import os from 'os'; import fs from 'fs'; @@ -278,6 +278,7 @@ export function streamClaude(options: ClaudeStreamOptions): ReadableStream({ @@ -465,15 +466,39 @@ export function streamClaude(options: ClaudeStreamOptions): ReadableStream { + // Delete expired locks first + db.prepare("DELETE FROM session_runtime_locks WHERE expires_at < ?").run(now); + // Try to insert — PK conflict means session is already locked + try { + db.prepare( + 'INSERT INTO session_runtime_locks (session_id, lock_id, owner, expires_at, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?)' + ).run(sessionId, lockId, owner, expiresAt, now, now); + return true; + } catch { + return false; + } + }); + + return txn(); +} + +/** + * Renew an existing session lock by extending its expiry. + */ +export function renewSessionLock( + sessionId: string, + lockId: string, + ttlSec: number = 300, +): boolean { + const db = getDb(); + const now = new Date().toISOString().replace('T', ' ').split('.')[0]; + const expiresAt = new Date(Date.now() + ttlSec * 1000).toISOString().replace('T', ' ').split('.')[0]; + + const result = db.prepare( + 'UPDATE session_runtime_locks SET expires_at = ?, updated_at = ? WHERE session_id = ? AND lock_id = ?' + ).run(expiresAt, now, sessionId, lockId); + + return result.changes > 0; +} + +/** + * Release a session lock. + */ +export function releaseSessionLock(sessionId: string, lockId: string): boolean { + const db = getDb(); + const result = db.prepare( + 'DELETE FROM session_runtime_locks WHERE session_id = ? AND lock_id = ?' + ).run(sessionId, lockId); + return result.changes > 0; +} + +/** + * Update the runtime status of a session. + */ +export function setSessionRuntimeStatus( + sessionId: string, + status: string, + error?: string, +): void { + const db = getDb(); + const now = new Date().toISOString().replace('T', ' ').split('.')[0]; + db.prepare( + 'UPDATE chat_sessions SET runtime_status = ?, runtime_updated_at = ?, runtime_error = ? WHERE id = ?' + ).run(status, now, error || '', sessionId); +} + +// ========================================== +// Permission Request Operations +// ========================================== + +/** + * Create a pending permission request record in DB. + */ +export function createPermissionRequest(params: { + id: string; + sessionId: string; + sdkSessionId?: string; + toolName: string; + toolInput: string; + decisionReason?: string; + expiresAt: string; +}): void { + const db = getDb(); + db.prepare( + `INSERT INTO permission_requests (id, session_id, sdk_session_id, tool_name, tool_input, decision_reason, status, expires_at) + VALUES (?, ?, ?, ?, ?, ?, 'pending', ?)` + ).run( + params.id, + params.sessionId, + params.sdkSessionId || '', + params.toolName, + params.toolInput, + params.decisionReason || '', + params.expiresAt, + ); +} + +/** + * Resolve a pending permission request. Only updates if status is still 'pending'. + * Returns true if the request was found and resolved, false otherwise. + */ +export function resolvePermissionRequest( + id: string, + status: 'allow' | 'deny' | 'timeout' | 'aborted', + opts?: { + updatedPermissions?: unknown[]; + updatedInput?: Record; + message?: string; + }, +): boolean { + const db = getDb(); + const now = new Date().toISOString().replace('T', ' ').split('.')[0]; + const result = db.prepare( + `UPDATE permission_requests + SET status = ?, resolved_at = ?, updated_permissions = ?, updated_input = ?, message = ? + WHERE id = ? AND status = 'pending'` + ).run( + status, + now, + JSON.stringify(opts?.updatedPermissions || []), + opts?.updatedInput ? JSON.stringify(opts.updatedInput) : null, + opts?.message || '', + id, + ); + return result.changes > 0; +} + +/** + * Expire all pending permission requests that have passed their expiry time. + */ +export function expirePermissionRequests(now?: string): number { + const db = getDb(); + const cutoff = now || new Date().toISOString().replace('T', ' ').split('.')[0]; + const result = db.prepare( + `UPDATE permission_requests + SET status = 'timeout', resolved_at = ?, message = 'Expired' + WHERE status = 'pending' AND expires_at < ?` + ).run(cutoff, cutoff); + return result.changes; +} + +/** + * Get a permission request by ID. + */ +export function getPermissionRequest(id: string): { + id: string; + session_id: string; + sdk_session_id: string; + tool_name: string; + tool_input: string; + decision_reason: string; + status: string; + updated_permissions: string; + updated_input: string | null; + message: string; + created_at: string; + expires_at: string; + resolved_at: string | null; +} | undefined { + const db = getDb(); + return db.prepare('SELECT * FROM permission_requests WHERE id = ?').get(id) as ReturnType; +} + // ========================================== // Graceful Shutdown // ========================================== diff --git a/src/lib/permission-registry.ts b/src/lib/permission-registry.ts index 96d3d30..88cb4ea 100644 --- a/src/lib/permission-registry.ts +++ b/src/lib/permission-registry.ts @@ -1,4 +1,5 @@ import type { PermissionResult } from '@anthropic-ai/claude-agent-sdk'; +import { resolvePermissionRequest as dbResolvePermission } from './db'; interface PendingPermission { resolve: (result: PermissionResult) => void; @@ -24,14 +25,20 @@ function getMap(): Map { /** * Helper to deny and remove a pending permission entry. + * Also writes the denial to DB for persistence/audit. */ -function denyAndRemove(id: string, message: string) { +function denyAndRemove(id: string, message: string, dbStatus: 'timeout' | 'aborted' = 'aborted') { const map = getMap(); const entry = map.get(id); if (!entry) return; clearTimeout(entry.timer); entry.resolve({ behavior: 'deny', message }); map.delete(id); + try { + dbResolvePermission(id, dbStatus, { message }); + } catch { + // DB write failure should not affect in-memory path + } } /** @@ -52,6 +59,11 @@ export function registerPendingPermission( console.warn(`[permission-registry] Permission request ${id} timed out after ${TIMEOUT_MS / 1000}s`); resolve({ behavior: 'deny', message: 'Permission request timed out' }); map.delete(id); + try { + dbResolvePermission(id, 'timeout', { message: 'Permission request timed out' }); + } catch { + // DB write failure should not affect in-memory path + } } }, TIMEOUT_MS); @@ -88,6 +100,18 @@ export function resolvePendingPermission( result = { ...result, updatedInput: entry.toolInput }; } + // Dual-write: persist to DB before resolving in-memory + try { + const dbStatus = result.behavior === 'allow' ? 'allow' as const : 'deny' as const; + dbResolvePermission(id, dbStatus, { + updatedPermissions: result.behavior === 'allow' ? (result.updatedPermissions as unknown[]) : undefined, + updatedInput: result.behavior === 'allow' ? (result.updatedInput as Record) : undefined, + message: result.behavior === 'deny' ? result.message : undefined, + }); + } catch { + // DB write failure should not affect in-memory path + } + entry.resolve(result); map.delete(id); return true; diff --git a/src/types/index.ts b/src/types/index.ts index d7e3573..577ce41 100644 --- a/src/types/index.ts +++ b/src/types/index.ts @@ -17,6 +17,10 @@ export interface ChatSession { needs_approval?: boolean; provider_name: string; provider_id: string; + sdk_cwd: string; + runtime_status: string; + runtime_updated_at: string; + runtime_error: string; } // ========================================== @@ -599,4 +603,5 @@ export interface ClaudeStreamOptions { provider?: ApiProvider; /** Recent conversation history from DB — used as fallback context when SDK resume is unavailable or fails */ conversationHistory?: Array<{ role: 'user' | 'assistant'; content: string }>; + onRuntimeStatusChange?: (status: string) => void; }