CamelAGI Runtime Internals

This document covers every subsystem that makes up the CamelAGI runtime layer: from message ingestion through agent execution, concurrency control, persistence, and system management.

Table of Contents

1. Orchestrator
2. Queue
3. Lanes
4. Runs
5. Retry
6. Compaction
7. Sessions
8. Usage Tracking
9. Chunker
10. Boot Scripts
11. Daemon
12. Doctor

Orchestrator

File: src/runtime/orchestrate.ts

The orchestrator is the single source of truth for the chat flow. Before it existed, the same check-active / queue / acquire-lane / run-agent / save sequence was duplicated across routes.ts, ws-handler.ts, and agent-bot.ts. Now every entry point calls orchestrate().

Step-by-step flow

1. Queue check – If a run is already active on the target sessionId, the inbound message is handed to queueOrProcess(). When queued, the caller receives a promise that resolves once the queued message is eventually processed. The orchestrator returns early with queued: true.

2. Lane acquisition – acquireLane(Lane.Main) is called to obtain a concurrency slot. If all slots are taken, the call blocks until one frees up.

3. Run registration – A unique runId is generated and an active-run handle is registered via setActiveRun(). The handle carries sessionId, runId, startedAt timestamp, an abort() callback, and an isStreaming() indicator.

4. Abort wiring – If the caller supplied an AbortSignal, a dedicated AbortController is created and linked so that external abort propagates into the agent run.

5. History loading + compaction – loadMessages(sessionId) reads the JSONL session file. The history is then passed to compactHistory() which may summarize older turns if token estimates exceed the compaction threshold. Callers are notified via the onCompact callback.

6. Agent execution with retry – The agent is run inside withRetry(). Configuration values are resolved with per-call overrides taking precedence: model, agentSystemPrompt, thinking, effort, and maxTurns all fall back to config.* when no override is supplied.

7. Message persistence – On success, both the user message and the assistant response are appended to the session JSONL file via saveMessage().

8. Cleanup (finally block) – clearActiveRun(runId) removes the run handle (notifying any waiters) and release() frees the lane slot. After cleanup, the queue for the session is drained: the first queued message is processed by recursively calling orchestrate(), and any remaining queued messages are rejected with “Superseded by newer message”.

OrchestrateOpts

OrchestrateResult

Queue

File: src/runtime/queue.ts

The message queue prevents lost messages when a user sends input while the agent is still processing a previous turn on the same session.

Data model

Each queued message is stored as a QueuedMessage:

{ text: string, resolve, reject, enqueuedAt: number }

The backing store is a Map<sessionId, QueuedMessage[]> held in memory (not persisted to disk).

Key operations

Factory pattern

createMessageQueue() returns a fresh instance. A default singleton is also exported for backward compatibility.

Lanes

File: src/runtime/lanes.ts

Lanes provide concurrency control over parallel agent runs. They prevent the system from overwhelming the LLM provider with too many simultaneous requests.

Lane types

enum Lane {
  Main     = "main",      // User-initiated chat turns
  Cron     = "cron",      // Scheduled / cron-triggered runs
  Subagent = "subagent",  // Agent-spawned sub-agent calls
}

How it works

Each lane has:

• limit – maximum number of concurrent runs (defaults to Infinity if never configured).
• active – current count of in-flight runs.
• queue – FIFO array of () => void resolve callbacks from waiters.

Acquiring a lane:

1. If active < limit, increment active immediately and return a release function.
2. Otherwise, push a resolve callback onto the FIFO queue. The caller’s await acquireLane(lane) blocks until a slot opens.

Releasing a lane:

1. Decrement active.
2. Shift the next waiter off the queue (if any) and invoke it, which unblocks that caller so it can increment active and proceed.

API

Runs

File: src/runtime/runs.ts

The run tracker prevents concurrent runs on the same session and provides abort/wait primitives.

Data structures

• Primary index: Map<runId, RunHandle> – lookup by run ID.
• Secondary index: Map<sessionId, runId> – maps each session to its latest run.
• Waiters: Map<sessionId, Set<(ended: boolean) => void>> – callbacks waiting for a run to finish.

RunHandle

{
  sessionId: string;
  runId: string;
  startedAt: number;     // Date.now() at creation
  abort: () => void;     // Triggers the run's AbortController
  isStreaming: () => boolean;
}

Key operations

Retry

File: src/runtime/retry.ts

Error classification

Errors are classified using a two-tier approach for reliability:

1. Status code extraction — The classifier first checks for .status or .statusCode properties on the error object (set by OpenAI and Anthropic SDKs). If not found, it falls back to extracting standalone 3-digit HTTP codes from the error message (e.g., “Error 429” but not “model-429b”).

2. String matching fallback — If no status code is found, error messages are matched against known patterns.

Key improvements over the previous classifier:

• server_error is a distinct type (previously misclassified as rate_limit)
• Abort detection uses exact string match and AbortError name check — no false positives from timeout messages containing “abort”
• Status code priority avoids brittle substring matching (e.g., “500” in a URL won’t trigger misclassification)

Retry behavior

withRetry(fn, opts) wraps an async function:

• rate_limit / timeout / server_error: Retried up to maxRetries times with capped exponential backoff: delay = min(backoffMs * 2^attempt, maxBackoffMs). Default cap is 30 seconds.
• overflow: Calls onCompact() (which force-compacts the history) then retries once. A flag (overflowRetried) prevents infinite compact loops.
• unknown: Retried once with a flat backoffMs delay, then fails.
• auth / billing / format / abort: Thrown immediately, no retry.

The onRetry(attempt, kind, err) callback fires before each retry sleep.

RetryOpts

Compaction

File: src/runtime/compact.ts

Compaction prevents context overflow by summarizing old conversation turns when estimated token usage gets too high.

Trigger condition

Token count is estimated at 4 characters per token (CHARS_PER_TOKEN). Compaction fires when:

estimatedTokens >= maxTokens * 0.8   (COMPACTION_TRIGGER_RATIO)

If compaction is disabled in config (enabled: false), the function returns null immediately.

Process

1. Split history – Messages are divided into old and recent by counting user-message turn boundaries. The last keepTurns turns (default 6) are kept verbatim.

2. Memory flush – Before summarizing, the old messages are passed to memoryFlush(). This sends the old conversation text (up to MEMORY_FLUSH_MAX_CHARS = 30,000 characters) to the LLM with a prompt asking it to extract durable facts as bullet points. If meaningful facts are extracted, they are appended to a daily file under a timestamped ## HH:MM:SS (auto-flush) heading. The destination is agent-scoped: when agentId is provided, notes land in ~/.camelagi/agents/<agentId>/memory/{YYYY-MM-DD}.md; otherwise they go to the global ~/.camelagi/workspace/memory/{YYYY-MM-DD}.md. The flush is best-effort: errors are silently caught.

3. Summarize – The old messages are formatted as [role]: content blocks and sent to the LLM with a summarization prompt. The summary is wrapped in a synthetic user message:

   [Previous conversation summary]
   {summary text}
   [End of summary -- conversation continues below]
   

4. Validate – The compacted result’s estimated token count is compared against the original. If the compacted result is equal to or larger than the original (e.g., when recent turns dominate the context), compaction is skipped with a warning to stderr: "⚠ Compaction skipped: result (N tokens) >= original (M tokens)". This prevents compaction from making things worse.

5. Return – The compacted history is [summaryMessage, ...recentMessages], or null if validation failed.

Constants

Sessions

File: src/session.ts

Sessions are stored as JSONL (JSON Lines) files in ~/.camelagi/sessions/.

File format

Each session file is named {urlEncodedSessionId}.jsonl. The file structure:

• Line 1 (metadata): A JSON SessionMeta object:

  { "id": "abc123", "createdAt": 1710300000000, "model": "claude-sonnet-4-20250514", "label": "My Chat" }
  

• Lines 2+N (messages): One JSON object per message:

  { "type": "user", "content": "Hello" }
  { "type": "assistant", "content": "Hi there!" }
  

Type mapping (backward compatibility)

Old LangChain-era type names are mapped on load:

New messages are saved with the current role names (user, assistant, etc.).

API

Usage Tracking

File: src/usage.ts

Per-session token accounting, stored both in memory and on disk.

What is tracked

Each session accumulates a SessionUsage record:

Storage

• In-memory: Map<sessionId, SessionUsage> for fast lookups.
• On disk: ~/.camelagi/usage/{urlEncodedSessionId}.json – a single JSON object per session, overwritten on each update.

getSessionUsage() checks the in-memory map first, falls back to reading from disk, and returns a zeroed record if neither exists.

Formatting helpers

Chunker

File: src/chunker.ts

The BlockChunker buffers streamed text and emits sized blocks suitable for Telegram (or any channel with message-length constraints).

Configuration

Break preference cascade

When the buffer exceeds minChars, the chunker searches for the last occurrence of each break type within the [minChars, maxChars] window, in order:

1. Paragraph (\n\n) – only if breakPreference is "paragraph"
2. Newline (\n)
3. Sentence ([.!?]\s)
4. Word (\s)
5. Hard break at maxChars – forced if no natural break is found

Code fence tracking

The chunker tracks whether it is inside a fenced code block by counting ``` occurrences:

• Inside a fence and below maxChars: The chunker holds off on emitting, waiting for the fence to close naturally.
• Forced break inside a fence: The chunk gets a closing ``` appended, and the remainder buffer gets an opening ``` prepended, so both halves are valid Markdown.
• Flush with open fence: A closing ``` is appended before the final chunk is emitted.

API

Boot Scripts

File: src/boot.ts

BOOT.md provides a way to run automated tasks every time the gateway starts.

How it works

1. On gateway launch, runBoot(config, systemPrompt) checks for ~/.camelagi/BOOT.md.
2. If the file does not exist or is empty, returns { status: "skipped" }.
3. Otherwise, the file content is sent as a user message to the agent with these constraints:
   • Max turns: 10 (the agent can use tools but is limited to 10 think-act iterations)
   • Timeout: 60 seconds
   • Session: "boot" (a dedicated session so boot history is isolated)
4. The user message (BOOT.md content) and assistant response are persisted to the boot session.
5. Returns { status: "ran", response } on success or { status: "failed", error } on error.

Constraints

Daemon

File: src/daemon.ts

macOS launchd integration for running the CamelAGI server as a persistent background service.

Plist generation

The generatePlist() function produces a standard macOS property list with:

The plist is written to ~/Library/LaunchAgents/com.camelagi.server.plist.

Node.js path is resolved via which node at install time (falls back to /usr/local/bin/node).

Commands

Doctor

File: src/doctor.ts

The doctor runs a comprehensive suite of health checks and returns structured results.

Check format

Each check is a { name, status, message } object where status is one of:

• ok – green checkmark
• warn – yellow exclamation
• error – red X

Output is formatted with ANSI color codes for terminal display.

All checks performed

System Interaction Diagram

User message
     |
     v
orchestrate()
     |
     +-- isRunActive? --yes--> queueOrProcess() --> wait --> resolve later
     |
     +-- acquireLane(Main) ----------> [blocks if lane full]
     |
     +-- setActiveRun()
     |
     +-- loadMessages() + compactHistory()
     |         |
     |         +-- estimateTokens() >= 80% maxTokens?
     |         |       |
     |         |       +-- memoryFlush() --> agent or global memory/{date}.md
     |         |       +-- chatDirect()  --> summary
     |         |       +-- return [summary, ...recentTurns]
     |         |
     |         +-- no --> return null
     |
     +-- withRetry( runAgent(...) )
     |         |
     |         +-- on rate_limit/server_error/timeout --> min(backoff * 2^attempt, 30s) --> retry
     |         +-- on overflow --> onCompact() --> retry once
     |         +-- on auth/billing/format/abort --> throw
     |
     +-- saveMessage(user) + saveMessage(assistant)
     |
     +-- finally: clearActiveRun() + release()
     |         |
     |         +-- drainQueue(sessionId)
     |               |
     |               +-- queued[0] --> orchestrate() --> resolve promise
     |               +-- queued[1..n] --> reject("Superseded")
     |
     v
OrchestrateResult