TinyAGI: The Agent Teams Orchestrator Built for the One-Person Company

tinyagi/
├── packages/
│   ├── core/        # Shared types, config, SQLite queue, agent invocation
│   ├── main/        # Queue processor entry point
│   ├── teams/       # Team conversation orchestration, chain execution
│   ├── server/      # REST + SSE API server
│   ├── channels/    # Discord, Telegram, WhatsApp channel clients
│   ├── cli/         # CLI commands
│   └── visualizer/  # TUI dashboard and chatroom viewer
├── tinyoffice/      # Next.js web portal
└── .tinyagi/        # Runtime data: SQLite db, logs, channel state, chats

~/tinyagi-workspace/
├── tinyagi/         # default agent workspace
├── coder/           # coder agent workspace
└── writer/          # writer agent workspace

// packages/core/src/queue/queue.ts (reconstructed from architecture docs + README)
// The SQLite queue is the reliability foundation of TinyAGI.
// Why SQLite instead of Redis or an in-memory queue?
// ← Durability: survives process crashes without lost messages
// ← Simplicity: no separate database process to manage
// ← Atomicity: transactions prevent partial state on concurrent access
// ← The bottleneck is LLM API calls (seconds), not queue ops (microseconds)

import Database from 'better-sqlite3';
import path from 'path';
import os from 'os';

interface Message {
  id: string;
  agent_id: string;
  channel: string;
  sender: string;
  content: string;
  status: 'pending' | 'processing' | 'completed' | 'dead';
  receive_count: number;
  created_at: number;
  updated_at: number;
}

class AgentMessageQueue {
  private db: Database.Database;
  private readonly MAX_RECEIVE_COUNT = 3;  // after 3 failures → dead letter

  constructor() {
    const dbPath = path.join(os.homedir(), '.tinyagi', 'tinyagi.db');
    this.db = new Database(dbPath);

    // ← WAL mode: critical for concurrent access
    // Multiple agents reading + the queue processor writing simultaneously
    // would deadlock in journal mode. WAL allows concurrent reads.
    this.db.pragma('journal_mode = WAL');
    this.db.pragma('foreign_keys = ON');

    this.initSchema();
  }

  private initSchema(): void {
    // ← Status as a state machine: pending → processing → completed
    //   OR pending → processing → pending (retry) × N → dead
    // Dead-lettered messages are kept for inspection, not deleted.
    this.db.exec(`
      CREATE TABLE IF NOT EXISTS messages (
        id TEXT PRIMARY KEY,
        agent_id TEXT NOT NULL,
        channel TEXT NOT NULL,
        sender TEXT NOT NULL,
        content TEXT NOT NULL,
        status TEXT NOT NULL DEFAULT 'pending',
        receive_count INTEGER NOT NULL DEFAULT 0,
        created_at INTEGER NOT NULL,
        updated_at INTEGER NOT NULL
      );

      CREATE INDEX IF NOT EXISTS idx_messages_status_agent
        ON messages(status, agent_id);  -- ← critical for queue processor polling

      CREATE TABLE IF NOT EXISTS responses (
        id TEXT PRIMARY KEY,
        message_id TEXT NOT NULL REFERENCES messages(id),
        agent_id TEXT NOT NULL,
        content TEXT NOT NULL,
        status TEXT NOT NULL DEFAULT 'pending',
        created_at INTEGER NOT NULL
      );
    `);
  }

  enqueueMessage(agentId: string, channel: string, sender: string, content: string): string {
    const id = crypto.randomUUID();
    const now = Date.now();

    // ← Atomic insert: either fully written or not at all
    // No partial state if the process crashes mid-write
    const stmt = this.db.prepare(`
      INSERT INTO messages (id, agent_id, channel, sender, content, status, receive_count, created_at, updated_at)
      VALUES (?, ?, ?, ?, ?, 'pending', 0, ?, ?)
    `);
    stmt.run(id, agentId, channel, sender, content, now, now);
    return id;
  }

  receiveMessage(agentId: string): Message | null {
    // ← THIS is the trick: the entire receive-and-lock is one atomic transaction
    // Two queue processors can't pick up the same message because the
    // status update and the SELECT happen in the same transaction.
    const receive = this.db.transaction(() => {
      const msg = this.db.prepare(`
        SELECT * FROM messages
        WHERE agent_id = ? AND status = 'pending'
        ORDER BY created_at ASC
        LIMIT 1
      `).get(agentId) as Message | undefined;

      if (!msg) return null;

      // Check if this message has exceeded retry limit → dead letter
      if (msg.receive_count >= this.MAX_RECEIVE_COUNT) {
        this.db.prepare(`UPDATE messages SET status = 'dead', updated_at = ? WHERE id = ?`)
               .run(Date.now(), msg.id);
        return null;  // ← dead-lettered: caller doesn't process it
      }

      // Lock for processing
      this.db.prepare(`
        UPDATE messages
        SET status = 'processing', receive_count = receive_count + 1, updated_at = ?
        WHERE id = ?
      `).run(Date.now(), msg.id);

      return { ...msg, status: 'processing', receive_count: msg.receive_count + 1 };
    });

    return receive();
  }

  completeMessage(id: string): void {
    this.db.prepare(`UPDATE messages SET status = 'completed', updated_at = ? WHERE id = ?`)
           .run(Date.now(), id);
  }

  // If an agent crashes mid-processing, this resets stuck messages
  resetStalledMessages(stalledThresholdMs: number = 300_000): void {
    const cutoff = Date.now() - stalledThresholdMs;
    this.db.prepare(`
      UPDATE messages SET status = 'pending', updated_at = ?
      WHERE status = 'processing' AND updated_at < ?
    `).run(Date.now(), cutoff);
  }

  deadLetters(agentId?: string): Message[] {
    const query = agentId
      ? `SELECT * FROM messages WHERE status = 'dead' AND agent_id = ?`
      : `SELECT * FROM messages WHERE status = 'dead'`;
    return (agentId ? this.db.prepare(query).all(agentId) : this.db.prepare(query).all()) as Message[];
  }
}

// packages/teams/src/router.ts + packages/core/src/agent.ts (reconstructed)
// The routing protocol is implemented in two layers:
// 1. Tag parsing: [@agent_id: message] → {agentId, message, sharedContext}
// 2. Agent invocation: run the appropriate CLI command in the agent's workspace

interface RouteTarget {
  agentId: string;
  message: string;
  sharedContext?: string;
}

// Parse routing tags from a message
// Examples:
//   "[@coder: fix the bug]" → [{agentId: 'coder', message: 'fix the bug'}]
//   "[@coder: fix X] [@writer: document Y]" → two targets, invoked in PARALLEL
//   "Sprint ends Friday.\n[@coder: also list PRs] [@reviewer: also flag blockers]"
//     → shared context 'Sprint ends Friday.' delivered to both, plus each gets own tag
function parseRoutingTags(rawMessage: string): RouteTarget[] {
  const tagPattern = /\[@([\w-]+):\s*(.*?)\]/gs;
  const targets: RouteTarget[] = [];

  // Extract shared context: text OUTSIDE the [@agent: ...] tags
  const sharedContext = rawMessage.replace(tagPattern, '').trim();

  // Reset regex lastIndex for reuse
  tagPattern.lastIndex = 0;

  let match;
  while ((match = tagPattern.exec(rawMessage)) !== null) {
    const [, rawIds, message] = match;
    // Handle comma-separated agent IDs: [@coder,writer,tester: message]
    // ← All get the same message — broadcast mode
    const agentIds = rawIds.split(',').map(id => id.trim());

    for (const agentId of agentIds) {
      targets.push({
        agentId,
        message: message.trim(),
        sharedContext: sharedContext || undefined,
      });
    }
  }

  // If no tags found, route to default agent
  if (targets.length === 0) {
    targets.push({ agentId: 'tinyagi', message: rawMessage });
  }

  return targets;
}

// Invoke an agent: run the LLM CLI in the agent's isolated workspace
async function invokeAgent(
  agentId: string,
  message: string,
  config: AgentConfig,
): Promise<string> {
  const workspacePath = path.join(os.homedir(), 'tinyagi-workspace', agentId);

  // ← Each agent invocation runs in its OWN workspace directory
  // cd coder/ before running claude → Claude Code's file operations are isolated
  // No context bleed: coder agent's conversation history is separate from writer's
  const options = {
    cwd: workspacePath,
    env: { ...process.env, ANTHROPIC_API_KEY: config.apiKey },
  };

  if (config.provider === 'claude') {
    // -c: continue conversation (persistent session)
    // -p: prompt (non-interactive, stdout output only)
    // --dangerously-skip-permissions: allow file I/O without confirmation
    const result = await execa('claude', [
      '--dangerously-skip-permissions',
      '--model', config.model,  // e.g., claude-sonnet-4-6
      '-c',                     // ← continue from last session
      '-p', message,            // ← the actual message/task
    ], options);
    return result.stdout;

  } else if (config.provider === 'codex') {
    // resume --last: continue the most recent Codex session
    const result = await execa('codex', [
      'exec', 'resume', '--last',
      '--model', config.model,
      '--skip-git-repo-check',
      '--dangerously-bypass-approvals-and-sandbox',
      '--json',
      message,
    ], options);
    return JSON.parse(result.stdout).output;
  }

  throw new Error(`Unsupported provider: ${config.provider}`);
}

// Fan-out dispatcher: invoke all targets in parallel
async function dispatchToAgents(
  rawMessage: string,
  configs: Map<string, AgentConfig>,
): Promise<Map<string, string>> {
  const targets = parseRoutingTags(rawMessage);

  // ← Fan-out: all targets dispatched simultaneously
  // A message to [@coder] and [@writer] starts BOTH agent sessions at the same time
  // Neither waits for the other → parallel execution
  const results = await Promise.all(
    targets.map(async ({ agentId, message, sharedContext }) => {
      const config = configs.get(agentId) ?? configs.get('tinyagi')!;
      const fullMessage = sharedContext
        ? `${sharedContext}\n\n${message}`
        : message;
      const response = await invokeAgent(agentId, fullMessage, config);
      return [agentId, response] as const;
    })
  );

  return new Map(results);
}

Sprint ends Friday, 3 open bugs.
Reply with: (1) status (2) blockers (3) next step.

[@coder: Also list any PRs you have open.]
[@reviewer: Also flag any PRs waiting on you.]
[@tester: Also report test coverage for the auth module.]

Discord channel client receives the message.
enqueueMessage():
  → 3 queue entries created (one per @tag):
    {id: 'msg-a1', agent_id: 'coder',    content: '[shared context]\n[coder-specific]', status: 'pending'}
    {id: 'msg-b2', agent_id: 'reviewer', content: '[shared context]\n[reviewer-specific]', status: 'pending'}
    {id: 'msg-c3', agent_id: 'tester',   content: '[shared context]\n[tester-specific]', status: 'pending'}

SQLite write: ~0.1ms per enqueue (negligible vs LLM latency)

Queue processor polls every 500ms.
receiveMessage('coder'):    msg-a1 → status: 'processing'
receiveMessage('reviewer'): msg-b2 → status: 'processing'
receiveMessage('tester'):   msg-c3 → status: 'processing'

All three invocations launched in parallel via Promise.all:

Agent coder:
  cd ~/tinyagi-workspace/coder/
  claude --dangerously-skip-permissions --model claude-sonnet-4-6 -c -p "[message]"
  → Claude Code opens coder's conversation context
  → Reads coder's workspace files (git log, open PRs, recent commits)
  → Response: "Status: 2 bugs fixed. Blockers: waiting on API docs. PRs: #47, #51"

Agent reviewer:
  cd ~/tinyagi-workspace/reviewer/
  claude --dangerously-skip-permissions --model claude-sonnet-4-6 -c -p "[message]"
  → Uses reviewer's isolated context (different from coder's)
  → Response: "Status: reviewing PR #47. Blockers: need unit tests on auth changes."

Agent tester:
  cd ~/tinyagi-workspace/tester/
  claude --dangerously-skip-permissions --model claude-sonnet-4-6 -c -p "[message]"
  → Checks test files in tester workspace
  → Response: "Auth module coverage: 73%. 2 missing edge cases identified."

All three running concurrently: total LLM latency ≈ max(coder, reviewer, tester) not sum
Typical: 8-15 seconds per agent call → fan-out saves ~2x wall-clock time vs sequential

All three responses enqueued to responses table.
Response delivery:
  msg-a1 response → Discord reply in thread (or DM): coder's status
  msg-b2 response → Discord reply: reviewer's status
  msg-c3 response → Discord reply: tester's status

Total latency: ~10-15 seconds from Discord message to all three replies
(bounded by slowest agent, not sum of all three)

For sequential handoff: "@coder: draft the fix. When done, @reviewer: review it"
Agent router detects chain instruction inside coder's response or message.
coder runs → response triggers reviewer invocation
Latency: sequential, not parallel (coder response time + reviewer response time)
Use when: second agent depends on first agent's output
Use fan-out when: agents are independent