Observability¶

The FCC observability layer provides structured tracing, metrics collection, and exporters for monitoring workflow executions, simulation runs, and action engine operations. It is OTel-compatible when OpenTelemetry is installed, and falls back to a built-in implementation when it is not.

flowchart TD
    subgraph IC["Instrumented Code"]
        SE[SimulationEngine] -->|"@traced"| FT[FccTracer]
        AE[ActionEngine] -->|"@traced"| FT
        SE -->|"record_*"| FM[FccMetrics]
        AE -->|"record_*"| FM
    end
    subgraph TP["Tracing Pipeline"]
        FT -->|"span context"| SC[SpanContext]
        SC -->|end| SD[SpanData]
    end
    subgraph MPG["Metrics Pipeline"]
        FM -->|"observe / increment"| MP[MetricPoint]
    end
    subgraph EX["Export"]
        SD --> CSE[ConsoleSpanExporter]
        SD --> JSE[JsonFileSpanExporter]
        SD -.->|optional| OTel[OpenTelemetry SDK]
        MP --> CME[ConsoleMetricExporter]
        MP --> JME[JsonFileMetricExporter]
    end

Architecture Overview¶

graph TD
    FccTracer --> SpanData
    FccMetrics --> MetricPoint
    SpanData --> SpanExporter
    MetricPoint --> MetricExporter
    SpanExporter --> ConsoleSpanExporter
    SpanExporter --> JsonFileSpanExporter
    MetricExporter --> ConsoleMetricExporter
    MetricExporter --> JsonFileMetricExporter

SpanData¶

SpanData is a frozen dataclass representing a completed span (an immutable snapshot):

Field	Type	Description
`span_id`	`str`	Unique identifier for this span
`trace_id`	`str`	Links all spans in a trace
`name`	`str`	Human-readable span name
`start_time`	`str`	ISO 8601 start timestamp
`parent_span_id`	`str \| None`	Parent span ID (None for root spans)
`end_time`	`str \| None`	ISO 8601 end timestamp (None if active)
`status`	`str`	`"ok"` or `"error"`
`attributes`	`dict[str, Any]`	Arbitrary key-value metadata

The duration_ms property calculates elapsed time:

span = tracer.completed_spans[0]
print(span.duration_ms)  # e.g., 42.5 (milliseconds)

SpanContext¶

SpanContext is the mutable counterpart used during span execution. It allows setting attributes and status before the span is finalized:

ctx.set_attribute("scenario_id", "GEN-001")
ctx.set_status("error")
span_data = ctx.end()  # Returns frozen SpanData

FccTracer¶

The FccTracer creates and manages spans for FCC operations. It uses OpenTelemetry when available and falls back to an internal implementation.

Creating Spans¶

from fcc.observability.tracing import FccTracer

tracer = FccTracer(service_name="fcc")

with tracer.span("simulation.run") as ctx:
    ctx.set_attribute("scenario_id", "GEN-001")
    ctx.set_attribute("mode", "mock")
    # ... do work ...

Nested Spans¶

Spans can be nested. Child spans automatically get the current span as their parent:

with tracer.span("workflow.execute") as parent:
    parent.set_attribute("workflow_id", "base_sequence")

    with tracer.span("node.process") as child:
        child.set_attribute("node_id", "n1")
        # child.parent_span_id == parent.span_id

    with tracer.span("node.process") as child2:
        child2.set_attribute("node_id", "n2")

Accessing Completed Spans¶

spans = tracer.completed_spans  # Returns list[SpanData]
tracer.clear()  # Clear all collected spans

OTel Integration¶

if tracer.otel_available:
    print("Using OpenTelemetry backend")
else:
    print("Using internal tracing implementation")

@traced Decorator¶

The @traced decorator automatically instruments a function with tracing:

from fcc.observability.tracing import traced

@traced("my_operation")
def process_data(tracer: FccTracer, data: list) -> dict:
    # Automatically wrapped in a span named "my_operation"
    return {"processed": len(data)}

# Call with a tracer
result = process_data(tracer=tracer, data=[1, 2, 3])

The decorated function must accept a tracer keyword argument. If tracer is None, the function runs without tracing overhead.

FccMetrics¶

The FccMetrics class collects metric data points for FCC operations.

MetricPoint¶

Each metric observation is recorded as a frozen MetricPoint:

Field	Type	Description
`name`	`str`	Metric name (e.g. `fcc.simulation.step`)
`value`	`float`	Numeric value
`metric_type`	`str`	`"counter"`, `"gauge"`, or `"histogram"`
`labels`	`dict[str, str]`	Key-value labels for filtering/grouping
`timestamp`	`str`	ISO 8601 timestamp

Generic Operations¶

from fcc.observability.metrics import FccMetrics

metrics = FccMetrics()

# Counter increment
metrics.increment("my.counter", value=1.0, labels={"env": "prod"})

# Gauge/histogram observation
metrics.observe("my.gauge", value=42.5, labels={"component": "engine"})

Pre-defined FCC Metrics¶

The class provides 7 pre-defined convenience methods:

Method	Metric Name	Type	Labels
`record_simulation_step()`	`fcc.simulation.step`	counter	`node_id`, `step`
`record_persona_activation()`	`fcc.persona.activation`	counter	`persona_id`
`record_gate_result()`	`fcc.governance.gate.result`	counter	`gate_id`, `passed`
`record_deliverable()`	`fcc.deliverable.created`	counter	`persona_id`, `action_type`
`record_action_duration()`	`fcc.action.duration_ms`	gauge	`persona_id`, `action_type`
`record_collaboration_turn()`	`fcc.collaboration.turn`	counter	`session_id`, `turn_type`
`record_event_published()`	`fcc.event.published`	counter	`event_type`

metrics.record_simulation_step(step=3, node_id="n1")
metrics.record_persona_activation(persona_id="RC")
metrics.record_gate_result(gate_id="completeness", passed=True)

Accessing Collected Points¶

points = metrics.points  # Returns list[MetricPoint]
metrics.clear()           # Clear all collected points

Exporters¶

Span Exporters¶

All span exporters implement the SpanExporter ABC with an export(spans) -> int method.

ConsoleSpanExporter -- Prints spans to stderr (or a custom file object):

from fcc.observability.exporters import ConsoleSpanExporter

exporter = ConsoleSpanExporter()
count = exporter.export(tracer.completed_spans)
# Output: [SPAN] simulation.run trace=abc12345... 42.5ms status=ok scenario_id=GEN-001

JsonFileSpanExporter -- Writes spans to a JSON file:

from fcc.observability.exporters import JsonFileSpanExporter

exporter = JsonFileSpanExporter("output/spans.json")
count = exporter.export(tracer.completed_spans)

The JSON file format:

{
  "span_count": 5,
  "spans": [
    {
      "span_id": "...",
      "trace_id": "...",
      "name": "simulation.run",
      "start_time": "2026-03-26T...",
      "end_time": "2026-03-26T...",
      "status": "ok",
      "attributes": {"scenario_id": "GEN-001"}
    }
  ]
}

Metric Exporters¶

All metric exporters implement the MetricExporter ABC.

ConsoleMetricExporter -- Prints metrics to stderr:

from fcc.observability.exporters import ConsoleMetricExporter

exporter = ConsoleMetricExporter()
count = exporter.export(metrics.points)
# Output: [METRIC] fcc.simulation.step=1.0 type=counter node_id=n1 step=3

JsonFileMetricExporter -- Writes metrics to a JSON file:

from fcc.observability.exporters import JsonFileMetricExporter

exporter = JsonFileMetricExporter("output/metrics.json")
count = exporter.export(metrics.points)

Custom Exporters¶

Create custom exporters by subclassing the ABC:

from fcc.observability.exporters import SpanExporter, SpanData

class PrometheusSpanExporter(SpanExporter):
    def export(self, spans: list[SpanData]) -> int:
        # Push spans to Prometheus
        ...
        return len(spans)

    def shutdown(self) -> None:
        # Clean up resources
        ...

Engine Instrumentation¶

The fcc.observability.integration module provides helpers to instrument the simulation and action engines:

from fcc.observability.integration import (
    instrument_simulation_engine,
    instrument_action_engine,
)

instrument_simulation_engine(sim_engine, tracer, metrics)
instrument_action_engine(action_engine, tracer, metrics)

These functions wrap engine methods to automatically create spans and record metrics for each operation.