> ## Documentation Index
> Fetch the complete documentation index at: https://docs.doclo.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Observability

> Hooks, metrics, and distributed tracing for flow execution

The Doclo SDK provides comprehensive observability through lifecycle hooks, metrics aggregation, and cloud integration. Monitor every step of your document processing pipelines.

## Local Observability

Configure observability hooks when creating a flow:

```typescript theme={null}
import { createFlow, extract } from '@doclo/flows';

const flow = createFlow({
  observability: {
    onFlowStart: (ctx) => {
      console.log(`Flow ${ctx.flowId} started`);
    },
    onFlowEnd: (ctx) => {
      console.log(`Flow completed in ${ctx.duration}ms`);
      console.log(`Cost: $${ctx.stats?.totalCostUSD.toFixed(4)}`);
    },
    onFlowError: (ctx) => {
      console.error(`Flow failed: ${ctx.error.message}`);
    },
    onStepStart: (ctx) => {
      console.log(`Step ${ctx.stepId} starting...`);
    },
    onStepEnd: (ctx) => {
      console.log(`Step ${ctx.stepId}: ${ctx.duration}ms, $${ctx.cost?.toFixed(4)}`);
    },
    onStepError: (ctx) => {
      console.error(`Step ${ctx.stepId} failed: ${ctx.error.message}`);
    }
  }
})
  .step('extract', extract({ provider, schema }))
  .build();
```

## Available Hooks

### Flow-Level Hooks

| Hook          | Trigger                     | Context                                       |
| ------------- | --------------------------- | --------------------------------------------- |
| `onFlowStart` | Flow execution begins       | `flowId`, `executionId`, `input`, `metadata`  |
| `onFlowEnd`   | Flow completes successfully | `duration`, `output`, `stats`, `traceContext` |
| `onFlowError` | Flow fails                  | `error`, `errorCode`, `failedAtStepIndex`     |

### Step-Level Hooks

| Hook          | Trigger                     | Context                                                |
| ------------- | --------------------------- | ------------------------------------------------------ |
| `onStepStart` | Step execution begins       | `stepId`, `stepIndex`, `stepType`, `provider`, `model` |
| `onStepEnd`   | Step completes successfully | `duration`, `output`, `usage`, `cost`                  |
| `onStepError` | Step fails                  | `error`, `willRetry`, `retryAttempt`                   |

### Consensus Hooks

| Hook                     | Trigger                  | Context                                  |
| ------------------------ | ------------------------ | ---------------------------------------- |
| `onConsensusStart`       | Consensus voting begins  | `runsPlanned`, `strategy`                |
| `onConsensusRunComplete` | Individual run completes | `runIndex`, `output`, `status`           |
| `onConsensusComplete`    | All runs complete        | `agreement`, `agreedOutput`, `totalCost` |

### Batch/forEach Hooks

| Hook             | Trigger               | Context                          |
| ---------------- | --------------------- | -------------------------------- |
| `onBatchStart`   | forEach begins        | `totalItems`, `batchId`          |
| `onBatchItemEnd` | Single item processed | `itemIndex`, `result`, `status`  |
| `onBatchEnd`     | All items processed   | `successfulItems`, `failedItems` |

### Provider-Level Hooks

| Hook                 | Trigger               | Context                                     |
| -------------------- | --------------------- | ------------------------------------------- |
| `onProviderRequest`  | API request sent      | `provider`, `model`, `input`                |
| `onProviderResponse` | API response received | `output`, `usage`, `cost`, `httpStatusCode` |
| `onProviderRetry`    | Request will retry    | `error`, `attemptNumber`, `nextRetryDelay`  |

## Cloud Observability

Send execution events to the Doclo Cloud dashboard for monitoring and analytics.

### Using createCloudObservability

```typescript theme={null}
import { createCloudObservability } from '@doclo/client';
import { createFlow, extract } from '@doclo/flows';

// Create cloud observability transport
const cloudObs = createCloudObservability({
  client: docloClient,
  flowId: 'invoice-extraction',
  flowVersion: '1.0.0',
  mode: 'stream',           // Real-time events
  flushIntervalMs: 5000,    // Flush every 5 seconds
  includeInputs: false,     // Privacy: exclude input data
  includeOutputs: true      // Include extraction results
});

// Create flow with cloud observability
const flow = createFlow({ observability: cloudObs })
  .step('extract', extract({ provider, schema }))
  .build();

// Run the flow
const result = await flow.run(input);

// Ensure all events are sent
await cloudObs.flush();
```

### Configuration Options

| Option            | Type                         | Default    | Description                     |
| ----------------- | ---------------------------- | ---------- | ------------------------------- |
| `client`          | `DocloClient`                | Required   | Doclo client instance           |
| `flowId`          | `string`                     | Required   | Flow ID for tracking            |
| `flowVersion`     | `string`                     | -          | Flow version                    |
| `mode`            | `'stream' \| 'batch-at-end'` | `'stream'` | When to send events             |
| `flushIntervalMs` | `number`                     | `5000`     | Flush interval (0 = immediate)  |
| `batchSize`       | `number`                     | `50`       | Events before auto-flush        |
| `maxRetries`      | `number`                     | `3`        | Retry attempts for failed sends |
| `retryDelayMs`    | `number`                     | `1000`     | Base retry delay                |
| `maxBufferSize`   | `number`                     | `1000`     | Max buffered events             |
| `includeInputs`   | `boolean`                    | `false`    | Include input data in events    |
| `includeOutputs`  | `boolean`                    | `false`    | Include output data in events   |
| `onError`         | `function`                   | -          | Called when events are dropped  |

### Transport Modes

#### Stream Mode (Default)

Events are sent periodically during execution for real-time dashboard updates:

```typescript theme={null}
createCloudObservability({
  client,
  flowId: 'my-flow',
  mode: 'stream',
  flushIntervalMs: 5000,  // Flush every 5 seconds
  batchSize: 50           // Or when 50 events buffered
});
```

Set `flushIntervalMs: 0` for immediate per-event sending:

```typescript theme={null}
createCloudObservability({
  client,
  flowId: 'my-flow',
  mode: 'stream',
  flushIntervalMs: 0  // Send immediately
});
```

#### Batch-at-End Mode

Events are collected and sent only when `flush()` is called:

```typescript theme={null}
const obs = createCloudObservability({
  client,
  flowId: 'my-flow',
  mode: 'batch-at-end'
});

await flow.run(input);
await obs.flush();  // Send all events at once
```

### Privacy Controls

Control what data is sent to the cloud:

```typescript theme={null}
createCloudObservability({
  client,
  flowId: 'my-flow',
  includeInputs: false,   // Don't send document content
  includeOutputs: true    // Send extraction results
});
```

### Error Handling

Handle dropped events and transport errors:

```typescript theme={null}
createCloudObservability({
  client,
  flowId: 'my-flow',
  onError: (error, droppedCount) => {
    console.error(`Dropped ${droppedCount} events:`, error.message);
  }
});
```

## Metrics

Every flow execution returns aggregated metrics:

```typescript theme={null}
const result = await flow.run(input);

console.log('Metrics:', result.aggregated);
// {
//   totalDurationMs: 2500,
//   totalCostUSD: 0.0042,
//   totalInputTokens: 1500,
//   totalOutputTokens: 200,
//   stepCount: 2
// }

// Per-step metrics
console.log('Step metrics:', result.metrics);
// [
//   { step: 'parse', ms: 1200, costUSD: 0.0010 },
//   { step: 'extract', ms: 1300, costUSD: 0.0032 }
// ]
```

## Distributed Tracing

The SDK supports W3C trace context for distributed tracing:

```typescript theme={null}
const flow = createFlow({
  observability: {
    onFlowStart: (ctx) => {
      // W3C trace context available
      console.log('Trace ID:', ctx.traceContext.traceId);
      console.log('Span ID:', ctx.traceContext.spanId);
    }
  }
});
```

### Propagating Trace Context

Pass trace context from upstream services:

```typescript theme={null}
const result = await flow.run(input, {
  metadata: {
    traceParent: req.headers['traceparent']
  }
});
```

## Combining Local and Cloud

Use both local logging and cloud observability:

```typescript theme={null}
const cloudObs = createCloudObservability({
  client,
  flowId: 'my-flow'
});

const flow = createFlow({
  observability: {
    ...cloudObs,
    // Add local logging on top of cloud
    onStepEnd: (ctx) => {
      cloudObs.onStepEnd?.(ctx);
      console.log(`Step ${ctx.stepId}: ${ctx.duration}ms`);
    }
  }
});
```

## Next Steps

<CardGroup cols={2}>
  <Card title="Hybrid Client" icon="network-wired" href="/cloud/hybrid-client">
    Local execution with cloud observability
  </Card>

  <Card title="Error Handling" icon="triangle-exclamation" href="/sdk/advanced/error-handling">
    Handle failures gracefully
  </Card>
</CardGroup>