Skip to content

Observability

arcana.observability provides a structured audit log, OpenTelemetry tracing, and in-process metrics. Everything writes to ~/.arcana/logs/ by default.

from arcana.observability import configure_observability, get_audit_log

configure_observability()        # call once at startup
log = get_audit_log()
for event in log.tail(n=20):
    print(event)

Install the OTel extras for span export to Jaeger, Grafana, etc.:

pip install arcana-core[observability]

Configuration

arcana.observability.configure_observability 

configure_observability(base_dir=None)

Set up the global audit log and OTel tracing.

Safe to call multiple times — re-calling replaces the audit log path and reconfigures the OTel tracer provider.

Parameters:

Name Type Description Default
base_dir Path | None

Root for all observability data. Defaults to ~/.arcana.

 None
Source code in packages/arcana-core/arcana/observability/__init__.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def configure_observability(base_dir: Path | None = None) -> None:
    """Set up the global audit log and OTel tracing.

    Safe to call multiple times — re-calling replaces the audit log path
    and reconfigures the OTel tracer provider.

    Args:
        base_dir: Root for all observability data. Defaults to ``~/.arcana``.
    """
    global _audit_log

    root = base_dir or (Path.home() / ".arcana")
    log_dir = root / "logs"
    session_dir = log_dir / "sessions"

    _audit_log = AuditLog(log_dir / "audit.jsonl")

    configure_tracing(session_dir)

arcana.observability.get_audit_log 

get_audit_log()

Return the global AuditLog, or None if configure_observability() has not been called.

Source code in packages/arcana-core/arcana/observability/__init__.py 
58
59
60
def get_audit_log() -> AuditLog | None:
    """Return the global AuditLog, or None if configure_observability() has not been called."""
    return _audit_log

arcana.observability.tracer.configure_tracing 

configure_tracing(session_dir)

Configure OTel with a FileSpanExporter writing to session_dir.

No-op if opentelemetry-sdk is not installed. Install via: pip install arcana-core[observability]

Source code in packages/arcana-core/arcana/observability/tracer.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def configure_tracing(session_dir: Path) -> None:
    """Configure OTel with a FileSpanExporter writing to session_dir.

    No-op if opentelemetry-sdk is not installed. Install via:
        pip install arcana-core[observability]
    """
    try:
        from opentelemetry import trace  # type: ignore[import]
        from opentelemetry.sdk.trace import TracerProvider  # type: ignore[import]
        from opentelemetry.sdk.trace.export import SimpleSpanProcessor  # type: ignore[import]

        from arcana.observability.exporters.file import FileSpanExporter

        provider = TracerProvider()  # pyright: ignore[reportUnknownVariableType]
        provider.add_span_processor(  # pyright: ignore[reportUnknownMemberType]
            SimpleSpanProcessor(FileSpanExporter(session_dir))  # pyright: ignore[reportUnknownVariableType,reportArgumentType]
        )
        trace.set_tracer_provider(provider)  # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType]
    except ImportError:
        pass

arcana.observability.tracer.get_tracer 

get_tracer(name='arcana')

Return an OTel Tracer, or a no-op tracer if opentelemetry-api is not installed.

Source code in packages/arcana-core/arcana/observability/tracer.py 
69
70
71
72
73
74
75
76
def get_tracer(name: str = "arcana") -> Any:
    """Return an OTel Tracer, or a no-op tracer if opentelemetry-api is not installed."""
    try:
        from opentelemetry import trace  # type: ignore[import]

        return trace.get_tracer(name)  # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType]
    except ImportError:
        return _NOOP

Audit log

arcana.observability.audit.AuditLog 

AuditLog(path=None)

Append-only JSONL log. Each event is one line.

Errors during write are silently swallowed — observability must never break the main agent call path.

Source code in packages/arcana-core/arcana/observability/audit.py 
19
20
21
def __init__(self, path: Path | None = None) -> None:
    self._path = path or self.DEFAULT_PATH
    self._path.parent.mkdir(parents=True, exist_ok=True)

append 

append(event)

Serialize event to JSONL and append. Non-fatal on I/O error.

Source code in packages/arcana-core/arcana/observability/audit.py 
27
28
29
30
31
32
33
34
def append(self, event: AuditEvent) -> None:
    """Serialize event to JSONL and append. Non-fatal on I/O error."""
    try:
        line = json.dumps(event_to_dict(event), default=str) + "\n"
        with open(self._path, "a", encoding="utf-8") as f:
            f.write(line)
    except Exception:
        pass

tail 

tail(n=50, event_type=None)

Return the last n events, optionally filtered by type.

Source code in packages/arcana-core/arcana/observability/audit.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def tail(self, n: int = 50, event_type: str | None = None) -> list[dict[str, Any]]:
    """Return the last n events, optionally filtered by type."""
    if not self._path.exists():
        return []
    with open(self._path, encoding="utf-8") as f:
        lines = f.readlines()

    collected: list[dict[str, Any]] = []
    for line in reversed(lines):
        line = line.strip()
        if not line:
            continue
        try:
            event = json.loads(line)
        except json.JSONDecodeError:
            continue
        if event_type is None or event.get("type") == event_type:
            collected.append(event)
            if len(collected) >= n:
                break
    return list(reversed(collected))

clear 

clear()

Delete the log file. Useful in tests.

Source code in packages/arcana-core/arcana/observability/audit.py 
58
59
60
61
def clear(self) -> None:
    """Delete the log file. Useful in tests."""
    if self._path.exists():
        self._path.unlink()

Events

arcana.observability.events.AuditEvent module-attribute 

AuditEvent = (
    SessionEvent
    | ModelCallEvent
    | RoutingEvent
    | MemoryReadEvent
    | MemoryWriteEvent
    | MemoryPruneEvent
    | MemoryDegradedEvent
)

arcana.observability.events.SessionEvent dataclass 

SessionEvent(
    session_id,
    agent_id,
    agent_name,
    card,
    modifier_cards,
    model,
    input_tokens,
    output_tokens,
    duration_ms,
    status,
    timestamp=_now_iso(),
    cost=None,
)

Emitted by Agent after each run() or stream() completes.

arcana.observability.events.ModelCallEvent dataclass 

ModelCallEvent(
    session_id,
    model,
    latency_ms,
    input_tokens,
    output_tokens,
    attempt,
    success,
    timestamp=_now_iso(),
    error=None,
)

Emitted by ModelGateway after each adapter call (including retries).

arcana.observability.events.RoutingEvent dataclass 

RoutingEvent(
    session_id,
    prompt_preview,
    outcome,
    matched_rule_trigger,
    target_agent_name,
    target_card,
    rules_evaluated,
    confidence,
    duration_ms,
    timestamp=_now_iso(),
    namespace_id="local",
    workspace_id="default",
)

Emitted by the World Engine before routing each prompt.

arcana.observability.events.MemoryReadEvent dataclass 

MemoryReadEvent(
    session_id,
    agent_id,
    query_text,
    results_count,
    latency_ms,
    timestamp=_now_iso(),
)

Emitted on each memory retrieval.

arcana.observability.events.MemoryWriteEvent dataclass 

MemoryWriteEvent(
    session_id,
    agent_id,
    memory_type,
    importance,
    timestamp=_now_iso(),
)

Emitted on each memory write.

arcana.observability.events.MemoryPruneEvent dataclass 

MemoryPruneEvent(
    agent_id,
    scanned,
    archived,
    purged,
    timestamp=_now_iso(),
)

Emitted after a memory prune pass over one store.

arcana.observability.events.MemoryDegradedEvent dataclass 

MemoryDegradedEvent(
    agent_id,
    session_id,
    tier,
    operation,
    reason,
    message="",
    timestamp=_now_iso(),
)

Emitted when a memory tier is skipped or drops out of an operation.

A degraded read returns partial context; a degraded shared/global write is dropped. Either way the session proceeds — this event is how the thinning is surfaced instead of being silent.

Emitters

emit_degraded records a memory tier degradation to both the audit log and metrics. It is the default sink the resilience layer uses when a tier is skipped or drops out of an operation; the emission is best-effort and never raises, so observability can never break the memory path.

arcana.observability.emit_degraded 

emit_degraded(event)

Record a memory degradation to the audit log and metrics (best effort).

Observability must never break the memory path, so every failure here is swallowed. Mirrors the best-effort emission the adapters already use.

Source code in packages/arcana-core/arcana/observability/__init__.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
def emit_degraded(event: MemoryDegradedEvent) -> None:
    """Record a memory degradation to the audit log and metrics (best effort).

    Observability must never break the memory path, so every failure here is
    swallowed. Mirrors the best-effort emission the adapters already use.
    """
    try:
        audit = get_audit_log()
        if audit is not None:
            audit.append(event)
        get_metrics().record_memory_degraded(tier=event.tier, operation=event.operation, reason=event.reason)
    except Exception:
        pass

Metrics

Alongside the session and model-call instruments, the memory federation publishes its resilience signals here: arcana.memory.tier.degraded (a counter labelled by tier / operation / reason), arcana.memory.tier.latency_ms, arcana.memory.circuit.state (0=closed, 1=half_open, 2=open per tier), and the background-queue gauges arcana.memory.queue.depth and arcana.memory.queue.drain_seconds.

arcana.observability.metrics.ArcanaMetrics 

ArcanaMetrics()

All OTel metric instruments used by arcana. One instance per process.

Source code in packages/arcana-core/arcana/observability/metrics.py 
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def __init__(self) -> None:
    meter = _get_otel_meter("arcana")
    self.sessions_total = meter.create_counter(
        "arcana.sessions.total",
        description="Total number of agent sessions",
    )
    self.input_tokens = meter.create_counter(
        "arcana.model.tokens.input",
        description="Total input tokens consumed",
    )
    self.output_tokens = meter.create_counter(
        "arcana.model.tokens.output",
        description="Total output tokens generated",
    )
    self.session_duration = meter.create_histogram(
        "arcana.session.duration_ms",
        description="Session duration in milliseconds",
        unit="ms",
    )
    self.model_latency = meter.create_histogram(
        "arcana.model.latency_ms",
        description="LLM call latency in milliseconds",
        unit="ms",
    )
    # --- Memory federation resilience ---
    self.memory_tier_degraded = meter.create_counter(
        "arcana.memory.tier.degraded",
        description="Memory tier operations that degraded (timeout, breaker, error, corruption)",
    )
    self.memory_tier_latency = meter.create_histogram(
        "arcana.memory.tier.latency_ms",
        description="Per-tier memory operation latency in milliseconds",
        unit="ms",
    )
    self.memory_circuit_state = _create_gauge(
        meter,
        "arcana.memory.circuit.state",
        description="Per-tier circuit breaker state (0=closed, 1=half_open, 2=open)",
    )
    self.memory_queue_depth = _create_gauge(
        meter,
        "arcana.memory.queue.depth",
        description="Pending background memory jobs (extraction / consolidation)",
    )
    self.memory_queue_drain_seconds = _create_gauge(
        meter,
        "arcana.memory.queue.drain_seconds",
        description="Estimated seconds to drain the background memory job queue",
    )

arcana.observability.metrics.get_metrics 

get_metrics()

Return the process-wide ArcanaMetrics instance, creating it on first call.

Source code in packages/arcana-core/arcana/observability/metrics.py 
155
156
157
158
159
160
def get_metrics() -> ArcanaMetrics:
    """Return the process-wide ArcanaMetrics instance, creating it on first call."""
    global _metrics
    if _metrics is None:
        _metrics = ArcanaMetrics()
    return _metrics