How the Event-Driven SessionManager Handles Session Lifecycle Events and Message Queuing in Claude-mem

The event-driven SessionManager orchestrates Claude-mem's session lifecycles using Node.js EventEmitters to enable zero-latency message queuing, persisting all observations to SQLite before emitting events that wake async generators for immediate processing.

The Claude-mem project implements a sophisticated worker process architecture centered around the SessionManager class in src/services/worker/SessionManager.ts. This event-driven SessionManager serves as the central nervous system of the application, managing everything from session initialization to graceful shutdown while maintaining crash-resilient message queues backed by SQLite.

Session Creation and Lazy Loading

The initializeSession(sessionDbId, …) method in src/services/worker/SessionManager.ts constructs an ActiveSession object the first time a session ID is referenced (lines 48-78). If the session already exists in memory, the method returns the cached instance and refreshes mutable fields (e.g., project) from the database to prevent stale data.

A fresh session always starts with a cleared memorySessionId to ensure a new SDK context is captured on the first response, preventing crashes when the worker restarts (lines 135-144). Each session maintains its own EventEmitter stored in sessionQueues (lines 71-73), enabling event-driven signaling without polling overhead.

Message Queuing Architecture

Both queueObservation and queueSummarize methods follow an identical crash-resilient pattern:

1. Auto-initialize: If the session isn't in memory, initializeSession is called so a restarted worker can resume work that only lives in the DB (lines 200-206).
2. Persist to DB first: A PendingMessage is built and enqueued in PendingMessageStore. This guarantees crash-safe durability before any in-memory handling (lines 208-219).
3. Emit event: After the DB write, the per-session EventEmitter fires a 'message' event, waking the generator instantly (zero-latency) (lines 232-235).
4. Logging: Detailed structured logs record the enqueue operation, queue depth, and a formatted tool summary (lines 220-231).

Observations store full tool data (tool_name, tool_input, tool_response, …) while summarization messages are lightweight (type: 'summarize'). Both are later claimed by the SDK agent.

Event-Driven Message Consumption

The getMessageIterator(sessionDbId) method serves as the consumer side of the queue. It ensures the session exists (lazily re-creating lost sessions from the DB), retrieves the session's EventEmitter, and instantiates SessionQueueProcessor with the shared PendingMessageStore and emitter.

The processor's createIterator method awaits the 'message' event instead of polling. When a message is yielded, the iterator updates earliestPendingTimestamp so timestamps remain accurate across back-logs (lines 29-35). The iterator also passes an onIdleTimeout callback that aborts the session's AbortController if no work arrives for a configurable period, preventing the underlying Claude subprocess from becoming a zombie (lines 24-27).

Graceful Session Termination

The deleteSession method aborts the SDK agent, waits for any outstanding generator promise, then ensures the child process exits (with a 5-second timeout) before cleaning in-memory maps and emitting a status-update callback (lines 78-106). The removeSessionImmediate method serves as a lightweight fallback used when an SDK resume fails, removing the session without awaiting the generator to avoid deadlocks (lines 27-38).

The shutdownAll method iterates over all active IDs and runs deleteSession in parallel, guaranteeing a clean shutdown of the whole worker process.

Observability and Metrics

The manager offers helpers for UI components and monitoring:

• hasPendingMessages() returns true if any DB-stored work is pending or processing.
• getActiveSessionCount() returns the number of sessions held in memory.
• getTotalQueueDepth() and getTotalActiveWork() provide aggregate pending counts across sessions.
• isAnySessionProcessing() is used by spinners to keep the UI alive while the SDK is busy.

All helpers delegate to the single PendingMessageStore which tracks pending versus processing states, making the queue crash-resilient and allowing accurate activity indicators.

Summary

• The event-driven SessionManager in src/services/worker/SessionManager.ts uses Node.js EventEmitter per session to eliminate polling latency.
• All messages are persisted to SQLite via PendingMessageStore before in-memory processing, ensuring crash resilience.
• The getMessageIterator creates zero-latency consumption by awaiting 'message' events rather than polling.
• Lazy loading via initializeSession allows workers to resume sessions from the database after restarts.
• Graceful termination ensures SDK subprocesses exit cleanly with timeouts and abort controllers.

Frequently Asked Questions

How does the SessionManager ensure messages aren't lost if the worker crashes?

The SessionManager writes every observation and summarize request to the PendingMessageStore (SQLite database) before emitting any in-memory events. This pattern, implemented in queueObservation and queueSummarize (lines 208-219), guarantees that messages survive worker restarts. When the worker resumes, initializeSession reconstructs the session state from the database, and getMessageIterator picks up unprocessed messages.

What prevents the Claude SDK subprocess from becoming a zombie session?

The getMessageIterator method passes an onIdleTimeout callback to the SessionQueueProcessor (lines 24-27). If no messages arrive for a configurable timeout period, this callback triggers the session's AbortController, forcibly terminating the underlying Claude subprocess. Additionally, deleteSession implements a 5-second timeout when waiting for the child process to exit (lines 78-106).

How does the event-driven architecture achieve zero-latency message delivery?

Instead of polling the database, the SessionManager stores a dedicated EventEmitter for each active session in sessionQueues (lines 71-73). When queueObservation or queueSummarize completes the database write, it emits a 'message' event on that specific emitter (lines 232-235). The SessionQueueProcessor awaits this event in createIterator, waking the async generator instantly when new work arrives.

Can the SessionManager handle multiple sessions simultaneously?

Yes. The SessionManager maintains a Map of active sessions and creates isolated EventEmitter instances for each session ID. The shutdownAll method demonstrates parallel session handling by mapping deleteSession across all active IDs. Observability helpers like getActiveSessionCount and getTotalQueueDepth aggregate state across all concurrent sessions, while PendingMessageStore ensures thread-safe SQLite operations.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}