fix: resolve message processing failures in multi-session scenarios

[hahaschool]

Summary

• Fix AbortController state issue causing generators to abort within ~70ms
• Fix FOREIGN KEY constraint violation when worker restarts with existing observations
• Fix deprecated queueDepth logging showing 0 instead of actual count

Problem

When multiple Claude Code sessions were running in the same project folder, messages would accumulate without being processed. Root causes:

1. AbortController reuse: When a generator finished naturally, it aborted its controller but kept the session in memory. New messages would try to start a generator with the already-aborted controller, causing immediate exit.

2. FK constraint violation: After worker restart, SDK generates new session ID. Updating sdk_sessions.memory_session_id failed because existing observations referenced the old ID (FK without ON UPDATE CASCADE).

Solution

1. Check and replace aborted AbortController before starting generators
2. Use database's existing memory_session_id for observation storage (FK integrity)
3. Only update memory_session_id in database when it's NULL

Test plan

• Verified messages flow correctly: ENQUEUED → CLAIMED → STORED
• No FK constraint errors in logs
• Pending queue stays empty (no accumulation)
• New observations stored successfully after fix

🤖 Generated with Claude Code

[greptile-apps]

Greptile Overview

Greptile Summary

This PR fixes three critical issues affecting message processing in multi-session scenarios:

• AbortController reuse bug: When generators completed naturally, they aborted their controller but kept sessions in memory. New messages would try to start generators with already-aborted controllers, causing immediate exit within ~70ms. Fixed by checking and replacing aborted controllers before starting generators.

• FK constraint violation on worker restart: After worker restart, SDK generates new session ID. Updating sdk_sessions.memory_session_id failed because existing observations referenced the old ID (FK constraint lacks ON UPDATE CASCADE). Fixed by implementing dual-tracking: in-memory value tracks current SDK session for resume, while database value maintains FK integrity with existing observations.

• Stale resume crash recovery: Added detection of stale resume failures ("aborted by user") with automatic recovery that clears memorySessionId, sets forceInit flag, and retries with fresh session.

The implementation correctly handles the tension between resume functionality (needs current SDK session_id) and data integrity (needs stable memory_session_id for existing observations). The forceInit flag provides one-time override for stale resume recovery.

Confidence Score: 4/5

• Safe to merge with minor type safety concern that should be addressed
• The PR addresses real bugs with well-reasoned solutions. The AbortController fix is straightforward and correct. The FK constraint handling is sophisticated and correct, using dual-tracking to balance resume functionality with data integrity. The crash recovery logic is comprehensive. One syntax issue exists: updateMemorySessionId is called with null using type assertion as any, which bypasses type safety.
• src/services/worker-service.ts line 391 has type safety issue with null as any

Important Files Changed

Filename	Overview
src/services/worker-service.ts	Adds AbortController replacement logic and comprehensive crash recovery with stale resume detection
src/services/worker-types.ts	Adds forceInit flag to ActiveSession interface for stale resume recovery
src/services/worker/SDKAgent.ts	Implements FK-safe memory_session_id handling and forceInit support for crash recovery
src/services/worker/agents/ResponseProcessor.ts	Uses database-authoritative memory_session_id to prevent FK constraint violations during storage
src/services/worker/http/routes/SessionRoutes.ts	Adds AbortController replacement and fixes deprecated queueDepth logging

Sequence Diagram

sequenceDiagram
    participant User as User Session
    participant WS as WorkerService
    participant SM as SessionManager
    participant SDK as SDKAgent
    participant RP as ResponseProcessor
    participant DB as Database

    Note over User,DB: Multi-Session Message Processing Flow

    User->>WS: Message arrives
    WS->>SM: getSession(sessionDbId)
    SM->>DB: Load session (if not in memory)
    DB-->>SM: session data + memory_session_id
    
    Note over WS: Check AbortController state
    alt AbortController is aborted
        WS->>WS: Create new AbortController
        Note right of WS: Fix #1: Replace aborted controller<br/>to prevent immediate exit
    end

    WS->>SDK: startSession(session)
    
    Note over SDK: Resume decision logic
    alt forceInit flag set
        SDK->>SDK: Skip resume (stale session recovery)
        SDK->>SDK: Clear forceInit flag
    else memorySessionId exists AND lastPromptNumber > 1
        SDK->>SDK: Resume with memorySessionId
    else
        SDK->>SDK: Start fresh session
    end

    SDK->>DB: Check existing memory_session_id
    DB-->>SDK: dbSession.memory_session_id
    
    alt database has memory_session_id
        Note right of SDK: Fix #2: Use DB value for FK integrity
        SDK->>SDK: Keep DB value, don't update
    else database has NULL
        SDK->>SDK: Capture new session_id from SDK
        SDK->>DB: updateMemorySessionId(new_id)
    end

    SDK->>RP: Process assistant response
    RP->>DB: Get dbSession.memory_session_id
    Note right of RP: Fix #2: Use authoritative DB value
    RP->>DB: storeObservations(memorySessionIdForStorage)
    DB-->>RP: Success (no FK violation)

    SDK-->>WS: Generator completes

    alt Generator crashed (not aborted)
        WS->>SM: Check pending messages
        SM-->>WS: pendingCount > 0
        WS->>WS: Replace AbortController
        WS->>WS: Abort old controller (cleanup)
        Note right of WS: Crash recovery with delay
        WS-->>WS: setTimeout restart
    else Generator naturally completed
        WS->>WS: Abort controller (cleanup)
    end

    Note over WS: Stale Resume Error Handling
    alt SDK error: "aborted by user"
        WS->>WS: Set forceInit flag
        WS->>DB: Clear memory_session_id
        WS->>WS: Retry with fresh session
    end

Loading

[greptile-apps]

_{5 files reviewed, 1 comment}

_{Edit Code Review Agent Settings | Greptile}

[hahaschool]

Thanks for the thorough review, @greptile-apps!

I've addressed the type safety concern:

Fixed in commit 76227ec:

• Updated updateMemorySessionId function signature to accept string | null instead of just string
• Removed the as any type assertion in worker-service.ts line 391
• Updated both SessionStore.ts and sessions/create.ts for consistency

The function now properly supports clearing memory_session_id during stale resume recovery without bypassing type safety.

[greptile-apps]

Perfect! The type safety improvements look excellent. You've properly addressed the concern by:

1. Updated function signatures to accept string | null instead of just string in both SessionStore.ts and sessions/create.ts
2. Removed the as any type assertion from worker-service.ts line 391, which was the main type safety issue
3. Maintained consistency across all related files

This is a much cleaner solution that properly expresses the intent of the code - that memory_session_id can legitimately be cleared (set to null) during stale resume recovery scenarios. The type system now accurately reflects the runtime behavior without any unsafe casting.

The changes are minimal and focused, which is ideal for this type of type safety fix. Great work addressing this feedback!

[thedotmack]

Thank you for this thorough fix, @hahaschool! You identified some real issues in the multi-session message processing flow. However, this PR now has significant overlap with several PRs that were merged after yours was opened, and the rebase produces conflicts in 3 files (worker-service.ts, worker-types.ts, SDKAgent.ts).

Already addressed by recently merged PRs:

• Crash recovery with restart limits → PR fix: prevent infinite restart loop that causes runaway API costs #693 (merged): Adds consecutiveRestarts counter, limits to 3 restarts with exponential backoff in SessionRoutes.ts
• Orphaned message handling / session termination detection → PR fix(worker): gracefully process orphaned pending messages after session termination #937 (merged): Adds isSessionTerminatedError(), runFallbackForTerminatedSession() with Gemini/OpenRouter fallback, and markAllSessionMessagesAbandoned() in worker-service.ts
• AbortController replacement during crash recovery → PR fix: prevent infinite restart loop that causes runaway API costs #693 already replaces the AbortController in the restart path (SessionRoutes.ts:211-214)

Still valuable — consider submitting as focused PRs:

1. AbortController check at generator start — The check for session.abortController.signal.aborted before starting a new generator (in both startSessionProcessor and startGeneratorWithProvider) is NOT covered by existing merges. When a generator completes naturally and aborts its controller, a new incoming message could try to start with the stale aborted controller. This is a real bug. Would be a clean ~10-line PR.

2. FK constraint protection in SDKAgent/ResponseProcessor — The logic to check the DB's memory_session_id before overwriting it (to avoid FK violations when the SDK generates a new session ID after restart) is genuinely valuable and not addressed elsewhere. This would be a focused PR touching SDKAgent.ts and ResponseProcessor.ts.

3. queueDepth logging using pendingStore.getPendingCount() — Minor but correct fix for deprecated session.pendingMessages.length. Could be included with #1.

If you'd like to re-submit these as 1-2 focused PRs, they'd be straightforward to review and merge. Thanks for the thorough investigation!