# loopy.observe

## `loopy.observe.Tracer` (class)

Distributed tracer for LLM operations.

Example:
    tracer = Tracer(service="my_app")

    with tracer.start("llm_call") as span:
        span.set_attribute("model", "gpt-4")
        response = await llm.complete(prompt)
        span.set_attribute("tokens", response.usage.total_tokens)

```python
class Tracer:
    """
    Distributed tracer for LLM operations.

    Example:
        tracer = Tracer(service="my_app")

        with tracer.start("llm_call") as span:
            span.set_attribute("model", "gpt-4")
            response = await llm.complete(prompt)
            span.set_attribute("tokens", response.usage.total_tokens)
    """

    def __init__(self, service: str = "loopy", redactor: Redactor | None = None):
        self.service = service
        self._spans: list[Span] = []
        self._current_trace_id: str | None = None
        # v0.7.9 - optional redactor applied at span completion.
        self.redactor: Redactor | None = redactor
        # v0.8.0 - instrumentation controls. ``disabled`` is a runtime
        # flag (sets it to True to skip span recording without
        # touching call sites); ``shutdown`` is a one-way latch set
        # by :meth:`shutdown` and makes every public entry point a
        # graceful no-op (used by the @observe() decorator so it
        # never raises after a tracer is torn down).
        self.disabled: bool = False
        self._shutdown: bool = False

    def _generate_id(self) -> str:
        """Generate a unique ID using UUID4."""
        return uuid.uuid4().hex[:16]

    def start_span(
        self,
        name: str,
        parent_id: str | None = None,
        **attributes: Any,
    ) -> Span:
        """
        Start a new span.

        Args:
            name: Span name (e.g., "llm_call", "tool_use")
            parent_id: Optional parent span ID
            **attributes: Initial attributes

        Returns:
            New Span instance. When ``self.disabled`` is True or the
            tracer has been :meth:`shutdown`, a sentinel
            :class:`Span` with ``recorded=False`` is returned so the
            @observe() decorator can drive its lifecycle without
            raising.
        """
        trace_id = self._current_trace_id or self._generate_id()
        span_id = self._generate_id()

        if self.disabled or self._shutdown:
            return Span(
                name=name,
                trace_id=trace_id,
                span_id=span_id,
                parent_id=parent_id,
                recorded=False,
            )

        span = Span(
            name=name,
            trace_id=trace_id,
            span_id=span_id,
            parent_id=parent_id,
            attributes={"service": self.service, **attributes},
        )

        # v0.7.9 - scrub attributes/events before storage.
        if self.redactor is not None:
            span.attributes = self.redactor.redact_value(span.attributes)
            span.events = self.redactor.redact_value(span.events)

        self._spans.append(span)
        logger.debug("Started span: %s (%s)", name, span_id)

        return span

    def shutdown(self) -> None:
        """v0.8.0 — one-way latch: subsequent ``start_span`` calls return
        non-recording sentinel spans so instrumentation never raises.
        """
        self._shutdown = True

    def start(self, name: str, **attributes: Any) -> SpanContext:
        """
        Start a span with context manager support.

        Example:
            with tracer.start("llm_call") as span:
                span.set_attribute("model", "gpt-4")
                # ... do work ...
        """
        span = self.start_span(name, **attributes)
        return SpanContext(span)

    def get_spans(self) -> list[Span]:
        """Get all recorded spans."""
        return self._spans.copy()

    def get_trace(self, trace_id: str) -> list[Span]:
        """Get all spans for a trace."""
        return [s for s in self._spans if s.trace_id == trace_id]

    def export_json(self) -> str:
        """Export all spans as JSON."""
        return json.dumps([s.to_dict() for s in self._spans], indent=2)

    def export_otlp(self) -> dict[str, Any]:
        """
        Export spans in OTLP-compatible format.

        Delegates to :meth:`export_opentelemetry` for consistent
        output across all export methods.

        Returns:
            A dict with resource attributes and span data.
        """
        return self.export_opentelemetry()

    def clear(self) -> None:
        """Clear all spans."""
        self._spans.clear()

    def export_opentelemetry(self) -> dict[str, Any]:
        """
        Export spans in OpenTelemetry-compatible format.

        Returns a dict with resource info and spans ready for OTLP export.
        """
        return {
            "resource": {
                "attributes": {
                    "service.name": self.service,
                    "service.version": __version__,
                }
            },
            "spans": [
                {
                    "traceId": s.trace_id,
                    "spanId": s.span_id,
                    "parentSpanId": s.parent_id,
                    "operationName": s.name,
                    "startTime": int(s.start_time * 1e6),  # microseconds
                    "endTime": int((s.end_time or time.time()) * 1e6),
                    "attributes": [
                        {"key": k, "value": {"stringValue": str(v)}}
                        for k, v in s.attributes.items()
                    ],
                    "events": [
                        {
                            "name": e["name"],
                            "time": int(e["timestamp"] * 1e6),
                        }
                        for e in s.events
                    ],
                }
                for s in self._spans
            ],
        }
```

## `loopy.observe.Span` (class)

A single trace span representing an operation.

Example:
    span = Tracer.start_span("llm_call", model="gpt-4")
    # ... do work ...
    span.set_attribute("tokens", 150)
    span.set_status(SpanStatus.OK)
    span.end()

```python
@dataclass
class Span:
    """
    A single trace span representing an operation.

    Example:
        span = Tracer.start_span("llm_call", model="gpt-4")
        # ... do work ...
        span.set_attribute("tokens", 150)
        span.set_status(SpanStatus.OK)
        span.end()
    """

    name: str
    trace_id: str
    span_id: str
    parent_id: str | None = None

    start_time: float = field(default_factory=time.time)
    end_time: float | None = None

    status: SpanStatus = SpanStatus.UNSET
    attributes: dict[str, Any] = field(default_factory=dict)
    events: list[dict[str, Any]] = field(default_factory=list)
    # v0.8.0 — False for sentinel spans returned by a disabled or
    # shutdown tracer. Drives cheap no-ops in the @observe() decorator
    # and keeps export methods from surfacing internal noise.
    recorded: bool = True

    @property
    def duration_ms(self) -> float | None:
        if self.end_time is None:
            return None
        return (self.end_time - self.start_time) * 1000

    def set_attribute(self, key: str, value: Any) -> None:
        """Set a span attribute."""
        self.attributes[key] = value

    def add_event(self, name: str, attributes: dict[str, Any] | None = None) -> None:
        """Add an event to the span."""
        self.events.append(
            {
                "name": name,
                "timestamp": time.time(),
                "attributes": attributes or {},
            }
        )

    def set_status(self, status: SpanStatus, message: str = "") -> None:
        """Set span status."""
        self.status = status
        if message:
            self.attributes["status_message"] = message

    def end(self) -> None:
        """End the span."""
        self.end_time = time.time()

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary."""
        return {
            "name": self.name,
            "trace_id": self.trace_id,
            "span_id": self.span_id,
            "parent_id": self.parent_id,
            "start_time": self.start_time,
            "end_time": self.end_time,
            "duration_ms": self.duration_ms,
            "status": self.status.value,
            "attributes": self.attributes,
            "events": self.events,
        }
```

## `loopy.observe.SpanStatus` (class)

str(object='') -> str
str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or
errors is specified, then the object must expose a data buffer
that will be decoded using the given encoding and error handler.
Otherwise, returns the result of object.__str__() (if defined)
or repr(object).
encoding defaults to sys.getdefaultencoding().
errors defaults to 'strict'.

```python
class SpanStatus(str, Enum):
    OK = "ok"
    ERROR = "error"
    UNSET = "unset"
```

## `loopy.observe.MetricsCollector` (class)

Simple metrics collector for LLM observability.

Supports counters, histograms, and gauges with tag-based
grouping for summary aggregation.

Example:
    metrics = MetricsCollector()

    metrics.increment("llm.requests", tags={"model": "gpt-4"})
    metrics.histogram("llm.tokens", 150, tags={"model": "gpt-4"})
    metrics.gauge("cache.size", 42)

    summary = metrics.summary()

```python
class MetricsCollector:
    """
    Simple metrics collector for LLM observability.

    Supports counters, histograms, and gauges with tag-based
    grouping for summary aggregation.

    Example:
        metrics = MetricsCollector()

        metrics.increment("llm.requests", tags={"model": "gpt-4"})
        metrics.histogram("llm.tokens", 150, tags={"model": "gpt-4"})
        metrics.gauge("cache.size", 42)

        summary = metrics.summary()
    """

    def __init__(self):
        self._metrics: list[MetricPoint] = []

    def increment(self, name: str, value: float = 1, **tags: str) -> None:
        """Record a counter increment.

        Args:
            name: Metric name.
            value: Amount to increment by (default 1).
            **tags: Key-value tag pairs for grouping.
        """
        self._metrics.append(MetricPoint(name=name, value=value, tags=tags))

    def histogram(self, name: str, value: float, **tags: str) -> None:
        """Record a histogram observation.

        Args:
            name: Metric name.
            value: Observed value.
            **tags: Key-value tag pairs for grouping.
        """
        self._metrics.append(MetricPoint(name=name, value=value, tags=tags))

    def gauge(self, name: str, value: float, **tags: str) -> None:
        """Set a gauge to a value.

        Args:
            name: Metric name.
            value: Current gauge value.
            **tags: Key-value tag pairs for grouping.
        """
        self._metrics.append(MetricPoint(name=name, value=value, tags=tags))

    def summary(self) -> dict[str, Any]:
        """Get summary of collected metrics."""
        by_name: dict[str, list[float]] = {}
        for m in self._metrics:
            if m.name not in by_name:
                by_name[m.name] = []
            by_name[m.name].append(m.value)

        return {
            name: {
                "count": len(values),
                "sum": sum(values),
                "avg": sum(values) / len(values) if values else 0,
                "min": min(values) if values else 0,
                "max": max(values) if values else 0,
            }
            for name, values in by_name.items()
        }

    def export(self) -> list[dict[str, Any]]:
        """Export all metrics."""
        return [
            {
                "name": m.name,
                "value": m.value,
                "timestamp": m.timestamp,
                "tags": m.tags,
            }
            for m in self._metrics
        ]

    def clear(self) -> None:
        """Clear all metrics."""
        self._metrics.clear()
```

## `loopy.observe.Redactor` (class)

v0.7.9 - PII / secret aware redaction for traces and exports.

Replaces sensitive substrings with stable placeholders so trace
storage and HTTP export never leak credentials, tokens, or PII.

Built-in patterns (all enabled by default):

| name            | matches                                  |
|-----------------|------------------------------------------|
| ``email``       | RFC-ish email addresses                  |
| ``phone``       | US/International phone-shaped numbers    |
| ``ssn``         | US Social Security Numbers               |
| ``credit_card`` | 13-19 digit card-shaped numbers          |
| ``openai_key``  | ``sk-...``, ``sk-proj-...`` tokens       |
| ``aws_key``     | ``AKIA``/``ASIA`` access keys            |
| ``jwt``         | Three-segment dot-delimited JWTs         |
| ``bearer``      | ``Bearer <token>`` headers               |
| ``ipv4``        | IPv4 addresses                           |

Patterns can be removed (``redactor.disable("phone")``) or extended
(``redactor.add_pattern("employee_id", r"EID-d{6}")``).

The redactor is *pure-string* and side-effect free: ``redact()``
never mutates its input, only returns a new string.

```python
@dataclass
class Redactor:
    """v0.7.9 - PII / secret aware redaction for traces and exports.

    Replaces sensitive substrings with stable placeholders so trace
    storage and HTTP export never leak credentials, tokens, or PII.

    Built-in patterns (all enabled by default):

    | name            | matches                                  |
    |-----------------|------------------------------------------|
    | ``email``       | RFC-ish email addresses                  |
    | ``phone``       | US/International phone-shaped numbers    |
    | ``ssn``         | US Social Security Numbers               |
    | ``credit_card`` | 13-19 digit card-shaped numbers          |
    | ``openai_key``  | ``sk-...``, ``sk-proj-...`` tokens       |
    | ``aws_key``     | ``AKIA``/``ASIA`` access keys            |
    | ``jwt``         | Three-segment dot-delimited JWTs         |
    | ``bearer``      | ``Bearer <token>`` headers               |
    | ``ipv4``        | IPv4 addresses                           |

    Patterns can be removed (``redactor.disable("phone")``) or extended
    (``redactor.add_pattern("employee_id", r"EID-d{6}")``).

    The redactor is *pure-string* and side-effect free: ``redact()``
    never mutates its input, only returns a new string.
    """

    name: str = "default"
    enabled: dict[str, re.Pattern[str]] = field(default_factory=dict)
    extra: dict[str, re.Pattern[str]] = field(default_factory=dict)
    placeholder_format: str = "[{name}_REDACTED]"

    def __post_init__(self) -> None:
        if not self.enabled:
            self.enabled = {
                "email": re.compile(r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}\b"),
                "phone": re.compile(r"\b(?:\+?1[-.\s]?)?\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}\b"),
                "ssn": re.compile(r"\b\d{3}-\d{2}-\d{4}\b"),
                "credit_card": re.compile(r"\b(?:\d[ -]*?){13,19}\b"),
                "openai_key": re.compile(r"\bsk-(?:proj-)?[A-Za-z0-9_-]{20,}\b"),
                "aws_key": re.compile(r"\b(?:AKIA|ASIA)[A-Z0-9]{16}\b"),
                "jwt": re.compile(r"\beyJ[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\b"),
                "bearer": re.compile(r"(?i)Bearer\s+[A-Za-z0-9._\-+/=]+"),
                "ipv4": re.compile(r"\b(?:\d{1,3}\.){3}\d{1,3}\b"),
            }

    def add_pattern(self, name: str, pattern: str) -> None:
        """Register a custom regex pattern under ``name``.

        Raises ``ValueError`` if ``name`` collides with a built-in
        (use ``disable`` first if you really want to override).
        """
        if name in self.enabled:
            raise ValueError(
                f"{name!r} is a built-in pattern; disable it first if you want to override."
            )
        self.extra[name] = re.compile(pattern)

    def disable(self, name: str) -> None:
        """Remove a pattern from the active set."""
        self.enabled.pop(name, None)
        self.extra.pop(name, None)

    @property
    def active_patterns(self) -> dict[str, re.Pattern[str]]:
        """Combined dict of built-in + custom active patterns."""
        return {**self.enabled, **self.extra}

    def redact(self, text: str) -> str:
        """Return a copy of ``text`` with every match replaced."""
        if not isinstance(text, str) or not text:
            return text
        result = text
        for name, pattern in self.active_patterns.items():
            replacement = self.placeholder_format.format(name=name.upper())
            result = pattern.sub(replacement, result)
        return result

    def find_all(self, text: str) -> list[RedactionMatch]:
        """Return every match (name, span) without modifying ``text``."""
        if not isinstance(text, str) or not text:
            return []
        matches: list[RedactionMatch] = []
        for name, pattern in self.active_patterns.items():
            replacement = self.placeholder_format.format(name=name.upper())
            for m in pattern.finditer(text):
                matches.append(
                    RedactionMatch(
                        name=name,
                        start=m.start(),
                        end=m.end(),
                        replacement=replacement,
                    )
                )
        matches.sort(key=lambda m: m.start)
        return matches

    def redact_value(self, value: Any) -> Any:
        """Recursively redact string leaves inside dicts / lists / tuples."""
        if isinstance(value, str):
            return self.redact(value)
        if isinstance(value, dict):
            return {k: self.redact_value(v) for k, v in value.items()}
        if isinstance(value, list):
            return [self.redact_value(v) for v in value]
        if isinstance(value, tuple):
            return tuple(self.redact_value(v) for v in value)
        return value
```

## `loopy.observe.RedactionMatch` (class)

v0.7.9 - A single redacted substring.

```python
@dataclass
class RedactionMatch:
    """v0.7.9 - A single redacted substring."""

    name: str
    start: int
    end: int
    replacement: str

    def __repr__(self) -> str:
        return f"RedactionMatch(name={self.name!r}, len={self.end - self.start})"
```
