feat: add context sharing and storage consistency (P0)

Session runtime mutex:
- Add session_runtime_locks table with TTL-based locking via SQLite
  single-writer guarantee (transaction: DELETE expired + INSERT, PK
  conflict = already locked → 409 SESSION_BUSY)
- POST /api/chat acquires lock before processing, releases in
  collectStreamResponse finally block and error catch
- Add runtime_status/runtime_updated_at/runtime_error columns to
  chat_sessions for tracking session state

Permission request persistence:
- Add permission_requests table with full lifecycle tracking
  (pending → allow/deny/timeout/aborted)
- Dual-write in permission-registry: in-memory Map for real-time
  resolution + DB for audit and recovery
- claude-client canUseTool now persists to DB before SSE event
- permission/route.ts validates against DB before resolving:
  404 (not found), 409 (already resolved), 410 (waiter gone)

Immutable sdk_cwd:
- Add sdk_cwd column, snapshot from working_directory at session
  creation time, never changes afterward
- route.ts uses sdk_cwd for SDK cwd to prevent resume breakage
  when user modifies working_directory
- Backfill existing sessions via migration

Startup recovery (migrateDb):
- Reset running/waiting_permission sessions to idle with error msg
- Clear all runtime locks (stale from previous process)
- Abort pending permission requests (no in-memory waiter exists)

All DB writes are try/catch guarded — failure degrades gracefully
without breaking the in-memory hot path.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
7418andClaude Opus 4.6 committed 2026-02-27 12:20:01 +08:00
1 parent 09908cfcb7
commit 2a7dbd45c2
6 files changed
+364 -12

No files matched your search

+21 -2
View File
@@ -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' } },
);
}
+38 -5
View File
@@ -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<string>, sessionId: string) {
async function collectStreamResponse(stream: ReadableStream<string>, sessionId: string, onComplete?: () => void) {
const reader = stream.getReader();
const contentBlocks: MessageContentBlock[] = [];
let currentText = '';
@@ -332,5 +363,7 @@ async function collectStreamResponse(stream: ReadableStream<string>, sessionId:
addMessage(sessionId, 'assistant', content);
}
}
} finally {
onComplete?.();
}
}
+27 -2
View File
@@ -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<strin
files,
toolTimeoutSeconds = 0,
conversationHistory,
onRuntimeStatusChange,
} = options;
return new ReadableStream<string>({
@@ -465,15 +466,39 @@ export function streamClaude(options: ClaudeStreamOptions): ReadableStream<strin
description: undefined,
};
// Persist permission request to DB for audit/recovery
const expiresAt = new Date(Date.now() + 5 * 60 * 1000).toISOString().replace('T', ' ').split('.')[0];
try {
createPermissionRequest({
id: permissionRequestId,
sessionId,
sdkSessionId: sdkSessionId || '',
toolName,
toolInput: JSON.stringify(input),
decisionReason: opts.decisionReason || '',
expiresAt,
});
} catch (e) {
console.warn('[claude-client] Failed to persist permission request to DB:', e);
}
// Send permission_request SSE event to the client
controller.enqueue(formatSSE({
type: 'permission_request',
data: JSON.stringify(permEvent),
}));
// Notify runtime status change
onRuntimeStatusChange?.('waiting_permission');
// Wait for user response (resolved by POST /api/chat/permission)
// Store original input so registry can inject updatedInput on allow
return registerPendingPermission(permissionRequestId, input, opts.signal);
const result = await registerPendingPermission(permissionRequestId, input, opts.signal);
// Restore runtime status after permission resolved
onRuntimeStatusChange?.('running');
return result;
};
// Hooks: capture notifications and tool completion events
+248 -2
View File
@@ -243,6 +243,21 @@ function migrateDb(db: Database.Database): void {
if (!colNames.includes('provider_id')) {
db.exec("ALTER TABLE chat_sessions ADD COLUMN provider_id TEXT NOT NULL DEFAULT ''");
}
if (!colNames.includes('sdk_cwd')) {
db.exec("ALTER TABLE chat_sessions ADD COLUMN sdk_cwd TEXT NOT NULL DEFAULT ''");
// Backfill sdk_cwd from working_directory for existing sessions
db.exec("UPDATE chat_sessions SET sdk_cwd = working_directory WHERE sdk_cwd = '' AND working_directory != ''");
}
if (!colNames.includes('runtime_status')) {
db.exec("ALTER TABLE chat_sessions ADD COLUMN runtime_status TEXT NOT NULL DEFAULT 'idle'");
}
if (!colNames.includes('runtime_updated_at')) {
db.exec("ALTER TABLE chat_sessions ADD COLUMN runtime_updated_at TEXT NOT NULL DEFAULT ''");
}
if (!colNames.includes('runtime_error')) {
db.exec("ALTER TABLE chat_sessions ADD COLUMN runtime_error TEXT NOT NULL DEFAULT ''");
}
db.exec("CREATE INDEX IF NOT EXISTS idx_sessions_runtime_status ON chat_sessions(runtime_status)");
// Migrate is_active provider to default_provider_id setting
const defaultProviderSetting = db.prepare("SELECT value FROM settings WHERE key = 'default_provider_id'").get() as { value: string } | undefined;
@@ -406,6 +421,59 @@ function migrateDb(db: Database.Database): void {
WHERE status = 'processing'
`);
// Create session_runtime_locks table
db.exec(`
CREATE TABLE IF NOT EXISTS session_runtime_locks (
session_id TEXT PRIMARY KEY,
lock_id TEXT NOT NULL,
owner TEXT NOT NULL,
expires_at TEXT NOT NULL,
created_at TEXT NOT NULL DEFAULT (datetime('now')),
updated_at TEXT NOT NULL DEFAULT (datetime('now')),
FOREIGN KEY (session_id) REFERENCES chat_sessions(id) ON DELETE CASCADE
);
CREATE INDEX IF NOT EXISTS idx_runtime_locks_expires_at ON session_runtime_locks(expires_at);
`);
// Create permission_requests table
db.exec(`
CREATE TABLE IF NOT EXISTS permission_requests (
id TEXT PRIMARY KEY,
session_id TEXT NOT NULL,
sdk_session_id TEXT NOT NULL DEFAULT '',
tool_name TEXT NOT NULL,
tool_input TEXT NOT NULL,
decision_reason TEXT NOT NULL DEFAULT '',
status TEXT NOT NULL CHECK(status IN ('pending','allow','deny','timeout','aborted')),
updated_permissions TEXT NOT NULL DEFAULT '[]',
updated_input TEXT,
message TEXT NOT NULL DEFAULT '',
created_at TEXT NOT NULL DEFAULT (datetime('now')),
expires_at TEXT NOT NULL,
resolved_at TEXT,
FOREIGN KEY (session_id) REFERENCES chat_sessions(id) ON DELETE CASCADE
);
CREATE INDEX IF NOT EXISTS idx_permission_session_status ON permission_requests(session_id, status);
CREATE INDEX IF NOT EXISTS idx_permission_expires_at ON permission_requests(expires_at);
`);
// Startup recovery: reset stale runtime states from previous process
db.exec(`
UPDATE chat_sessions
SET runtime_status = 'idle',
runtime_error = 'Process restarted',
runtime_updated_at = datetime('now')
WHERE runtime_status IN ('running', 'waiting_permission')
`);
db.exec("DELETE FROM session_runtime_locks");
db.exec(`
UPDATE permission_requests
SET status = 'aborted',
resolved_at = datetime('now'),
message = 'Process restarted'
WHERE status = 'pending'
`);
// Migrate existing settings to a default provider if api_providers is empty
const providerCount = db.prepare('SELECT COUNT(*) as count FROM api_providers').get() as { count: number };
if (providerCount.count === 0) {
@@ -449,8 +517,8 @@ export function createSession(
const projectName = path.basename(wd);
db.prepare(
'INSERT INTO chat_sessions (id, title, created_at, updated_at, model, system_prompt, working_directory, sdk_session_id, project_name, status, mode) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)'
).run(id, title || 'New Chat', now, now, model || '', systemPrompt || '', wd, '', projectName, 'active', mode || 'code');
'INSERT INTO chat_sessions (id, title, created_at, updated_at, model, system_prompt, working_directory, sdk_session_id, project_name, status, mode, sdk_cwd) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)'
).run(id, title || 'New Chat', now, now, model || '', systemPrompt || '', wd, '', projectName, 'active', mode || 'code', wd);
return getSession(id)!;
}
@@ -1078,6 +1146,184 @@ export function markContextEventSynced(id: string): void {
db.prepare('UPDATE media_context_events SET synced_at = ? WHERE id = ?').run(now, id);
}
// ==========================================
// Session Runtime Lock Operations
// ==========================================
/**
* Acquire an exclusive lock for a session.
* Uses SQLite's single-writer guarantee: within a transaction, delete expired
* locks then INSERT. PK conflict = already locked → return false.
*/
export function acquireSessionLock(
sessionId: string,
lockId: string,
owner: 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 txn = db.transaction(() => {
// 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<string, unknown>;
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<typeof getPermissionRequest>;
}
// ==========================================
// Graceful Shutdown
// ==========================================
+25 -1
View File
@@ -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<string, PendingPermission> {
/**
* 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<string, unknown>) : 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;
+5
View File
@@ -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;
}