Channel State & Queuing

Relevant source files

The following files were used as context for generating this wiki page:

• src/channel/queue.ts
• src/channel/state.ts
• src/main.ts
• src/slack/router.ts
• src/slack/client.ts
• src/agent/setup.ts

The Channel State & Queuing system manages the lifecycle, persistence, and execution flow of agent interactions within individual communication channels (e.g., Slack channels or Direct Messages). It ensures that requests are processed sequentially per channel, maintains a historical log of interactions, and tracks the operational status of active agent runs.

This system acts as the intermediary between incoming events (from Slack or Linear) and the agent execution logic, providing a stable environment for the agent to operate within a specific context.

Channel Runtime Management

The project maintains a global registry of channel states using a Map where each key is a channelId. This allows the bot to handle multiple channels concurrently while keeping their internal states isolated.

ChannelRunState Structure

Each channel is associated with a ChannelRunState object that tracks:

• Execution Status: Whether an agent is currently running in the channel.
• Persistence: A ChannelState instance for logging and directory management.
• Concurrency Control: A ChannelQueue to ensure messages are processed one at a time.
• Control Signals: Flags for stopRequested and tracking the timestamp of stop-indicator messages.

Sources: src/main.ts:63-71, src/main.ts:73-86

Component Relationship

The following diagram illustrates how the per-channel state components interact during an event lifecycle.

The system ensures that for every unique channel ID, a consistent set of state and queue objects is retrieved or created.
Sources: src/main.ts:75-84, src/main.ts:139-140

Sequential Execution Queuing

To prevent race conditions and conflicting tool executions within the same workspace, each channel utilizes a ChannelQueue. This queue serializes asynchronous work, ensuring that only one handleEvent operation is active at any given time for a specific channel.

The ChannelQueue Class

The queue operates on QueuedWork (functions returning a Promise<void>). It uses a processing boolean flag to guard the execution loop. When new work is enqueued, it is added to an internal array and processNext() is called. If the queue is already processing, the call returns early, and the currently running work will trigger the next item upon completion.

Sources: src/channel/queue.ts:3-29

Event Enqueuing Flow

When the Slack router receives a valid event, it uses the ChannelQueue to schedule processing.

Channel State Persistence

The ChannelState class manages the physical directory structure and message logging for a specific channel. It handles the mapping between a channel ID and its corresponding directory on the host machine.

Log Management

All messages (user messages and bot responses) are persisted to a log.jsonl file within the channel's directory. This file serves as the primary history for the agent and is used for "backfilling" missed messages during startup.

Method	Description	Data Logged
logUserMessage	Records incoming user events.	ISO Date, TS, User ID, Username, Text, Attachments.
logBotResponse	Records final agent replies.	ISO Date, TS, Bot ID, Text, threadTs.
getLogTimestamps	Extracts all ts values.	Used to deduplicate messages during backfill.

Sources: src/channel/state.ts:7-63

Directory Structure

ChannelState ensures that the following structure exists within the configured workingDir:

• {workingDir}/{channelId}/: Root directory for the channel.
• {workingDir}/{channelId}/log.jsonl: Sequential message log.
• {workingDir}/{channelId}/attachments/: Stored files shared by users.
• {workingDir}/{channelId}/context.jsonl: Detailed agent session history (managed by AgentSession).

Sources: src/channel/state.ts:11-18, src/agent/setup.ts:117-118, src/agent/prompt.ts:60-65

Backfill and Synchronization

Upon startup, the system performs a "backfill" operation to sync logs for all channels where the bot is a member.

1. Deduplication: It retrieves existing timestamps from the local log.jsonl via getLogTimestamps().
2. API Fetch: It calls the Slack conversations.history API to fetch messages newer than the latest local entry.
3. Persistence: New relevant messages (excluding other bots and non-standard subtypes) are appended to the local log.

Sources: src/main.ts:200-221, src/slack/client.ts:167-217

Stop Request Handling

The state management system includes a mechanism to gracefully abort running agents. When a "stop" command is received:

1. The stopRequested flag in ChannelRunState is set to true.
2. The runner.abort() method is called to signal the LLM and tool execution loop to terminate.
3. A placeholder message ("Stopping...") is posted and tracked via stopMessageTs.
4. Once aborted, the UI is updated to reflect the stopped state.

Sources: src/main.ts:187-194, src/agent/setup.ts:321-324

Summary

The Channel State & Queuing system provides the foundational structure for pi-digby's multi-channel operations. By isolating state into per-channel maps, serializing execution through a non-blocking queue, and maintaining strict file-based logging, the system ensures that the agent remains context-aware and stable even in high-activity environments.