Observability API Reference¶

Module: fcc.observability

The observability module provides structured tracing, metrics collection, and exporters for monitoring FCC operations.

flowchart LR
    subgraph Instrumented Code
        SE[SimulationEngine]
        AE[ActionEngine]
    end
    subgraph Collection
        FT[FccTracer] -->|creates| SD[SpanData]
        FM[FccMetrics] -->|records| MP[MetricPoint]
    end
    subgraph Exporters
        SD --> CSE[ConsoleSpanExporter]
        SD --> JSE[JsonFileSpanExporter]
        MP --> CME[ConsoleMetricExporter]
        MP --> JME[JsonFileMetricExporter]
    end
    SE --> FT
    AE --> FT
    SE --> FM
    AE --> FM

Module Structure¶

Module	Description
`fcc.observability.tracing`	`FccTracer`, `SpanData`, `SpanContext`, `@traced`
`fcc.observability.metrics`	`FccMetrics`, `MetricPoint`
`fcc.observability.exporters`	Console and JSON file exporters
`fcc.observability.integration`	Engine instrumentation helpers

SpanData¶

from fcc.observability.tracing import SpanData

Frozen dataclass representing a completed span.

Property/Method	Signature	Returns	Description
`duration_ms`	property	`float \| None`	Elapsed time in ms; None if active
`to_dict()`	`()`	dict	Serialize to dictionary
`from_dict()`	`(data: dict)`	SpanData	Deserialize from dictionary

Fields: span_id, trace_id, name, start_time, parent_span_id, end_time, status, attributes.

SpanContext¶

from fcc.observability.tracing import SpanContext

Mutable span context used during execution.

Method	Signature	Returns	Description
`set_attribute()`	`(key: str, value: Any)`	None	Set a key-value attribute
`set_status()`	`(status: str)`	None	Set status (`"ok"` or `"error"`)
`end()`	`()`	SpanData	Finalize and return frozen snapshot

FccTracer¶

from fcc.observability.tracing import FccTracer

Method/Property	Signature	Returns	Description
`span()`	`(name: str, attributes: dict = None)`	context manager yielding SpanContext	Create a span
`completed_spans`	property	list[SpanData]	All completed spans
`clear()`	`()`	None	Clear collected spans
`otel_available`	property	bool	Whether OpenTelemetry is installed

Usage¶

from fcc.observability.tracing import FccTracer

tracer = FccTracer(service_name="fcc")

with tracer.span("simulation.run", {"scenario": "GEN-001"}) as ctx:
    ctx.set_attribute("steps", 5)
    # ... work ...

for span in tracer.completed_spans:
    print(f"{span.name}: {span.duration_ms}ms")

@traced¶

from fcc.observability.tracing import traced

Decorator that wraps a function in a span. The function must accept a tracer keyword argument.

@traced("my_operation")
def process(tracer: FccTracer, data: list) -> int:
    return len(data)

result = process(tracer=tracer, data=[1, 2, 3])

MetricPoint¶

from fcc.observability.metrics import MetricPoint

Frozen dataclass for a single metric observation.

Fields: name: str, value: float, metric_type: str ("counter", "gauge", "histogram"), labels: dict[str, str], timestamp: str.

Method	Signature	Returns	Description
`to_dict()`	`()`	dict	Serialize
`from_dict()`	`(data: dict)`	MetricPoint	Deserialize

FccMetrics¶

from fcc.observability.metrics import FccMetrics

Method	Signature	Returns	Description
`increment()`	`(name, value=1.0, labels=None)`	None	Record a counter increment
`observe()`	`(name, value, labels=None)`	None	Record a gauge observation
`record_simulation_step()`	`(step: int, node_id: str)`	None	Pre-defined: simulation step
`record_persona_activation()`	`(persona_id: str)`	None	Pre-defined: persona activation
`record_gate_result()`	`(gate_id: str, passed: bool)`	None	Pre-defined: gate result
`record_deliverable()`	`(persona_id, action_type)`	None	Pre-defined: deliverable created
`points`	property	list[MetricPoint]	All collected points
`clear()`	`()`	None	Clear collected points

Usage¶

from fcc.observability.metrics import FccMetrics

metrics = FccMetrics()
metrics.record_simulation_step(step=1, node_id="n1")
metrics.record_persona_activation(persona_id="RC")
metrics.increment("custom.counter", labels={"env": "test"})
print(len(metrics.points))  # 3

Span Exporters¶

SpanExporter (ABC)¶

from fcc.observability.exporters import SpanExporter

Method	Signature	Returns	Description
`export()`	`(spans: list[SpanData])`	int	Export spans; returns count
`shutdown()`	`()`	None	Clean up resources (optional)

ConsoleSpanExporter¶

Prints spans to stderr. Constructor: ConsoleSpanExporter(file=None).

JsonFileSpanExporter¶

Writes spans to JSON file. Constructor: JsonFileSpanExporter(path: str | Path).

Metric Exporters¶

MetricExporter (ABC)¶

Method	Signature	Returns	Description
`export()`	`(points: list[MetricPoint])`	int	Export metrics; returns count
`shutdown()`	`()`	None	Clean up resources (optional)

ConsoleMetricExporter¶

Prints metrics to stderr. Constructor: ConsoleMetricExporter(file=None).

JsonFileMetricExporter¶

Writes metrics to JSON file. Constructor: JsonFileMetricExporter(path: str | Path).

Integration Helpers¶

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

Function	Signature	Description
`instrument_simulation_engine()`	`(engine, tracer, metrics)`	Wrap simulation engine with tracing and metrics
`instrument_action_engine()`	`(engine, tracer, metrics)`	Wrap action engine with tracing and metrics