Wiring Agents Into Event Streams Without Losing the Wheel: Control Plane Patterns for Agentic Event-Driven Systems

Addy Osmani's two-axis framing — agency (how independently the agent decides) and orchestration (how much it coordinates with others) — gives platform teams a practical vocabulary for annotating each event type in their schema registry. In practice, this means a 'payment.initiated' event might land at high agency but low orchestration (one agent, full authority), while a 'fraud.flagged' event lands at low agency and high orchestration (multi-agent review, human sign-off before any action). That classification then drives your routing topology: events flow to different handler queues, each with different confirmation and escalation hooks pre-wired. Source: [Agentic Autonomy Levels] — Addy Osmani (https://addyosmani.com/blog/agentic-autonomy-levels/)

AIThe connection to event-driven architecture is direct: the event schema becomes the contract, and the autonomy classification becomes a field in that schema — not metadata, but a first-class routing attribute. This prevents the common failure mode where a newly deployed agent silently handles event categories it was never explicitly cleared to handle autonomously, because the bus itself rejects mismatched autonomy claims.

// Event schema with autonomy classification as a first-class field
interface AgentRoutedEvent<T = unknown> {
  eventId: string;
  eventType: string;
  occurredAt: string;
  payload: T;
  autonomy: {
    agencyLevel: 1 | 2 | 3 | 4 | 5 | 6; // Osmani scale
    orchestrationLevel: 1 | 2 | 3 | 4 | 5 | 6;
    requiresHumanApproval: boolean;
    maxRetryWithoutEscalation: number;
  };
}

// Router that enforces the contract at dispatch time
async function routeToHandler(
  event: AgentRoutedEvent,
  handlers: HandlerRegistry
): Promise<void> {
  const { agencyLevel, requiresHumanApproval } = event.autonomy;

  if (requiresHumanApproval) {
    await handlers.humanInLoopQueue.enqueue(event);
    return;
  }

  if (agencyLevel >= 4) {
    await handlers.supervisedMultiAgent.dispatch(event);
    return;
  }

  await handlers.directAgent.process(event);
}

The ByteByteGo agent loop breakdown makes the observe-reason-act cycle explicit, but the engineering discipline is in deciding how much internal state leaks out of each cycle as structured data. In an event-driven context, each loop iteration is consuming one or more events from upstream and potentially emitting one or more actions downstream — the loop itself becomes a processing node in your topology, and it needs the same instrumentation discipline you'd apply to any other consumer. That means span IDs, parent event IDs, tool call records, and a final 'action committed' or 'action deferred' status all flowing to your observability backend on every iteration. Source: [The Agent Loop: How AI Goes From Answering Questions to Doing Things] — ByteByteGo (https://blog.bytebytego.com/p/the-agent-loop-how-ai-goes-from-answering)

AIThis connects directly to the per-phase provenance logging trend: the agent loop is the natural unit of provenance, and each phase boundary is the natural emission point. If you're using something like OpenTelemetry already for your event consumers, the agent loop becomes another instrumented span tree — which means your existing dashboards can surface agent reasoning latency, decision frequency by event type, and tool call failure rates without any bespoke tooling.

// Phase-emitting agent loop wrapper in TypeScript
type LoopPhase = 'OBSERVE' | 'ORIENT' | 'DECIDE' | 'ACT';

interface PhaseRecord {
  loopId: string;
  eventId: string;
  phase: LoopPhase;
  timestamp: string;
  data: Record<string, unknown>;
}

class ObservableAgentLoop {
  constructor(
    private readonly emit: (record: PhaseRecord) => Promise<void>,
    private readonly agent: AgentCore
  ) {}

  async process(event: AgentRoutedEvent): Promise<void> {
    const loopId = crypto.randomUUID();
    const base = { loopId, eventId: event.eventId };

    await this.emit({ ...base, phase: 'OBSERVE', timestamp: now(), data: { eventType: event.eventType } });

    const orientation = await this.agent.orient(event);
    await this.emit({ ...base, phase: 'ORIENT', timestamp: now(), data: { toolsConsidered: orientation.tools } });

    const decision = await this.agent.decide(orientation);
    await this.emit({ ...base, phase: 'DECIDE', timestamp: now(), data: { action: decision.action, confidence: decision.confidence } });

    const result = await this.agent.act(decision);
    await this.emit({ ...base, phase: 'ACT', timestamp: now(), data: { status: result.status, actionId: result.actionId } });
  }
}

const now = () => new Date().toISOString();

Huyen Chip's GenAI platform decomposition makes the layered components explicit — input guardrails, context enrichment, model gateway, output validation — but in an event-driven integration, each of those layers needs to be treated as a consumer-producer pair, not a middleware chain. A guardrail that evaluates an event payload and rejects it must emit a 'guardrail.rejected' event with a structured reason code, because that rejection is a platform-level decision that downstream monitoring, compliance, and on-call engineers need to see in real time. Without that, your event stream gives you throughput metrics but no visibility into how often the platform is self-censoring agent actions, which is exactly the kind of invisible governance gap that causes audit failures. Source: [Building A Generative AI Platform] — Huyen Chip (https://huyenchip.com//2024/07/25/genai-platform.html)

AIThis connects to the draft-then-verify trend: in an event-driven agent system, 'draft' maps to the agent emitting a proposed action event, and 'verify' maps to the guardrail consumer evaluating that event before the action consumer is allowed to process it. The event bus becomes the verification protocol — which means you get audit trails, retry semantics, and dead-letter handling for verification failures essentially for free if you model the pattern correctly from the start.

// Guardrail as an event-stream filter node — emits rejection events rather than silently dropping
async function guardrailFilterNode(
  event: AgentRoutedEvent,
  producer: EventProducer
): Promise<AgentRoutedEvent | null> {
  const evaluation = await evaluateGuardrails(event);

  if (!evaluation.passed) {
    await producer.emit({
      eventType: 'guardrail.rejected',
      correlationId: event.eventId,
      timestamp: new Date().toISOString(),
      payload: {
        originalEventType: event.eventType,
        ruleId: evaluation.failedRuleId,
        reason: evaluation.reason,
        severity: evaluation.severity
      }
    });
    return null; // downstream consumer never sees this event
  }

  return event;
}

The Antithesis hostile simulation approach — running systems in an adversarial environment to surface correctness bugs — maps directly onto agentic event-driven testing in a way that's underappreciated. An agent loop consuming from an event stream has temporal dependencies, external tool calls, and LLM non-determinism stacked on top of each other, which means the state space of possible failures is orders of magnitude larger than a conventional microservice. Running the full topology under adversarial event injection — out-of-order delivery, schema drift, backpressure spikes, tool call timeouts — is the only way to build empirical confidence that your governance mechanisms (guardrails, escalation hooks, dead-letter handlers) actually fire under realistic conditions. Source: [The Pragmatic Engineer AMA] — Pragmatic Engineer (https://newsletter.pragmaticengineer.com/p/the-pragmatic-engineer-ama)

Lilian Weng's harness engineering framing adds a useful lens here: the test harness around an agent loop is itself an engineered artifact with correctness properties, not just a test runner. For event-driven agents, that means the harness needs to simulate the event bus, inject adversarial events with controlled timing, capture every phase emission from the agent loop, and assert on the governance record — not just the final output. That's a non-trivial build, but it's the foundation that makes the rest of your observability and governance work verifiable. Source: [Harness Engineering for Self-Improvement] — Lilian Weng (https://lilianweng.github.io/posts/2026-07-04-harness/)

// Adversarial event sequence builder for agent loop integration tests
type AdversarialScenario = 'out-of-order' | 'duplicate-flood' | 'schema-drift' | 'tool-timeout';

function buildAdversarialSequence(
  baseEvent: AgentRoutedEvent,
  scenario: AdversarialScenario
): AgentRoutedEvent[] {
  switch (scenario) {
    case 'duplicate-flood':
      return Array.from({ length: 50 }, (_, i) => ({
        ...baseEvent,
        eventId: i === 0 ? baseEvent.eventId : crypto.randomUUID(), // first is legit, rest are dupes
        occurredAt: new Date(Date.now() + i * 10).toISOString()
      }));

    case 'out-of-order':
      const events = buildChronologicalSequence(baseEvent);
      return events.sort(() => Math.random() - 0.5); // shuffle delivery order

    case 'schema-drift':
      return [{ ...baseEvent, payload: { unexpected_field: true } as unknown }];

    case 'tool-timeout':
      return [{ ...baseEvent, autonomy: { ...baseEvent.autonomy, maxRetryWithoutEscalation: 0 } }];

    default:
      return [baseEvent];
  }
}

The etcd v3.7 release removes the legacy v2store entirely and ships RangeStream, which allows a single watch connection to stream changes across a key prefix rather than requiring per-key watchers. In an agentic event-driven system, this is directly useful for supervisor nodes that need to react when any agent in a pool changes its state — for example, when an agent escalates an event to human review, the supervisor needs to pick that up immediately, not on the next polling cycle. With RangeStream, you can maintain a key prefix like `/agents/{poolId}/state/` and have a single supervisor watch stream that fires on any transition, with the key name encoding the agent ID and the value encoding the current phase and event ID being processed. Source: [Announcing etcd v3.7.0] — SIG etcd / Kubernetes.io (https://kubernetes.io/blog/2026/07/08/announcing-etcd-3.7/)

AIThe protobuf overhaul in v3.7 also reduces serialization overhead for high-frequency watch events, which matters in agent pools processing event streams at scale — each agent loop phase transition becomes a write to etcd, and at hundreds of concurrent agents that's a meaningful throughput concern. The key architectural decision is scoping what goes into etcd versus what goes into your observability stream: etcd handles live coordination state (who owns what, what phase is in flight, what locks are held), while your observability stream gets the immutable provenance record. These are complementary, not redundant.

// Supervisor using etcd RangeStream to watch all agents in a pool
import { Etcd3 } from 'etcd3';

const client = new Etcd3();
const POOL_PREFIX = '/agents/pool1/state/';

async function superviseAgentPool(): Promise<void> {
  const watcher = await client
    .watch()
    .prefix(POOL_PREFIX)
    .create();

  watcher.on('put', (kv) => {
    const agentId = kv.key.toString().replace(POOL_PREFIX, '');
    const state: AgentStateRecord = JSON.parse(kv.value.toString());

    if (state.phase === 'ESCALATED') {
      handleEscalation({ agentId, loopId: state.loopId, eventId: state.eventId });
    }

    if (isStalled(state)) {
      triggerReassignment({ agentId, stalledSince: state.lastUpdated });
    }
  });
}

function isStalled(state: AgentStateRecord): boolean {
  const stallThresholdMs = 30_000;
  return Date.now() - new Date(state.lastUpdated).getTime() > stallThresholdMs;
}