airkjw/claude-mem
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Improve error handling and logging across worker services (#528)

Browse Source 
* fix: prevent memory_session_id from equaling content_session_id

The bug: memory_session_id was initialized to contentSessionId as a
"placeholder for FK purposes". This caused the SDK resume logic to
inject memory agent messages into the USER's Claude Code transcript,
corrupting their conversation history.

Root cause:
- SessionStore.createSDKSession initialized memory_session_id = contentSessionId
- SDKAgent checked memorySessionId !== contentSessionId but this check
  only worked if the session was fetched fresh from DB

The fix:
- SessionStore: Initialize memory_session_id as NULL, not contentSessionId
- SDKAgent: Simple truthy check !!session.memorySessionId (NULL = fresh start)
- Database migration: Ran UPDATE to set memory_session_id = NULL for 1807
  existing sessions that had the bug

Also adds [ALIGNMENT] logging across the session lifecycle to help debug
session continuity issues:
- Hook entry: contentSessionId + promptNumber
- DB lookup: contentSessionId → memorySessionId mapping proof
- Resume decision: shows which memorySessionId will be used for resume
- Capture: logs when memorySessionId is captured from first SDK response

UI: Added "Alignment" quick filter button in LogsModal to show only
alignment logs for debugging session continuity.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* refactor: improve error handling in worker-service.ts

- Fix GENERIC_CATCH anti-patterns by logging full error objects instead of just messages
- Add [ANTI-PATTERN IGNORED] markers for legitimate cases (cleanup, hot paths)
- Simplify error handling comments to be more concise
- Improve httpShutdown() error discrimination for ECONNREFUSED
- Reduce LARGE_TRY_BLOCK issues in initialization code

Part of anti-pattern cleanup plan (132 total issues)

* refactor: improve error logging in SearchManager.ts

- Pass full error objects to logger instead of just error.message
- Fixes PARTIAL_ERROR_LOGGING anti-patterns (10 instances)
- Better debugging visibility when Chroma queries fail

Part of anti-pattern cleanup (133 remaining)

* refactor: improve error logging across SessionStore and mcp-server

- SessionStore.ts: Fix error logging in column rename utility
- mcp-server.ts: Log full error objects instead of just error.message
- Improve error handling in Worker API calls and tool execution

Part of anti-pattern cleanup (133 remaining)

* Refactor hooks to streamline error handling and loading states

- Simplified error handling in useContextPreview by removing try-catch and directly checking response status.
- Refactored usePagination to eliminate try-catch, improving readability and maintaining error handling through response checks.
- Cleaned up useSSE by removing unnecessary try-catch around JSON parsing, ensuring clarity in message handling.
- Enhanced useSettings by streamlining the saving process, removing try-catch, and directly checking the result for success.

* refactor: add error handling back to SearchManager Chroma calls

- Wrap queryChroma calls in try-catch to prevent generator crashes
- Log Chroma errors as warnings and fall back gracefully
- Fixes generator failures when Chroma has issues
- Part of anti-pattern cleanup recovery

* feat: Add generator failure investigation report and observation duplication regression report

- Created a comprehensive investigation report detailing the root cause of generator failures during anti-pattern cleanup, including the impact, investigation process, and implemented fixes.
- Documented the critical regression causing observation duplication due to race conditions in the SDK agent, outlining symptoms, root cause analysis, and proposed fixes.

* fix: address PR #528 review comments - atomic cleanup and detector improvements

This commit addresses critical review feedback from PR #528:

## 1. Atomic Message Cleanup (Fix Race Condition)

**Problem**: SessionRoutes.ts generator error handler had race condition
- Queried messages then marked failed in loop
- If crash during loop → partial marking → inconsistent state

**Solution**:
- Added `markSessionMessagesFailed()` to PendingMessageStore.ts
- Single atomic UPDATE statement replaces loop
- Follows existing pattern from `resetProcessingToPending()`

**Files**:
- src/services/sqlite/PendingMessageStore.ts (new method)
- src/services/worker/http/routes/SessionRoutes.ts (use new method)

## 2. Anti-Pattern Detector Improvements

**Problem**: Detector didn't recognize logger.failure() method
- Lines 212 & 335 already included "failure"
- Lines 112-113 (PARTIAL_ERROR_LOGGING detection) did not

**Solution**: Updated regex patterns to include "failure" for consistency

**Files**:
- scripts/anti-pattern-test/detect-error-handling-antipatterns.ts

## 3. Documentation

**PR Comment**: Added clarification on memory_session_id fix location
- Points to SessionStore.ts:1155
- Explains why NULL initialization prevents message injection bug

## Review Response

Addresses "Must Address Before Merge" items from review:
✅ Clarified memory_session_id bug fix location (via PR comment)
✅ Made generator error handler message cleanup atomic
❌ Deferred comprehensive test suite to follow-up PR (keeps PR focused)

## Testing

- Build passes with no errors
- Anti-pattern detector runs successfully
- Atomic cleanup follows proven pattern from existing methods

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

* fix: FOREIGN KEY constraint and missing failed_at_epoch column

Two critical bugs fixed:

1. Missing failed_at_epoch column in pending_messages table
   - Added migration 20 to create the column
   - Fixes error when trying to mark messages as failed

2. FOREIGN KEY constraint failed when storing observations
   - All three agents (SDK, Gemini, OpenRouter) were passing
     session.contentSessionId instead of session.memorySessionId
   - storeObservationsAndMarkComplete expects memorySessionId
   - Added null check and clear error message

However, observations still not saving - see investigation report.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

* Refactor hook input parsing to improve error handling

- Added a nested try-catch block in new-hook.ts, save-hook.ts, and summary-hook.ts to handle JSON parsing errors more gracefully.
- Replaced direct error throwing with logging of the error details using logger.error.
- Ensured that the process exits cleanly after handling input in all three hooks.

---------

Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Alex Newman
2026-01-03 18:51:59 -05:00
committed by GitHub
parent e830157e77
commit 817b9e8f27
31 changed files with 4490 additions and 3292 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/services/worker/GeminiAgent.ts
+145 -160
View File
@@ -212,16 +212,12 @@ export class GeminiAgent {
const tokensUsed = obsResponse.tokensUsed || 0;
session.cumulativeInputTokens += Math.floor(tokensUsed * 0.7);
session.cumulativeOutputTokens += Math.floor(tokensUsed * 0.3);
await this.processGeminiResponse(session, obsResponse.content, worker, tokensUsed, originalTimestamp);
} else {
// Empty response - still mark messages as processed to avoid stuck state
logger.warn('SDK', 'Empty Gemini response for observation, marking as processed', {
sessionId: session.sessionDbId,
toolName: message.tool_name
});
await this.markMessagesProcessed(session, worker);
}
// Process response (even if empty) - empty responses will have no observations/summaries
// but messages still need to be marked complete atomically
await this.processGeminiResponse(session, obsResponse.content || '', worker, tokensUsed, originalTimestamp);
} else if (message.type === 'summarize') {
// Build summary prompt
const summaryPrompt = buildSummaryPrompt({
@@ -243,14 +239,11 @@ export class GeminiAgent {
const tokensUsed = summaryResponse.tokensUsed || 0;
session.cumulativeInputTokens += Math.floor(tokensUsed * 0.7);
session.cumulativeOutputTokens += Math.floor(tokensUsed * 0.3);
await this.processGeminiResponse(session, summaryResponse.content, worker, tokensUsed, originalTimestamp);
} else {
// Empty response - still mark messages as processed to avoid stuck state
logger.warn('SDK', 'Empty Gemini response for summary, marking as processed', {
sessionId: session.sessionDbId
});
await this.markMessagesProcessed(session, worker);
}
// Process response (even if empty) - empty responses will have no observations/summaries
// but messages still need to be marked complete atomically
await this.processGeminiResponse(session, summaryResponse.content || '', worker, tokensUsed, originalTimestamp);
}
}
@@ -374,163 +367,155 @@ export class GeminiAgent {
discoveryTokens: number,
originalTimestamp: number | null
): Promise<void> {
// Parse observations (same XML format)
// Parse observations and summary
const observations = parseObservations(text, session.contentSessionId);
// Store observations with original timestamp (if processing backlog) or current time
for (const obs of observations) {
const { id: obsId, createdAtEpoch } = this.dbManager.getSessionStore().storeObservation(
session.contentSessionId,
session.project,
obs,
session.lastPromptNumber,
discoveryTokens,
originalTimestamp ?? undefined
);
logger.info('SDK', 'Gemini observation saved', {
sessionId: session.sessionDbId,
obsId,
type: obs.type,
title: obs.title || '(untitled)'
});
// Sync to Chroma
this.dbManager.getChromaSync().syncObservation(
obsId,
session.contentSessionId,
session.project,
obs,
session.lastPromptNumber,
createdAtEpoch,
discoveryTokens
).catch(err => {
logger.warn('SDK', 'Gemini chroma sync failed', { obsId }, err);
});
// Broadcast to SSE clients
if (worker && worker.sseBroadcaster) {
worker.sseBroadcaster.broadcast({
type: 'new_observation',
observation: {
id: obsId,
memory_session_id: session.memorySessionId,
session_id: session.contentSessionId,
type: obs.type,
title: obs.title,
subtitle: obs.subtitle,
text: null,
narrative: obs.narrative || null,
facts: JSON.stringify(obs.facts || []),
concepts: JSON.stringify(obs.concepts || []),
files_read: JSON.stringify(obs.files_read || []),
files_modified: JSON.stringify(obs.files_modified || []),
project: session.project,
prompt_number: session.lastPromptNumber,
created_at_epoch: createdAtEpoch
}
});
}
}
// Parse summary
const summary = parseSummary(text, session.sessionDbId);
if (summary) {
// Convert nullable fields to empty strings for storeSummary
const summaryForStore = {
request: summary.request || '',
investigated: summary.investigated || '',
learned: summary.learned || '',
completed: summary.completed || '',
next_steps: summary.next_steps || '',
notes: summary.notes
};
// Convert nullable fields to empty strings for storeSummary (if summary exists)
const summaryForStore = summary ? {
request: summary.request || '',
investigated: summary.investigated || '',
learned: summary.learned || '',
completed: summary.completed || '',
next_steps: summary.next_steps || '',
notes: summary.notes
} : null;
const { id: summaryId, createdAtEpoch } = this.dbManager.getSessionStore().storeSummary(
session.contentSessionId,
session.project,
summaryForStore,
session.lastPromptNumber,
discoveryTokens,
originalTimestamp ?? undefined
);
logger.info('SDK', 'Gemini summary saved', {
sessionId: session.sessionDbId,
summaryId,
request: summary.request || '(no request)'
});
// Sync to Chroma
this.dbManager.getChromaSync().syncSummary(
summaryId,
session.contentSessionId,
session.project,
summaryForStore,
session.lastPromptNumber,
createdAtEpoch,
discoveryTokens
).catch(err => {
logger.warn('SDK', 'Gemini chroma sync failed', { summaryId }, err);
});
// Broadcast to SSE clients
if (worker && worker.sseBroadcaster) {
worker.sseBroadcaster.broadcast({
type: 'new_summary',
summary: {
id: summaryId,
session_id: session.contentSessionId,
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
}
});
}
// Update Cursor context file for registered projects (fire-and-forget)
updateCursorContextForProject(session.project, getWorkerPort()).catch(error => {
logger.warn('CURSOR', 'Context update failed (non-critical)', { project: session.project }, error as Error);
});
}
// Mark messages as processed
await this.markMessagesProcessed(session, worker);
}
/**
* Mark pending messages as processed
*/
private async markMessagesProcessed(session: ActiveSession, worker: any | undefined): Promise<void> {
// Get the pending message ID(s) for this response
const pendingMessageStore = this.sessionManager.getPendingMessageStore();
if (session.pendingProcessingIds.size > 0) {
for (const messageId of session.pendingProcessingIds) {
pendingMessageStore.markProcessed(messageId);
}
logger.debug('SDK', 'Gemini messages marked as processed', {
sessionId: session.sessionDbId,
count: session.pendingProcessingIds.size
});
session.pendingProcessingIds.clear();
const sessionStore = this.dbManager.getSessionStore();
if (session.pendingProcessingIds.size > 0) {
// ATOMIC TRANSACTION: Store observations + summary + mark message(s) complete
for (const messageId of session.pendingProcessingIds) {
// CRITICAL: Must use memorySessionId (not contentSessionId) for FK constraint
if (!session.memorySessionId) {
throw new Error('Cannot store observations: memorySessionId not yet captured');
}
const result = sessionStore.storeObservationsAndMarkComplete(
session.memorySessionId,
session.project,
observations,
summaryForStore,
messageId,
pendingMessageStore,
session.lastPromptNumber,
discoveryTokens,
originalTimestamp ?? undefined
);
logger.info('SDK', 'Gemini observations and summary saved atomically', {
sessionId: session.sessionDbId,
messageId,
observationCount: result.observationIds.length,
hasSummary: !!result.summaryId,
atomicTransaction: true
});
// AFTER transaction commits - async operations (can fail safely)
for (let i = 0; i < observations.length; i++) {
const obsId = result.observationIds[i];
const obs = observations[i];
this.dbManager.getChromaSync().syncObservation(
obsId,
session.contentSessionId,
session.project,
obs,
session.lastPromptNumber,
result.createdAtEpoch,
discoveryTokens
).catch(err => {
logger.warn('SDK', 'Gemini chroma sync failed', { obsId }, err);
});
// Broadcast to SSE clients
if (worker && worker.sseBroadcaster) {
worker.sseBroadcaster.broadcast({
type: 'new_observation',
observation: {
id: obsId,
memory_session_id: session.memorySessionId,
session_id: session.contentSessionId,
type: obs.type,
title: obs.title,
subtitle: obs.subtitle,
text: null,
narrative: obs.narrative || null,
facts: JSON.stringify(obs.facts || []),
concepts: JSON.stringify(obs.concepts || []),
files_read: JSON.stringify(obs.files_read || []),
files_modified: JSON.stringify(obs.files_modified || []),
project: session.project,
prompt_number: session.lastPromptNumber,
created_at_epoch: result.createdAtEpoch
}
});
}
}
// Sync summary to Chroma (if present)
if (summaryForStore && result.summaryId) {
this.dbManager.getChromaSync().syncSummary(
result.summaryId,
session.contentSessionId,
session.project,
summaryForStore,
session.lastPromptNumber,
result.createdAtEpoch,
discoveryTokens
).catch(err => {
logger.warn('SDK', 'Gemini chroma sync failed', { summaryId: result.summaryId }, err);
});
// Broadcast to SSE clients
if (worker && worker.sseBroadcaster) {
worker.sseBroadcaster.broadcast({
type: 'new_summary',
summary: {
id: result.summaryId,
session_id: session.contentSessionId,
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: result.createdAtEpoch
}
});
}
// Update Cursor context file for registered projects (fire-and-forget)
updateCursorContextForProject(session.project, getWorkerPort()).catch(error => {
logger.warn('CURSOR', 'Context update failed (non-critical)', { project: session.project }, error as Error);
});
}
}
// Clear the processed message IDs
session.pendingProcessingIds.clear();
session.earliestPendingTimestamp = null;
// Clean up old processed messages
const deletedCount = pendingMessageStore.cleanupProcessed(100);
if (deletedCount > 0) {
logger.debug('SDK', 'Gemini cleaned up old processed messages', { deletedCount });
logger.debug('SDK', 'Cleaned up old processed messages', { deletedCount });
}
// Broadcast activity status after processing
if (worker && typeof worker.broadcastProcessingStatus === 'function') {
worker.broadcastProcessingStatus();
}
}
if (worker && typeof worker.broadcastProcessingStatus === 'function') {
worker.broadcastProcessingStatus();
}
}
// REMOVED: markMessagesProcessed() - replaced by atomic transaction in processGeminiResponse()
// Messages are now marked complete atomically with observation storage to prevent duplicates
/**
* Get Gemini configuration from settings or environment
*/
src/services/worker/OpenRouterAgent.ts
+145 -160
View File
@@ -171,16 +171,12 @@ export class OpenRouterAgent {
const tokensUsed = obsResponse.tokensUsed || 0;
session.cumulativeInputTokens += Math.floor(tokensUsed * 0.7);
session.cumulativeOutputTokens += Math.floor(tokensUsed * 0.3);
await this.processOpenRouterResponse(session, obsResponse.content, worker, tokensUsed, originalTimestamp);
} else {
// Empty response - still mark messages as processed to avoid stuck state
logger.warn('SDK', 'Empty OpenRouter response for observation, marking as processed', {
sessionId: session.sessionDbId,
toolName: message.tool_name
});
await this.markMessagesProcessed(session, worker);
}
// Process response (even if empty) - empty responses will have no observations/summaries
// but messages still need to be marked complete atomically
await this.processOpenRouterResponse(session, obsResponse.content || '', worker, tokensUsed, originalTimestamp);
} else if (message.type === 'summarize') {
// Build summary prompt
const summaryPrompt = buildSummaryPrompt({
@@ -202,14 +198,11 @@ export class OpenRouterAgent {
const tokensUsed = summaryResponse.tokensUsed || 0;
session.cumulativeInputTokens += Math.floor(tokensUsed * 0.7);
session.cumulativeOutputTokens += Math.floor(tokensUsed * 0.3);
await this.processOpenRouterResponse(session, summaryResponse.content, worker, tokensUsed, originalTimestamp);
} else {
// Empty response - still mark messages as processed to avoid stuck state
logger.warn('SDK', 'Empty OpenRouter response for summary, marking as processed', {
sessionId: session.sessionDbId
});
await this.markMessagesProcessed(session, worker);
}
// Process response (even if empty) - empty responses will have no observations/summaries
// but messages still need to be marked complete atomically
await this.processOpenRouterResponse(session, summaryResponse.content || '', worker, tokensUsed, originalTimestamp);
}
}
@@ -417,163 +410,155 @@ export class OpenRouterAgent {
discoveryTokens: number,
originalTimestamp: number | null
): Promise<void> {
// Parse observations (same XML format)
// Parse observations and summary
const observations = parseObservations(text, session.contentSessionId);
// Store observations with original timestamp (if processing backlog) or current time
for (const obs of observations) {
const { id: obsId, createdAtEpoch } = this.dbManager.getSessionStore().storeObservation(
session.contentSessionId,
session.project,
obs,
session.lastPromptNumber,
discoveryTokens,
originalTimestamp ?? undefined
);
logger.info('SDK', 'OpenRouter observation saved', {
sessionId: session.sessionDbId,
obsId,
type: obs.type,
title: obs.title || '(untitled)'
});
// Sync to Chroma
this.dbManager.getChromaSync().syncObservation(
obsId,
session.contentSessionId,
session.project,
obs,
session.lastPromptNumber,
createdAtEpoch,
discoveryTokens
).catch(err => {
logger.warn('SDK', 'OpenRouter chroma sync failed', { obsId }, err);
});
// Broadcast to SSE clients
if (worker && worker.sseBroadcaster) {
worker.sseBroadcaster.broadcast({
type: 'new_observation',
observation: {
id: obsId,
memory_session_id: session.memorySessionId,
session_id: session.contentSessionId,
type: obs.type,
title: obs.title,
subtitle: obs.subtitle,
text: null,
narrative: obs.narrative || null,
facts: JSON.stringify(obs.facts || []),
concepts: JSON.stringify(obs.concepts || []),
files_read: JSON.stringify(obs.files_read || []),
files_modified: JSON.stringify(obs.files_modified || []),
project: session.project,
prompt_number: session.lastPromptNumber,
created_at_epoch: createdAtEpoch
}
});
}
}
// Parse summary
const summary = parseSummary(text, session.sessionDbId);
if (summary) {
// Convert nullable fields to empty strings for storeSummary
const summaryForStore = {
request: summary.request || '',
investigated: summary.investigated || '',
learned: summary.learned || '',
completed: summary.completed || '',
next_steps: summary.next_steps || '',
notes: summary.notes
};
// Convert nullable fields to empty strings for storeSummary (if summary exists)
const summaryForStore = summary ? {
request: summary.request || '',
investigated: summary.investigated || '',
learned: summary.learned || '',
completed: summary.completed || '',
next_steps: summary.next_steps || '',
notes: summary.notes
} : null;
const { id: summaryId, createdAtEpoch } = this.dbManager.getSessionStore().storeSummary(
session.contentSessionId,
session.project,
summaryForStore,
session.lastPromptNumber,
discoveryTokens,
originalTimestamp ?? undefined
);
logger.info('SDK', 'OpenRouter summary saved', {
sessionId: session.sessionDbId,
summaryId,
request: summary.request || '(no request)'
});
// Sync to Chroma
this.dbManager.getChromaSync().syncSummary(
summaryId,
session.contentSessionId,
session.project,
summaryForStore,
session.lastPromptNumber,
createdAtEpoch,
discoveryTokens
).catch(err => {
logger.warn('SDK', 'OpenRouter chroma sync failed', { summaryId }, err);
});
// Broadcast to SSE clients
if (worker && worker.sseBroadcaster) {
worker.sseBroadcaster.broadcast({
type: 'new_summary',
summary: {
id: summaryId,
session_id: session.contentSessionId,
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
}
});
}
// Update Cursor context file for registered projects (fire-and-forget)
updateCursorContextForProject(session.project, getWorkerPort()).catch(error => {
logger.warn('CURSOR', 'Context update failed (non-critical)', { project: session.project }, error as Error);
});
}
// Mark messages as processed
await this.markMessagesProcessed(session, worker);
}
/**
* Mark pending messages as processed
*/
private async markMessagesProcessed(session: ActiveSession, worker: any | undefined): Promise<void> {
// Get the pending message ID(s) for this response
const pendingMessageStore = this.sessionManager.getPendingMessageStore();
if (session.pendingProcessingIds.size > 0) {
for (const messageId of session.pendingProcessingIds) {
pendingMessageStore.markProcessed(messageId);
}
logger.debug('SDK', 'OpenRouter messages marked as processed', {
sessionId: session.sessionDbId,
count: session.pendingProcessingIds.size
});
session.pendingProcessingIds.clear();
const sessionStore = this.dbManager.getSessionStore();
if (session.pendingProcessingIds.size > 0) {
// ATOMIC TRANSACTION: Store observations + summary + mark message(s) complete
for (const messageId of session.pendingProcessingIds) {
// CRITICAL: Must use memorySessionId (not contentSessionId) for FK constraint
if (!session.memorySessionId) {
throw new Error('Cannot store observations: memorySessionId not yet captured');
}
const result = sessionStore.storeObservationsAndMarkComplete(
session.memorySessionId,
session.project,
observations,
summaryForStore,
messageId,
pendingMessageStore,
session.lastPromptNumber,
discoveryTokens,
originalTimestamp ?? undefined
);
logger.info('SDK', 'OpenRouter observations and summary saved atomically', {
sessionId: session.sessionDbId,
messageId,
observationCount: result.observationIds.length,
hasSummary: !!result.summaryId,
atomicTransaction: true
});
// AFTER transaction commits - async operations (can fail safely)
for (let i = 0; i < observations.length; i++) {
const obsId = result.observationIds[i];
const obs = observations[i];
this.dbManager.getChromaSync().syncObservation(
obsId,
session.contentSessionId,
session.project,
obs,
session.lastPromptNumber,
result.createdAtEpoch,
discoveryTokens
).catch(err => {
logger.warn('SDK', 'OpenRouter chroma sync failed', { obsId }, err);
});
// Broadcast to SSE clients
if (worker && worker.sseBroadcaster) {
worker.sseBroadcaster.broadcast({
type: 'new_observation',
observation: {
id: obsId,
memory_session_id: session.memorySessionId,
session_id: session.contentSessionId,
type: obs.type,
title: obs.title,
subtitle: obs.subtitle,
text: null,
narrative: obs.narrative || null,
facts: JSON.stringify(obs.facts || []),
concepts: JSON.stringify(obs.concepts || []),
files_read: JSON.stringify(obs.files_read || []),
files_modified: JSON.stringify(obs.files_modified || []),
project: session.project,
prompt_number: session.lastPromptNumber,
created_at_epoch: result.createdAtEpoch
}
});
}
}
// Sync summary to Chroma (if present)
if (summaryForStore && result.summaryId) {
this.dbManager.getChromaSync().syncSummary(
result.summaryId,
session.contentSessionId,
session.project,
summaryForStore,
session.lastPromptNumber,
result.createdAtEpoch,
discoveryTokens
).catch(err => {
logger.warn('SDK', 'OpenRouter chroma sync failed', { summaryId: result.summaryId }, err);
});
// Broadcast to SSE clients
if (worker && worker.sseBroadcaster) {
worker.sseBroadcaster.broadcast({
type: 'new_summary',
summary: {
id: result.summaryId,
session_id: session.contentSessionId,
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: result.createdAtEpoch
}
});
}
// Update Cursor context file for registered projects (fire-and-forget)
updateCursorContextForProject(session.project, getWorkerPort()).catch(error => {
logger.warn('CURSOR', 'Context update failed (non-critical)', { project: session.project }, error as Error);
});
}
}
// Clear the processed message IDs
session.pendingProcessingIds.clear();
session.earliestPendingTimestamp = null;
// Clean up old processed messages
const deletedCount = pendingMessageStore.cleanupProcessed(100);
if (deletedCount > 0) {
logger.debug('SDK', 'OpenRouter cleaned up old processed messages', { deletedCount });
logger.debug('SDK', 'Cleaned up old processed messages', { deletedCount });
}
// Broadcast activity status after processing
if (worker && typeof worker.broadcastProcessingStatus === 'function') {
worker.broadcastProcessingStatus();
}
}
if (worker && typeof worker.broadcastProcessingStatus === 'function') {
worker.broadcastProcessingStatus();
}
}
// REMOVED: markMessagesProcessed() - replaced by atomic transaction in processOpenRouterResponse()
// Messages are now marked complete atomically with observation storage to prevent duplicates
/**
* Get OpenRouter configuration from settings or environment
*/
src/services/worker/SDKAgent.ts
+175 -190
View File
@@ -69,11 +69,10 @@ export class SDKAgent {
// Create message generator (event-driven)
const messageGenerator = this.createMessageGenerator(session);
// CRITICAL: Only resume if memorySessionId is a REAL captured SDK session ID,
// not the placeholder (which equals contentSessionId). The placeholder is set
// for FK purposes but would cause the bug where we try to resume the USER's session!
const hasRealMemorySessionId = session.memorySessionId &&
session.memorySessionId !== session.contentSessionId;
// CRITICAL: Only resume if memorySessionId exists (was captured from a previous SDK response).
// memorySessionId starts as NULL and is captured on first SDK message.
// NEVER use contentSessionId for resume - that would inject messages into the user's transcript!
const hasRealMemorySessionId = !!session.memorySessionId;
logger.info('SDK', 'Starting SDK query', {
sessionDbId: session.sessionDbId,
@@ -84,13 +83,20 @@ export class SDKAgent {
lastPromptNumber: session.lastPromptNumber
});
// SESSION ALIGNMENT LOG: Resume decision proof - show if we're resuming with correct memorySessionId
if (session.lastPromptNumber > 1) {
logger.info('SDK', `[ALIGNMENT] Resume Decision | contentSessionId=${session.contentSessionId} | memorySessionId=${session.memorySessionId} | prompt#=${session.lastPromptNumber} | hasRealMemorySessionId=${hasRealMemorySessionId} | resumeWith=${hasRealMemorySessionId ? session.memorySessionId : 'NONE (fresh SDK session)'}`);
} else {
logger.info('SDK', `[ALIGNMENT] First Prompt | contentSessionId=${session.contentSessionId} | prompt#=${session.lastPromptNumber} | Will capture memorySessionId from first SDK response`);
}
// Run Agent SDK query loop
// Only resume if we have a REAL captured memory session ID (not the placeholder)
// Only resume if we have a captured memory session ID
const queryResult = query({
prompt: messageGenerator,
options: {
model: modelId,
// Only resume if memorySessionId differs from contentSessionId (meaning it was captured)
// Resume with captured memorySessionId (null on first prompt, real ID on subsequent)
...(hasRealMemorySessionId && { resume: session.memorySessionId }),
disallowedTools,
abortController: session.abortController,
@@ -113,6 +119,8 @@ export class SDKAgent {
sessionDbId: session.sessionDbId,
memorySessionId: message.session_id
});
// SESSION ALIGNMENT LOG: Memory session ID captured - now contentSessionId→memorySessionId mapping is complete
logger.info('SDK', `[ALIGNMENT] Captured | contentSessionId=${session.contentSessionId} → memorySessionId=${message.session_id} | Future prompts will resume with this ID`);
}
// Handle assistant messages
@@ -164,13 +172,11 @@ export class SDKAgent {
sessionId: session.sessionDbId,
promptNumber: session.lastPromptNumber
}, truncatedResponse);
// Parse and process response with discovery token delta and original timestamp
await this.processSDKResponse(session, textContent, worker, discoveryTokens, originalTimestamp);
} else {
// Empty response - still need to mark pending messages as processed
await this.markMessagesProcessed(session, worker);
}
// Parse and process response (even if empty) with discovery token delta and original timestamp
// Empty responses will result in empty observations array and null summary
await this.processSDKResponse(session, textContent, worker, discoveryTokens, originalTimestamp);
}
// Log result messages
@@ -316,6 +322,8 @@ export class SDKAgent {
*
* Also captures assistant responses to shared conversation history for provider interop.
* This allows Gemini to see full context if provider is switched mid-session.
*
* CRITICAL: Uses atomic transaction to prevent observation duplication on crash recovery.
*/
private async processSDKResponse(session: ActiveSession, text: string, worker: any | undefined, discoveryTokens: number, originalTimestamp: number | null): Promise<void> {
// Add assistant response to shared conversation history for provider interop
@@ -323,197 +331,174 @@ export class SDKAgent {
session.conversationHistory.push({ role: 'assistant', content: text });
}
// Parse observations
// Parse observations and summary
const observations = parseObservations(text, session.contentSessionId);
// Store observations with original timestamp (if processing backlog) or current time
for (const obs of observations) {
const { id: obsId, createdAtEpoch } = this.dbManager.getSessionStore().storeObservation(
session.contentSessionId,
session.project,
obs,
session.lastPromptNumber,
discoveryTokens,
originalTimestamp ?? undefined
);
// Log observation details
logger.info('SDK', 'Observation saved', {
sessionId: session.sessionDbId,
obsId,
type: obs.type,
title: obs.title || '(untitled)',
filesRead: obs.files_read?.length ?? 0,
filesModified: obs.files_modified?.length ?? 0,
concepts: obs.concepts?.length ?? 0
});
// Sync to Chroma
const chromaStart = Date.now();
const obsType = obs.type;
const obsTitle = obs.title || '(untitled)';
this.dbManager.getChromaSync().syncObservation(
obsId,
session.contentSessionId,
session.project,
obs,
session.lastPromptNumber,
createdAtEpoch,
discoveryTokens
).then(() => {
const chromaDuration = Date.now() - chromaStart;
logger.debug('CHROMA', 'Observation synced', {
obsId,
duration: `${chromaDuration}ms`,
type: obsType,
title: obsTitle
});
}).catch((error) => {
logger.warn('CHROMA', 'Observation sync failed, continuing without vector search', {
obsId,
type: obsType,
title: obsTitle
}, error);
});
// Broadcast to SSE clients (for web UI)
if (worker && worker.sseBroadcaster) {
worker.sseBroadcaster.broadcast({
type: 'new_observation',
observation: {
id: obsId,
memory_session_id: session.memorySessionId,
session_id: session.contentSessionId,
type: obs.type,
title: obs.title,
subtitle: obs.subtitle,
text: obs.text || null,
narrative: obs.narrative || null,
facts: JSON.stringify(obs.facts || []),
concepts: JSON.stringify(obs.concepts || []),
files_read: JSON.stringify(obs.files || []),
files_modified: JSON.stringify([]),
project: session.project,
prompt_number: session.lastPromptNumber,
created_at_epoch: createdAtEpoch
}
});
}
}
// Parse summary
const summary = parseSummary(text, session.sessionDbId);
// Store summary with original timestamp (if processing backlog) or current time
if (summary) {
const { id: summaryId, createdAtEpoch } = this.dbManager.getSessionStore().storeSummary(
session.contentSessionId,
session.project,
summary,
session.lastPromptNumber,
discoveryTokens,
originalTimestamp ?? undefined
);
// Log summary details
logger.info('SDK', 'Summary saved', {
sessionId: session.sessionDbId,
summaryId,
request: summary.request || '(no request)',
hasCompleted: !!summary.completed,
hasNextSteps: !!summary.next_steps
});
// Sync to Chroma
const chromaStart = Date.now();
const summaryRequest = summary.request || '(no request)';
this.dbManager.getChromaSync().syncSummary(
summaryId,
session.contentSessionId,
session.project,
summary,
session.lastPromptNumber,
createdAtEpoch,
discoveryTokens
).then(() => {
const chromaDuration = Date.now() - chromaStart;
logger.debug('CHROMA', 'Summary synced', {
summaryId,
duration: `${chromaDuration}ms`,
request: summaryRequest
});
}).catch((error) => {
logger.warn('CHROMA', 'Summary sync failed, continuing without vector search', {
summaryId,
request: summaryRequest
}, error);
});
// Broadcast to SSE clients (for web UI)
if (worker && worker.sseBroadcaster) {
worker.sseBroadcaster.broadcast({
type: 'new_summary',
summary: {
id: summaryId,
session_id: session.contentSessionId,
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
}
});
}
// Update Cursor context file for registered projects (fire-and-forget)
updateCursorContextForProject(session.project, getWorkerPort()).catch(error => {
logger.warn('CURSOR', 'Context update failed (non-critical)', { project: session.project }, error as Error);
});
}
// Mark messages as processed after successful observation/summary storage
await this.markMessagesProcessed(session, worker);
}
/**
* Mark all pending messages as successfully processed
* CRITICAL: Prevents message loss and duplicate processing
*/
private async markMessagesProcessed(session: ActiveSession, worker: any | undefined): Promise<void> {
// Get the pending message ID(s) for this response
// In normal operation, this should be ONE message (FIFO processing)
// But we handle multiple for safety (in case SDK batches messages)
const pendingMessageStore = this.sessionManager.getPendingMessageStore();
if (session.pendingProcessingIds.size > 0) {
for (const messageId of session.pendingProcessingIds) {
pendingMessageStore.markProcessed(messageId);
}
logger.debug('SDK', 'Messages marked as processed', {
sessionId: session.sessionDbId,
messageIds: Array.from(session.pendingProcessingIds),
count: session.pendingProcessingIds.size
});
session.pendingProcessingIds.clear();
const sessionStore = this.dbManager.getSessionStore();
// Clear timestamp for next batch (will be set fresh from next message)
if (session.pendingProcessingIds.size > 0) {
// ATOMIC TRANSACTION: Store observations + summary + mark message(s) complete
// This prevents duplicates if the worker crashes after storing but before marking complete
for (const messageId of session.pendingProcessingIds) {
// CRITICAL: Must use memorySessionId (not contentSessionId) for FK constraint
if (!session.memorySessionId) {
throw new Error('Cannot store observations: memorySessionId not yet captured');
}
const result = sessionStore.storeObservationsAndMarkComplete(
session.memorySessionId,
session.project,
observations,
summary || null,
messageId,
pendingMessageStore,
session.lastPromptNumber,
discoveryTokens,
originalTimestamp ?? undefined
);
// Log what was saved
logger.info('SDK', 'Observations and summary saved atomically', {
sessionId: session.sessionDbId,
messageId,
observationCount: result.observationIds.length,
hasSummary: !!result.summaryId,
atomicTransaction: true
});
// AFTER transaction commits - async operations (can fail safely without data loss)
// Sync observations to Chroma
for (let i = 0; i < observations.length; i++) {
const obsId = result.observationIds[i];
const obs = observations[i];
const chromaStart = Date.now();
this.dbManager.getChromaSync().syncObservation(
obsId,
session.contentSessionId,
session.project,
obs,
session.lastPromptNumber,
result.createdAtEpoch,
discoveryTokens
).then(() => {
const chromaDuration = Date.now() - chromaStart;
logger.debug('CHROMA', 'Observation synced', {
obsId,
duration: `${chromaDuration}ms`,
type: obs.type,
title: obs.title || '(untitled)'
});
}).catch((error) => {
logger.warn('CHROMA', 'Observation sync failed, continuing without vector search', {
obsId,
type: obs.type,
title: obs.title || '(untitled)'
}, error);
});
// Broadcast to SSE clients (for web UI)
if (worker && worker.sseBroadcaster) {
worker.sseBroadcaster.broadcast({
type: 'new_observation',
observation: {
id: obsId,
memory_session_id: session.memorySessionId,
session_id: session.contentSessionId,
type: obs.type,
title: obs.title,
subtitle: obs.subtitle,
text: obs.text || null,
narrative: obs.narrative || null,
facts: JSON.stringify(obs.facts || []),
concepts: JSON.stringify(obs.concepts || []),
files_read: JSON.stringify(obs.files || []),
files_modified: JSON.stringify([]),
project: session.project,
prompt_number: session.lastPromptNumber,
created_at_epoch: result.createdAtEpoch
}
});
}
}
// Sync summary to Chroma (if present)
if (summary && result.summaryId) {
const chromaStart = Date.now();
this.dbManager.getChromaSync().syncSummary(
result.summaryId,
session.contentSessionId,
session.project,
summary,
session.lastPromptNumber,
result.createdAtEpoch,
discoveryTokens
).then(() => {
const chromaDuration = Date.now() - chromaStart;
logger.debug('CHROMA', 'Summary synced', {
summaryId: result.summaryId,
duration: `${chromaDuration}ms`,
request: summary.request || '(no request)'
});
}).catch((error) => {
logger.warn('CHROMA', 'Summary sync failed, continuing without vector search', {
summaryId: result.summaryId,
request: summary.request || '(no request)'
}, error);
});
// Broadcast to SSE clients (for web UI)
if (worker && worker.sseBroadcaster) {
worker.sseBroadcaster.broadcast({
type: 'new_summary',
summary: {
id: result.summaryId,
session_id: session.contentSessionId,
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: result.createdAtEpoch
}
});
}
// Update Cursor context file for registered projects (fire-and-forget)
updateCursorContextForProject(session.project, getWorkerPort()).catch(error => {
logger.warn('CURSOR', 'Context update failed (non-critical)', { project: session.project }, error as Error);
});
}
}
// Clear the processed message IDs
session.pendingProcessingIds.clear();
session.earliestPendingTimestamp = null;
// Clean up old processed messages (keep last 100 for UI display)
const deletedCount = pendingMessageStore.cleanupProcessed(100);
if (deletedCount > 0) {
logger.debug('SDK', 'Cleaned up old processed messages', {
deletedCount
});
logger.debug('SDK', 'Cleaned up old processed messages', { deletedCount });
}
// Broadcast activity status after processing (queue may have changed)
if (worker && typeof worker.broadcastProcessingStatus === 'function') {
worker.broadcastProcessingStatus();
}
}
// Broadcast activity status after processing (queue may have changed)
if (worker && typeof worker.broadcastProcessingStatus === 'function') {
worker.broadcastProcessingStatus();
}
}
// REMOVED: markMessagesProcessed() - replaced by atomic transaction in processSDKResponse()
// Messages are now marked complete atomically with observation storage to prevent duplicates
// ============================================================================
// Configuration Helpers
// ============================================================================
src/services/worker/SearchManager.ts
+1471 -1649
View File
File diff suppressed because it is too large Load Diff
src/services/worker/http/routes/SessionRoutes.ts
+19 -12
View File
@@ -147,23 +147,18 @@ export class SessionRoutes extends BaseRouteHandler {
// Mark all processing messages as failed so they can be retried or abandoned
const pendingStore = this.sessionManager.getPendingMessageStore();
const db = this.dbManager.getSessionStore().db;
try {
const stmt = db.prepare(`
SELECT id FROM pending_messages
WHERE session_db_id = ? AND status = 'processing'
`);
const processingMessages = stmt.all(session.sessionDbId) as { id: number }[];
for (const msg of processingMessages) {
pendingStore.markFailed(msg.id);
logger.warn('SESSION', `Marked message as failed after generator error`, {
const failedCount = pendingStore.markSessionMessagesFailed(session.sessionDbId);
if (failedCount > 0) {
logger.warn('SESSION', `Marked messages as failed after generator error`, {
sessionId: session.sessionDbId,
messageId: msg.id
failedCount
});
}
} catch (dbError) {
logger.error('SESSION', 'Failed to mark messages as failed', { sessionId: session.sessionDbId }, dbError as Error);
logger.error('SESSION', 'Failed to mark messages as failed', {
sessionId: session.sessionDbId
}, dbError as Error);
}
})
.finally(() => {
@@ -570,6 +565,11 @@ export class SessionRoutes extends BaseRouteHandler {
contentSessionId
});
// SESSION ALIGNMENT LOG: DB lookup proof - show content→memory mapping
const dbSession = store.getSessionById(sessionDbId);
const memorySessionId = dbSession?.memory_session_id || null;
const hasCapturedMemoryId = !!memorySessionId;
// Step 2: Get next prompt number from user_prompts count
const currentCount = store.getPromptNumberFromUserPrompts(contentSessionId);
const promptNumber = currentCount + 1;
@@ -580,6 +580,13 @@ export class SessionRoutes extends BaseRouteHandler {
currentCount
});
// SESSION ALIGNMENT LOG: For prompt > 1, prove we looked up memorySessionId from contentSessionId
if (promptNumber > 1) {
logger.info('HTTP', `[ALIGNMENT] DB Lookup Proof | contentSessionId=${contentSessionId} → memorySessionId=${memorySessionId || '(not yet captured)'} | prompt#=${promptNumber} | hasCapturedMemoryId=${hasCapturedMemoryId}`);
} else {
logger.info('HTTP', `[ALIGNMENT] New Session | contentSessionId=${contentSessionId} | prompt#=${promptNumber} | memorySessionId will be captured on first SDK response`);
}
// Step 3: Strip privacy tags from prompt
const cleanedPrompt = stripMemoryTagsFromPrompt(prompt);