How the Archon Workflow Event Emitter Tracks Step-by-Step Execution

The WorkflowEventEmitter singleton in packages/workflows/src/event-emitter.ts publishes a typed stream of events—including workflow_started, node_started, tool_completed, and loop_iteration_started—that captures every phase of execution without blocking the main thread.

Archon's workflow engine, located in the coleam00/Archon repository, provides full observability into multi-step AI workflows through a centralized event system. The WorkflowEventEmitter class acts as the single source of truth for execution state, publishing granular events consumed by the Web UI, database layers, and test suites. This article examines how the emitter integrates with the executor to track every node transition, tool invocation, and loop iteration.

Core Architecture of the Workflow Event Emitter

Singleton Pattern and Run Registration

The emitter uses a singleton pattern exported via getWorkflowEventEmitter() (lines 47-54 in packages/workflows/src/event-emitter.ts), ensuring all workflow runs share a unified event bus. Before execution begins, the executor calls registerRun(runId, conversationId) (lines 71-84) to map each run ID to its parent conversation, enabling filtered subscriptions per chat session.

Typed Events and Fault-Tolerant Emission

All possible events are defined by the WorkflowEmitterEvent union type (lines 27-60), covering the full lifecycle from workflow_started through node_failed to workflow_completed. The emit(event) method (lines 200-208) wraps the underlying Node.js EventEmitter in a try/catch block that logs listener errors via the internal pino logger without propagating failures, ensuring fire-and-forget safety.

Step-by-Step Execution Tracking in the Executor

Workflow Lifecycle Events

The executor (packages/workflows/src/executor.ts) emits high-level events at run boundaries. Immediately after database row creation, it calls workflow_started (lines 13-18). Upon completion or unrecoverable failure, it emits workflow_completed or workflow_failed near lines 1100-1150.

Node-Level Execution Tracking

Before processing any node, the executor fires node_started (lines 50-57), including the node ID and input context. If the command loader fails or the assistant returns an error, node_failed (lines 79-86) is emitted with error details. On success, node_completed signals output availability.

Tool Calls and Loop Iterations

Inside node execution, each external tool call triggers tool_started followed by tool_completed (lines 100-110). For DAG-based loops, packages/workflows/src/dag-executor.ts emits loop_iteration_started and loop_iteration_completed (lines 900-940) for every cycle, enabling progress bars in the UI.

Consuming Workflow Events

Consuming applications register listeners through two APIs defined in packages/workflows/src/event-emitter.ts. subscribe(listener) (lines 212-221) receives all events system-wide, while subscribeForConversation(conversationId, listener) (lines 223-240) filters to a specific chat, returning an unsubscribe function for cleanup.

import { getWorkflowEventEmitter } from '@archon/workflows';

// Global logging of all workflow events
const unsubscribe = getWorkflowEventEmitter().subscribe(event => {
  console.log(`[${event.type}] run=${event.runId}`, event);
});

// Cleanup when done
unsubscribe();

For SSE-based real-time updates, scope the listener to a single conversation:

const unsubscribe = getWorkflowEventEmitter().subscribeForConversation(
  conversationId,
  ev => send({ event: ev.type, payload: ev })
);

Every emission is paired with deps.store.createWorkflowEvent to write the event to the workflow_events table, creating a durable audit trail.

Summary

• Singleton Architecture: getWorkflowEventEmitter() ensures a single event bus across the application, with registerRun() mapping runs to conversations for scoped filtering.
• Comprehensive Event Types: The WorkflowEmitterEvent union covers the full lifecycle from workflow_started through node_failed to loop_iteration_completed.
• Fault-Tolerant Delivery: The emit() method catches and logs listener errors without blocking execution, ensuring workflow stability.
• Dual Subscription Model: Use subscribe() for system-wide monitoring or subscribeForConversation() for chat-specific streams, both returning an unsubscribe cleanup function.
• Durability: Each event is persisted to the workflow_events table via deps.store.createWorkflowEvent, providing an audit trail alongside real-time updates.

Frequently Asked Questions

What event types does the Archon workflow event emitter support?

The emitter supports a union type WorkflowEmitterEvent defined in packages/workflows/src/event-emitter.ts (lines 27-60) that includes workflow_started, workflow_completed, workflow_failed, node_started, node_completed, node_failed, tool_started, tool_completed, and loop_iteration_started/completed events, among others.

How does the emitter handle errors in listeners?

The emit() method wraps each listener invocation in a try/catch block (lines 200-208). If a listener throws, the error is logged via the internal pino logger but is not propagated to the caller, ensuring that a buggy subscriber cannot crash a running workflow.

Can I filter events for a specific conversation?

Yes. The subscribeForConversation(conversationId, listener) method (lines 223-240) registers a listener that only receives events for runs mapped to that conversation ID via the registerRun() call. This is used by the SSE adapter to stream updates to specific web clients without broadcasting sensitive data.

Where are workflow events persisted?

Immediately after emission in the executor, events are written to the workflow_events table via deps.store.createWorkflowEvent. This happens alongside the EventEmitter call, ensuring that every node_started or tool_completed event exists in the database as a permanent audit record, not just as a transient broadcast.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}