Queue System & Processing Model
Relevant source files
The following files were used as context for generating this wiki page:
The Queue System in claude-mem is a robust, persistent messaging infrastructure designed to handle asynchronous AI processing tasks, specifically tool-use observations and session summaries. It utilizes a "Database as the Queue" architecture, moving away from fragile in-memory arrays to a single source of truth within a SQLite database. This ensure that tasks survive application crashes and prevents race conditions through atomic state transitions.
The processing model follows a linear "Pump" pattern, where dedicated processors for each session consume messages using event-driven triggers. This architecture minimizes latency by reacting immediately to new data while maintaining high reliability through persistent state management and automated recovery for stuck messages.
Sources: QUEUE-SIMPLIFICATION-PLAN.md:1-12, QUEUE-SYSTEM-ANALYSIS.md:18-25
Core Architecture: The "Pump" Model
The simplified architecture consolidates distributed logic into a unified "Pump" loop. This loop performs three primary steps: atomically claiming a message from the database, yielding it to the SDK agent for processing, and marking it as complete upon success.
Message Lifecycle State Machine
Messages transition through specific states to ensure exactly-once processing (under normal conditions) and reliable recovery during failures.
The state machine ensures that messages only move to terminal states (processed, failed) after definitive outcomes, with intermediate processing states protected by timestamps for stuck-detection.
Sources: QUEUE-SYSTEM-ANALYSIS.md:46-77, QUEUE-SIMPLIFICATION-PLAN.md:46-63
Atomic Message Claiming
The system implements a claimNextMessage() operation in the PendingMessageStore. This is an atomic transaction that selects the oldest pending message and updates its status to processing in a single step, preventing multiple workers from claiming the same task.
Sources: QUEUE-SIMPLIFICATION-REPORT.md:8-14, QUEUE-SIMPLIFICATION-PLAN.md:37-44
Components and Logic
The system is composed of several key layers that handle storage, coordination, and execution.
1. PendingMessageStore (Data Layer)
This component manages the SQLite table pending_messages. It handles the persistence of message data, including tool inputs and responses.
| Field | Type | Description |
|---|---|---|
status | TEXT | Current state: 'pending', 'processing', 'processed', or 'failed'. |
retry_count | INTEGER | Number of times processing has been attempted. |
started_processing_at_epoch | INTEGER | Timestamp when a worker claimed the message. |
created_at_epoch | INTEGER | Original timestamp for queue ordering. |
Sources: QUEUE-SYSTEM-ANALYSIS.md:86-130, scripts/check-pending-queue.ts:16-25
2. SessionQueueProcessor (The Pump)
The SessionQueueProcessor encapsulates the core processing loop. It uses an AsyncIterableIterator to yield messages to consumers. It is event-driven, utilizing an EventEmitter to wake up the loop immediately when a new message is enqueued, rather than relying on inefficient polling.
3. SDKAgent (The Consumer)
The SDKAgent consumes messages yielded by the processor. It builds prompts (Initialization or Continuation) and feeds them into the Claude Agent SDK via the query function.
// SDKAgent handles the query loop (SDKAgent.ts:75-100)
const queryResult = query({
prompt: messageGenerator,
options: {
model: modelId,
resume: session.memorySessionId,
disallowedTools,
abortController: session.abortController
}
});
Sources: src/services/worker/SDKAgent.ts:75-100, QUEUE-SYSTEM-ANALYSIS.md:397-477
Recovery and Health Mechanisms
The processing model includes several automated and manual mechanisms to handle system failures and backlog.
Stuck Message Recovery
At startup, the WorkerService invokes resetStuckMessages(). This function identifies messages that have been in the processing state for longer than a specific threshold (default 5 minutes) and resets them to pending. This recovers tasks that were interrupted by worker crashes or hung SDK calls.
Sources: src/services/worker-service.ts:627-633, QUEUE-SYSTEM-ANALYSIS.md:281-292
Manual Backlog Management
The project includes a CLI utility check-pending-queue.ts to manage the queue manually.
| Option | Description |
|---|---|
--process | Auto-process the backlog without prompting. |
--limit N | Process up to N sessions (default: 10). |
--help | Show status and summary of pending/stuck messages. |
Sources: scripts/check-pending-queue.ts:98-115
API Endpoints for Queue Control
The worker service exposes HTTP endpoints to interact with the queue system.
| Endpoint | Method | Description |
|---|---|---|
/api/pending-queue | GET | Returns a summary of pending, processing, and stuck messages. |
/api/pending-queue/process | POST | Triggers background processing of orphaned queues. |
Sources: scripts/check-pending-queue.ts:43-60
Conclusion
The Queue System and Processing Model in claude-mem provide a resilient foundation for asynchronous AI tasks. By leveraging the database as the single source of truth and employing an event-driven "Pump" architecture, the system achieves high reliability and low latency. Automated recovery for stuck messages and dedicated CLI tools for backlog management ensure the system remains healthy even across crashes and restarts.