Observability

Pass an observer to .execute() to hook into lifecycle events for logging, metrics, or distributed tracing.

import { PinoFlowObserver } from '@celom/prose';
import pino from 'pino';

const logger = pino();
const observer = new PinoFlowObserver(logger);

await flow.execute(input, deps, { observer });

Observer hooks

All hooks are optional — implement only what you need.

Hook	Called when
onFlowStart(flowName, input)	Flow begins
onFlowComplete(flowName, output, duration)	Flow finishes successfully
onFlowError(flowName, error, duration)	Flow fails
onFlowBreak(flowName, breakStepName, returnValue, duration)	Flow exits early via breakIf
onStepStart(stepName, context)	Step begins
onStepComplete(stepName, result, duration, context)	Step finishes
onStepError(stepName, error, duration, context)	Step fails (after exhausting retries)
onStepRetry(stepName, attempt, maxAttempts, error)	Step is about to be retried
onStepSkipped(stepName, context)	Conditional step is skipped

Inline observer

Pass a partial object with just the hooks you need:

await flow.execute(input, deps, {
  observer: {
    onStepComplete: (name, _result, duration) =>
      console.log(`${name} took ${duration}ms`),
  },
});

Built-in observers

DefaultObserver

Console logging with an optional custom logger interface.

import { DefaultObserver } from '@celom/prose';

const observer = new DefaultObserver();
// or with a custom logger:
const observer = new DefaultObserver(myLogger);

NoOpObserver

Silent observer that does nothing. Useful as a default value.

import { NoOpObserver } from '@celom/prose';

const observer = new NoOpObserver();

PinoFlowObserver

Structured logging with any Pino-compatible logger (including Fastify’s request.log).

import { PinoFlowObserver } from '@celom/prose';
import pino from 'pino';

const observer = new PinoFlowObserver(pino());

Each hook logs a structured JSON message with the flow name, step name, duration, and relevant data — ready for indexing in Elasticsearch, Datadog, or any log aggregator.