Deterministic Event Pipelines for AI Agents

1. The Problem: The "Forever Loop" Fallacy

In 95% of AI tutorials, agents are implemented as recursive loops:

// The "Forever Loop" Pattern
while (true) {
  const plan = await llm.think(history);
  if (plan.isFinished) break;
  const result = await tool.execute(plan.action); // BLOCKING
  history.push(result);
}

This pattern is acceptable for prototypes but catastrophic for production systems.

Critical Failures:

1. Concurrency Deadlocks: The runtime is blocked during tool execution. If your agent needs to fetch data (500ms) and simultaneously listen for a user interrupt, the loop architecture fails.
2. Observability Black Holes: State exists only within the closure of the loop. If the process crashes on iteration #4, you lose the stack. You cannot "replay" the session.
3. Testing Impossibility: You cannot unit test the "Tool Execution" logic without mocking the entire LLM response chain.

The Solution: Treat the Agent not as a function, but as a State Machine driven by a discrete Event Pipeline.

2. The Solution: The Cognitive Event Bus

We decouple Inference (Brain) from Execution (Hands). The components communicate only via typed events.

The Pipeline:

1. Ingestion: User input is normalized into an INPUT_RECEIVED event.
2. Cognition: The LLM listens to events, updates internal context, and emits INTENT_DETECTED.
3. Router: A deterministic layer validates the intent and emits EXECUTION_REQUESTED.
4. Effect: Workers execute the tool and emit TOOL_RESULT.
5. Loopback: The TOOL_RESULT is fed back into the Cognition layer.

3. Project Structure

Separation of concerns is enforced at the directory level.

/src
├── /core
│   ├── bus.ts             # Typed EventEmitter Singleton
│   ├── types.ts           # Zod Schemas for Event Payloads
│   └── store.ts           # Redis/Postgres State Persistence
├── /reactors
│   ├── cognition.ts       # Logic: Event -> LLM -> Intent
│   └── execution.ts       # Logic: Intent -> Tool -> Result
├── /tools
│   ├── network.ts         # Axios/Fetch Wrappers
│   └── filesystem.ts      # Safe FS Operations
└── index.ts               # Pipeline Composition

4. Code Implementation

We use TypeScript to enforce strict contracts on the event bus. This eliminates "magic string" bugs.

A. The Typed Event Bus

// src/core/bus.ts
import { EventEmitter } from 'events';

export enum AgentEvent {
  INPUT = 'AGENT:INPUT',
  INTENT = 'AGENT:INTENT',
  EXECUTE = 'AGENT:EXECUTE',
  RESULT = 'AGENT:RESULT',
  ERROR = 'AGENT:ERROR',
}

// Strict Payload Definitions
interface EventPayloads {
  [AgentEvent.INPUT]: { id: string; content: string };
  [AgentEvent.INTENT]: { id: string; tool: string; args: unknown };
  [AgentEvent.EXECUTE]: { id: string; tool: string; args: unknown };
  [AgentEvent.RESULT]: { id: string; output: string };
  [AgentEvent.ERROR]: { id: string; error: Error };
}

class TypedBus extends EventEmitter {
  emit<K extends keyof EventPayloads>(event: K, payload: EventPayloads[K]): boolean {
    console.log(`[BUS] ${event}`, payload); // Free Observability
    return super.emit(event, payload);
  }

  on<K extends keyof EventPayloads>(event: K, fn: (p: EventPayloads[K]) => void): this {
    return super.on(event, fn);
  }
}

export const bus = new TypedBus();

B. The Cognition Reactor (The Brain)

This component is pure. It creates no side effects. It simply transforms state into intent.

// src/reactors/cognition.ts
import { bus, AgentEvent } from '../core/bus';
import { llm } from '../lib/llm';

export const registerCognition = () => {
  const cycle = async (sessionId: string, context: any[]) => {
    try {
      const response = await llm.generate(context);

      if (response.tool_calls) {
        // Emit intent, do NOT execute
        bus.emit(AgentEvent.INTENT, {
          id: sessionId,
          tool: response.tool_calls[0].function.name,
          args: JSON.parse(response.tool_calls[0].function.arguments),
        });
      } else {
        // Handle final answer
      }
    } catch (err) {
      bus.emit(AgentEvent.ERROR, { id: sessionId, error: err as Error });
    }
  };

  // Trigger on Input OR Tool Results (The Loopback)
  bus.on(AgentEvent.INPUT, p => cycle(p.id, [{ role: 'user', content: p.content }]));
  bus.on(AgentEvent.RESULT, p => cycle(p.id, [{ role: 'tool', content: p.output }]));
};

C. The Execution Reactor (The Hands)

This component is "dumb." It knows nothing of the LLM. It simply executes commands.

// src/reactors/execution.ts
import { bus, AgentEvent } from '../core/bus';
import { tools } from '../tools';

export const registerExecution = () => {
  bus.on(AgentEvent.EXECUTE, async ({ id, tool, args }) => {
    try {
      // 1. Validation Logic
      if (!tools[tool]) throw new Error(`Unknown tool: ${tool}`);

      // 2. Execution
      const output = await tools[tool](args);

      // 3. Emit Result
      bus.emit(AgentEvent.RESULT, { id, output: JSON.stringify(output) });
    } catch (err) {
      // Even errors are events, allowing the LLM to self-correct
      bus.emit(AgentEvent.RESULT, { id, output: `Error: ${err.message}` });
    }
  });
};

5. The Strategic Advantage

Metric	Monolithic Loop	Event Pipeline
Scalability	Vertical (Single Process)	Horizontal (Worker Pools can listen to EXECUTE events)
Debugging	Parsing 50MB log files	Time-travel replay of the event stream
Reliability	One unhandled exception kills the agent	Errors are just data; the pipeline persists

> Architectural Axiom: Determinism in AI systems is not about controlling the model's output (which is impossible). It is about controlling the state transitions that result from that output. By adopting an Event Bus, you move from "scripting a bot" to orchestrating a distributed system. This is the only path to production-grade reliability.