Skip to content

feat(observability): replace console.log with diagnostics_channel, add typed subscribe helper and ai-chat events #1024

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
threepointone merged 3 commits into from
Feb 28, 2026
Merged

feat(observability): replace console.log with diagnostics_channel, add typed subscribe helper and ai-chat events #1024

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
7572a75
Use diagnostics_channel for agent observability
threepointone Feb 28, 2026
fa2b797
Use diagnostics_channel and add typed subscribe
threepointone Feb 28, 2026
cc06793
Observability: diagnostics_channel & typed events
threepointone Feb 28, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
60 changes: 60 additions & 0 deletions .changeset/diagnostics-channel-observability.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,60 @@
---
"agents": minor
---

Overhaul observability: `diagnostics_channel`, leaner events, error tracking.

### Breaking changes to `agents/observability` types

- **`BaseEvent`**: Removed `id` and `displayMessage` fields. Events now contain only `type`, `payload`, and `timestamp`. The `payload` type is now strict — accessing undeclared fields is a type error. Narrow on `event.type` before accessing payload properties.
- **`Observability.emit()`**: Removed the optional `ctx` second parameter.
- **`AgentObservabilityEvent`**: Split combined union types so each event has its own discriminant (enables proper `Extract`-based type narrowing). Added new error event types.

If you have a custom `Observability` implementation, update your `emit` signature to `emit(event: ObservabilityEvent): void`.

### diagnostics_channel replaces console.log

The default `genericObservability` implementation no longer logs every event to the console. Instead, events are published to named diagnostics channels using the Node.js `diagnostics_channel` API. Publishing to a channel with no subscribers is a no-op, eliminating logspam.

Seven named channels, one per event domain:

- `agents:state` — state sync events
- `agents:rpc` — RPC method calls and errors
- `agents:message` — message request/response/clear/cancel/error + tool result/approval
- `agents:schedule` — schedule and queue create/execute/cancel/retry/error events
- `agents:lifecycle` — connection and destroy events
- `agents:workflow` — workflow start/event/approve/reject/terminate/pause/resume/restart
- `agents:mcp` — MCP client connect/authorize/discover events

### New error events

Error events are now emitted at failure sites instead of (or alongside) `console.error`:

- `rpc:error` — RPC method failures (includes method name and error message)
- `schedule:error` — schedule callback failures after all retries exhausted
- `queue:error` — queue callback failures after all retries exhausted

### Reduced boilerplate

All 20+ inline `emit` blocks in the Agent class have been replaced with a private `_emit()` helper that auto-generates timestamps, reducing each call site from ~10 lines to 1.

### Typed subscribe helper

A new `subscribe()` function is exported from `agents/observability` with full type narrowing per channel:

```ts
import { subscribe } from "agents/observability";

const unsub = subscribe("rpc", (event) => {
// event is fully typed as rpc | rpc:error
console.log(event.payload.method);
});
```

### Tail Worker integration

In production, all diagnostics channel messages are automatically forwarded to Tail Workers via `event.diagnosticsChannelEvents` — no subscription needed in the agent itself.

### TracingChannel potential

The `diagnostics_channel` API also provides `TracingChannel` for start/end/error spans with `AsyncLocalStorage` integration, opening the door to end-to-end tracing of RPC calls, workflow steps, and schedule executions.
177 changes: 160 additions & 17 deletions docs/observability.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,44 +1,187 @@
# Observability

`Agent` instances uses the `observability` property to emit various internal events that can be used for logging and monitoring.
Agents emit structured events for every significant operation — RPC calls, state changes, schedule execution, workflow transitions, MCP connections, and more. These events are published to [diagnostics channels](https://developers.cloudflare.com/workers/runtime-apis/nodejs/diagnostics-channel/) and are silent by default (zero overhead when nobody is listening).

The default behavior is to `console.log()` the event value.
## Event structure

```
Every event has three fields:

```ts
{
displayMessage: 'State updated',
id: 'EnOzrS_tEo_8dHy5oyl8q',
payload: {},
timestamp: 1758005142787,
type: 'state:update'
type: "rpc", // what happened
payload: { method: "getWeather" }, // details
timestamp: 1758005142787 // when (ms since epoch)
}
```

This can be configured by overriding the property with an implementation of the `Observability` interface. This interface has a single `emit()` method that takes an `ObservabilityEvent`.
## Channels

Events are routed to seven named channels based on their type:

| Channel | Event types | Description |
| ------------------ | ---------------------------------------------------------------------------------------------------------------------------------------------------------------- | ----------------------------------- |
| `agents:state` | `state:update` | State sync events |
| `agents:rpc` | `rpc`, `rpc:error` | RPC method calls and failures |
| `agents:message` | `message:request`, `message:response`, `message:clear`, `message:cancel`, `message:error`, `tool:result`, `tool:approval` | Chat message and tool lifecycle |
| `agents:schedule` | `schedule:create`, `schedule:execute`, `schedule:cancel`, `schedule:retry`, `schedule:error`, `queue:retry`, `queue:error` | Scheduled and queued task lifecycle |
| `agents:lifecycle` | `connect`, `destroy` | Agent connection and teardown |
| `agents:workflow` | `workflow:start`, `workflow:event`, `workflow:approved`, `workflow:rejected`, `workflow:terminated`, `workflow:paused`, `workflow:resumed`, `workflow:restarted` | Workflow state transitions |
| `agents:mcp` | `mcp:client:preconnect`, `mcp:client:connect`, `mcp:client:authorize`, `mcp:client:discover` | MCP client operations |

## Subscribing to events

### Typed subscribe helper

The `subscribe()` function from `agents/observability` provides type-safe access to events on a specific channel:

```ts
import { subscribe } from "agents/observability";

const unsub = subscribe("rpc", (event) => {
if (event.type === "rpc") {
console.log(`RPC call: ${event.payload.method}`);
}
if (event.type === "rpc:error") {
console.error(
`RPC failed: ${event.payload.method} — ${event.payload.error}`
);
}
});

// Clean up when done
unsub();
```

The callback is fully typed — `event` is narrowed to only the event types that flow through that channel.

### Raw diagnostics_channel

You can also subscribe directly using the Node.js API:

```ts
import { subscribe } from "node:diagnostics_channel";

subscribe("agents:schedule", (event) => {
console.log(event);
});
```

## Tail Workers (production)

In production, all diagnostics channel messages are automatically forwarded to [Tail Workers](https://developers.cloudflare.com/workers/observability/tail-workers/). No subscription code is needed in the agent itself — attach a Tail Worker and access events via `event.diagnosticsChannelEvents`:

```ts
export default {
async tail(events) {
for (const event of events) {
for (const msg of event.diagnosticsChannelEvents) {
// msg.channel is "agents:rpc", "agents:workflow", etc.
// msg.message is the typed event payload
console.log(msg.timestamp, msg.channel, msg.message);
}
}
}
};
```

This gives you structured, filterable observability in production with zero overhead in the agent hot path.

## Custom observability

You can override the default implementation by providing your own `Observability` interface:

```ts
import { Agent } from "agents";
import { type Observability } from "agents/observability";
import type { Observability } from "agents/observability";

const observability: Observability = {
const myObservability: Observability = {
emit(event) {
if (event.type === "connect") {
console.log(event.timestamp, event.payload.connectionId);
// Send to your logging service, filter events, etc.
if (event.type === "rpc:error") {
myLogger.error(event.payload.method, event.payload.error);
}
}
};

class MyAgent extends Agent {
override observability = observability;
override observability = myObservability;
}
```

Or, alternatively, you can set the property to `undefined` to ignore all events.
Set `observability` to `undefined` to disable all event emission:

```ts
import { Agent } from "agents";

class MyAgent extends Agent {
override observability = undefined;
}
```

## Event reference

### RPC events

| Type | Payload | When |
| ----------- | ------------------------ | ------------------------------- |
| `rpc` | `{ method, streaming? }` | A `@callable` method is invoked |
| `rpc:error` | `{ method, error }` | A `@callable` method throws |

### State events

| Type | Payload | When |
| -------------- | ------- | ---------------------- |
| `state:update` | `{}` | `setState()` is called |

### Message and tool events (`AIChatAgent`)

These events are emitted by `AIChatAgent` from `@cloudflare/ai-chat`. They track the chat message lifecycle, including client-side tool interactions.

| Type | Payload | When |
| ------------------ | -------------------------- | ----------------------------------- |
| `message:request` | `{}` | A chat message is received |
| `message:response` | `{}` | A chat response stream completes |
| `message:clear` | `{}` | Chat history is cleared |
| `message:cancel` | `{ requestId }` | A streaming request is cancelled |
| `message:error` | `{ error }` | A chat stream fails |
| `tool:result` | `{ toolCallId, toolName }` | A client tool result is received |
| `tool:approval` | `{ toolCallId, approved }` | A tool call is approved or rejected |

### Schedule and queue events

| Type | Payload | When |
| ------------------ | ---------------------------------------- | -------------------------------------------- |
| `schedule:create` | `{ callback, id }` | A schedule is created |
| `schedule:execute` | `{ callback, id }` | A scheduled callback starts |
| `schedule:cancel` | `{ callback, id }` | A schedule is cancelled |
| `schedule:retry` | `{ callback, id, attempt, maxAttempts }` | A scheduled callback is retried |
| `schedule:error` | `{ callback, id, error, attempts }` | A scheduled callback fails after all retries |
| `queue:retry` | `{ callback, id, attempt, maxAttempts }` | A queued callback is retried |
| `queue:error` | `{ callback, id, error, attempts }` | A queued callback fails after all retries |

### Lifecycle events

| Type | Payload | When |
| --------- | ------------------ | ------------------------------------- |
| `connect` | `{ connectionId }` | A WebSocket connection is established |
| `destroy` | `{}` | The agent is destroyed |

### Workflow events

| Type | Payload | When |
| --------------------- | ------------------------------- | ------------------------------ |
| `workflow:start` | `{ workflowId, workflowName? }` | A workflow instance is started |
| `workflow:event` | `{ workflowId, eventType? }` | An event is sent to a workflow |
| `workflow:approved` | `{ workflowId, reason? }` | A workflow is approved |
| `workflow:rejected` | `{ workflowId, reason? }` | A workflow is rejected |
| `workflow:terminated` | `{ workflowId, workflowName? }` | A workflow is terminated |
| `workflow:paused` | `{ workflowId, workflowName? }` | A workflow is paused |
| `workflow:resumed` | `{ workflowId, workflowName? }` | A workflow is resumed |
| `workflow:restarted` | `{ workflowId, workflowName? }` | A workflow is restarted |

### MCP events

| Type | Payload | When |
| ----------------------- | --------------------------------------- | -------------------------------------------- |
| `mcp:client:preconnect` | `{ serverId }` | Before connecting to an MCP server |
| `mcp:client:connect` | `{ url, transport, state, error? }` | An MCP connection attempt completes or fails |
| `mcp:client:authorize` | `{ serverId, authUrl, clientId? }` | An MCP OAuth flow begins |
| `mcp:client:discover` | `{ url?, state?, error?, capability? }` | MCP capability discovery succeeds or fails |
2 changes: 0 additions & 2 deletions packages/agents/src/e2e-tests/worker.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,8 +29,6 @@ export type SlowFiberSnapshot = {
const FiberAgent = withFibers(Agent, { debugFibers: true });

export class FiberTestAgent extends FiberAgent<Record<string, unknown>> {
observability = undefined;

/**
* A slow fiber that takes ~1 second per step.
* Checkpoints after each step.
Expand Down
Loading
Loading