Agents in the Stream: How to Keep Visibility and Control When Agentic Workflows Touch Your Event Bus

// TypeScript: Event surface contract adapter
import { z } from 'zod';
import { v4 as uuidv4 } from 'uuid';

const BillingEventSchema = z.object({
  customerId: z.string().uuid(),
  amount: z.number().positive(),
  currency: z.string().length(3),
});

type BoundaryAuditEntry = {
  traceId: string;
  direction: 'inbound' | 'outbound';
  topic: string;
  valid: boolean;
  timestamp: string;
  payload?: unknown;
};

export function createEventSurfaceContract(
  allowedTopics: string[],
  auditLog: (entry: BoundaryAuditEntry) => Promise<void>
) {
  return {
    async receive(topic: string, raw: unknown, traceId = uuidv4()) {
      if (!allowedTopics.includes(topic)) {
        throw new Error(`Topic '${topic}' not in agent surface contract`);
      }
      const result = BillingEventSchema.safeParse(raw);
      await auditLog({ traceId, direction: 'inbound', topic, valid: result.success, timestamp: new Date().toISOString() });
      if (!result.success) throw new Error(`Schema violation: ${result.error.message}`);
      return { ...result.data, _traceId: traceId };
    },
    async emit(topic: string, payload: unknown, traceId: string) {
      if (!allowedTopics.includes(topic)) {
        throw new Error(`Agent attempted to emit to disallowed topic: '${topic}'`);
      }
      const validated = BillingEventSchema.safeParse(payload);
      await auditLog({ traceId, direction: 'outbound', topic, valid: validated.success, timestamp: new Date().toISOString(), payload });
      if (!validated.success) throw new Error(`Outbound schema violation on topic '${topic}'`);
      return { topic, payload: validated.data, traceId };
    }
  };
}

// TypeScript: Emit topology snapshot and delta as structured events
type AgentNode = { id: string; toolTopics: string[] };
type TopologyEvent =
  | { type: 'topology-snapshot'; flowId: string; traceId: string; graph: AgentNode[]; timestamp: string }
  | { type: 'topology-delta'; flowId: string; traceId: string; added: AgentNode[]; removed: AgentNode[]; timestamp: string };

export class FlowTopologyTracker {
  private graph: AgentNode[];
  constructor(
    private flowId: string,
    private traceId: string,
    private emitEvent: (e: TopologyEvent) => Promise<void>
  ) {
    this.graph = [];
  }

  async initialize(nodes: AgentNode[]) {
    this.graph = nodes;
    await this.emitEvent({
      type: 'topology-snapshot',
      flowId: this.flowId,
      traceId: this.traceId,
      graph: nodes,
      timestamp: new Date().toISOString()
    });
  }

  async addNode(node: AgentNode) {
    this.graph.push(node);
    await this.emitEvent({
      type: 'topology-delta',
      flowId: this.flowId,
      traceId: this.traceId,
      added: [node],
      removed: [],
      timestamp: new Date().toISOString()
    });
  }
}

// TypeScript: Subtask grader for agentic event handler evals
type EvalStage = 'parse' | 'schema' | 'topic-access' | 'semantic';

type StageResult = {
  stage: EvalStage;
  passed: boolean;
  score: number; // 0.0 - 1.0
  detail?: string;
};

type EvalResult = {
  totalScore: number;
  stages: StageResult[];
};

const STAGE_WEIGHTS: Record<EvalStage, number> = {
  parse: 0.2,
  schema: 0.3,
  'topic-access': 0.3,
  semantic: 0.2,
};

export function gradeEventHandlerOutput(
  agentOutput: unknown,
  expectedTopic: string,
  actualTopic: string,
  validateSchema: (v: unknown) => boolean,
  validateSemantics: (v: unknown) => boolean
): EvalResult {
  const stages: StageResult[] = [
    {
      stage: 'parse',
      passed: agentOutput !== null && agentOutput !== undefined,
      score: agentOutput != null ? 1.0 : 0.0,
    },
    {
      stage: 'schema',
      passed: validateSchema(agentOutput),
      score: validateSchema(agentOutput) ? 1.0 : 0.0,
      detail: validateSchema(agentOutput) ? undefined : 'Schema validation failed'
    },
    {
      stage: 'topic-access',
      passed: actualTopic === expectedTopic,
      score: actualTopic === expectedTopic ? 1.0 : 0.0,
      detail: actualTopic !== expectedTopic ? `Emitted to '${actualTopic}', expected '${expectedTopic}'` : undefined
    },
    {
      stage: 'semantic',
      passed: validateSemantics(agentOutput),
      score: validateSemantics(agentOutput) ? 1.0 : 0.0,
    }
  ];

  const totalScore = stages.reduce(
    (acc, s) => acc + s.score * STAGE_WEIGHTS[s.stage],
    0
  );

  return { totalScore, stages };
}

// TypeScript: Graceful shutdown handler for a stateful agent event processor
import { EventEmitter } from 'events';

type TopologyEventEmitter = (event: { type: string; processorId: string; timestamp: string }) => Promise<void>;

export function registerAgentProcessorShutdownHook(
  processorId: string,
  consumer: { pause: () => void; commitOffsets: () => Promise<void> },
  emitTopologyEvent: TopologyEventEmitter,
  inFlightTracker: { waitForDrain: () => Promise<void> }
) {
  const shutdown = async (signal: string) => {
    console.log(`[${processorId}] Received ${signal} — beginning graceful drain`);

    // Stop accepting new events
    consumer.pause();

    // Notify observability layer that this processor is draining
    await emitTopologyEvent({
      type: 'processor-draining',
      processorId,
      timestamp: new Date().toISOString()
    });

    // Wait for in-flight agent tasks to complete
    await inFlightTracker.waitForDrain();

    // Commit offsets only after all in-flight tasks are settled
    await consumer.commitOffsets();

    await emitTopologyEvent({
      type: 'processor-stopped',
      processorId,
      timestamp: new Date().toISOString()
    });

    process.exit(0);
  };

  process.on('SIGTERM', () => shutdown('SIGTERM'));
  process.on('SIGINT', () => shutdown('SIGINT'));
}

// TypeScript: Trust threshold circuit breaker for agent event output
type MetricWindow = { violations: number; total: number; windowMs: number };

export class AgentTrustCircuitBreaker {
  private violations = 0;
  private total = 0;
  private windowStart = Date.now();

  constructor(
    private readonly thresholdRate: number, // e.g. 0.02 for 2%
    private readonly windowMs: number,       // e.g. 300_000 for 5 min
    private readonly onThresholdBreached: (window: MetricWindow) => Promise<void>
  ) {}

  async record(isViolation: boolean): Promise<'auto-approve' | 'escalate'> {
    const now = Date.now();

    // Reset window if expired
    if (now - this.windowStart > this.windowMs) {
      this.violations = 0;
      this.total = 0;
      this.windowStart = now;
    }

    this.total++;
    if (isViolation) this.violations++;

    const rate = this.total > 0 ? this.violations / this.total : 0;

    if (rate > this.thresholdRate && this.total >= 10) {
      await this.onThresholdBreached({
        violations: this.violations,
        total: this.total,
        windowMs: this.windowMs
      });
      return 'escalate';
    }

    return 'auto-approve';
  }

  currentRate(): number {
    return this.total > 0 ? this.violations / this.total : 0;
  }
}