Event System
Explore asynchronous event dispatching, bail/waterfall interceptors, and Harness telemetry events.
During autonomous agent reasoning and tool execution, state transitions occur at high frequency—from session creation and prompt assembly, to streaming inference, tool interception, and Trajectory persistence.
Traditional global EventEmitter singletons frequently suffer from listener duplication and memory leaks during hot reloads. DeepSeek Harness leverages the Cordis event bus to provide scoped lifecycle tracking, short-circuit guard gates (Bail), and sequential pipeline transformations (Waterfall).
This tutorial uses a “Command Security Interceptor & Prompt Sanitizer” as a real-world case study to explain the three event dispatch paradigms in Harness.
Core Architecture: Three Event Dispatch Paradigms in Cordis
The Cordis event bus provides three distinct execution models based on operational intent:
┌─────────────────────────────────────────────────────────────┐
│ Cordis Microkernel Event Modes │
├─────────────────────┬───────────────────────────────────────┤
│ 1. emit (Broadcast) │ Notifies all listeners concurrently; │
│ │ ignores return values (Telemetry/Logs)│
├─────────────────────┼───────────────────────────────────────┤
│ 2. bail (Guard Gate)│ Invokes listeners sequentially; halts │
│ │ on first non-undefined value (Security)│
├─────────────────────┼───────────────────────────────────────┤
│ 3. waterfall (Pipe) │ Pipes output of one listener as input │
│ │ to the next (Prompt Sanitization) │
└─────────────────────┴───────────────────────────────────────┘
Step 1: Implement the Security Interceptor Plugin
Create custom-plugins/security-interceptor/src/index.ts:
import type { Context } from '@deepseek-ai/cordis';
export const name = 'security-interceptor-plugin';
export function apply(ctx: Context) {
// 1. [Bail Guard]: Intercept tools prior to execution
ctx.on('tool/before-execute', (call) => {
// Block dangerous recursive deletion commands
if (call.name === 'bash' && call.args?.command?.includes('rm -rf /')) {
console.warn(`[Security] Blocked dangerous command: ${call.args.command}`);
return {
blocked: true,
reason: 'Security policy forbids destructive global deletion commands'
};
}
// Returning undefined allows execution to proceed to the next listener
});
// 2. [Waterfall Pipeline]: Sanitize prompt text before LLM transmission
ctx.on('prompt/transform', (inputPrompt: string) => {
// Mask plaintext phone numbers
const sanitized = inputPrompt.replace(/(\d{3})\d{4}(\d{4})/g, '$1****$2');
return sanitized;
});
// 3. [Emit Broadcast]: Fire-and-forget telemetry logger
ctx.on('trajectory/step', (step) => {
console.log(`[AuditLog] Persisted session step #${step.index}, duration: ${step.durationMs}ms`);
});
}
Step 2: Emitting Custom Events in Domain Plugins
Custom adapter or workflow plugins can emit events onto the microkernel event bus:
import type { Context } from '@deepseek-ai/cordis';
export function runWorkflow(ctx: Context, taskPayload: any) {
// 1. Bail check: halt workflow if blocked by any security plugin
const guardResult = ctx.bail('workflow/before-run', taskPayload);
if (guardResult?.blocked) {
console.error(`[Workflow] Task blocked: ${guardResult.reason}`);
return;
}
// 2. Waterfall transformation: pipe payload through transformation filters
const finalPayload = ctx.waterfall('workflow/transform-payload', taskPayload);
// 3. Emit broadcast notification upon completion
ctx.emit('workflow/completed', { id: taskPayload.id, timestamp: Date.now() });
}
Built-in Harness Lifecycle Events Quick Reference
┌─────────────────────────────────────────────────────────────┐
│ Harness Core Lifecycle Event Flow │
├─────────────────────────────────────────────────────────────┤
│ • session/before-turn : Emitted before each reasoning turn│
│ • prompt/transform : Stream pipeline for Prompt filters │
│ • llm/stream-chunk : Token chunk received from LLM │
│ • tool/before-execute : Guard gate prior to tool execution │
│ • tool/after-execute : Triggered after tool returns data │
│ • session/after-turn : Emitted after turn trajectory saved│
└─────────────────────────────────────────────────────────────┘
Frequently Asked Questions (FAQ)
Q1: Why is manual removeListener cleanup unnecessary in Cordis?
Answer: ctx.on binds listeners to the plugin's dedicated Context scope. When the plugin unloads or hot reloads, the microkernel unbinds all attached listeners automatically, eliminating Node.js MaxListenersExceededWarning memory leaks.
Q2: What is the listener execution order in ctx.bail?
Answer: Listeners execute sequentially in the order plugins were mounted. The moment a listener returns any value other than undefined, the dispatch loop short-circuits immediately.
Q3: How does ctx.waterfall handle asynchronous promises?
Answer: ctx.waterfall natively awaits Promises. When an async listener resolves, its result is passed as the input argument to the next listener, making it ideal for multi-stage async prompt filters.
Q4: Can an event listener emit another event?
Answer: Yes. However, be cautious to avoid circular "Event Ping-Pong" loops. For complex state machines, manage shared state using a dedicated Service class.