# Sprint 15: Platform Hardening

> **Size**: M | **Owner**: Backend | **Prerequisites**: Phase 4 complete
> **Phase**: 5 - Platform Evolution

---

## Validation Status (2025-12-30)

### ✅ COMPLETE

| Component | Status | Notes |
|-----------|--------|-------|
| Central error taxonomy | ✅ Implemented | `domain/errors.py` with ErrorCode enum and DomainError hierarchy |
| Structured logging | ✅ Implemented | LogBuffer with trace_id/span_id fields, OTel context extraction |
| OpenTelemetry instrumentation | ✅ Implemented | `infrastructure/observability/otel.py` with graceful degradation |
| Usage event stream (from OTel) | ✅ Implemented | `infrastructure/observability/usage.py` with multiple sink implementations |
| Usage metadata persistence | ✅ Implemented | SummarizationService populates `tokens_used`/`latency_ms` before save |
| Correlation ID propagation | ✅ Implemented | LogBufferHandler extracts trace/span IDs from OTel context |
| LogBuffer | ✅ Implemented | Ring buffer with 1000 capacity |
| MetricsCollector | ✅ Implemented | CPU, memory, disk, network metrics |
| GetRecentLogs RPC | ✅ Implemented | With level/source filtering |
| GetPerformanceMetrics RPC | ✅ Implemented | History only grows when `collect_now()` is called |

**Dependencies unlocked**: Sprint 23 (Analytics) and Sprint 25 (LangGraph) can now proceed.

---

## Objective

Make the system diagnosable and stable before adding surface area. Centralize error handling, standardize logging, and add **OpenTelemetry** traces/metrics/logs so usage events are derived consistently across gRPC, CLI, and future entrypoints.

---

## What Already Exists

| Asset | Location | Status |
|-------|----------|--------|
| `LogBuffer` ring storage | `infrastructure/logging/log_buffer.py` | ✅ Works |
| `GetRecentLogs` RPC | `grpc/_mixins/observability.py` | ✅ Works |
| `MetricsCollector` | `infrastructure/metrics/collector.py` | ✅ Works |
| `GetPerformanceMetrics` RPC | `grpc/_mixins/observability.py` | ✅ Works |
| `parse_meeting_id_or_abort()` | `grpc/_mixins/converters.py` | ✅ Pattern exists |
| `abort_not_found()`, `abort_invalid_argument()` | `grpc/_mixins/errors.py` | ✅ Helpers exist |
| `SummaryModel.tokens_used` | `persistence/models/core/summary.py` | ✅ Field exists |
| `SummaryModel.latency_ms` | `persistence/models/core/summary.py` | ✅ Field exists |

---

## Scope

| Task | Effort | Builds On | Risk |
|------|--------|-----------|------|
| Central error taxonomy | M | Existing abort patterns | Low |
| Observability port + OTel sink | M | Application services | Low |
| OpenTelemetry instrumentation | M | gRPC server wiring | Low |
| Usage event stream (from OTel) | S | Usage port | Low |
| Usage metadata persistence | S | Summarization service | Low |
| Correlation ID propagation | S | OTel context | Low |
| Logging context injection | S | LogBuffer handler | Low |
| Metrics background collection (optional) | S | MetricsCollector | Low |

---

## Implementation Plan

### Task 1: Central Error Taxonomy

**Objective**: Single place for domain errors → gRPC status mapping.

**File**: `src/noteflow/domain/errors.py`

```python
"""Domain error hierarchy with gRPC status mapping."""

from __future__ import annotations

from enum import Enum
from typing import TYPE_CHECKING

import grpc

if TYPE_CHECKING:
    pass


class ErrorCode(Enum):
    """Standard error codes with gRPC status mapping."""

    # Not found errors → NOT_FOUND
    MEETING_NOT_FOUND = ("meeting_not_found", grpc.StatusCode.NOT_FOUND)
    SEGMENT_NOT_FOUND = ("segment_not_found", grpc.StatusCode.NOT_FOUND)
    ENTITY_NOT_FOUND = ("entity_not_found", grpc.StatusCode.NOT_FOUND)
    JOB_NOT_FOUND = ("job_not_found", grpc.StatusCode.NOT_FOUND)

    # Validation errors → INVALID_ARGUMENT
    INVALID_MEETING_ID = ("invalid_meeting_id", grpc.StatusCode.INVALID_ARGUMENT)
    INVALID_REQUEST = ("invalid_request", grpc.StatusCode.INVALID_ARGUMENT)
    MISSING_REQUIRED_FIELD = ("missing_required_field", grpc.StatusCode.INVALID_ARGUMENT)

    # State errors → FAILED_PRECONDITION
    MEETING_ALREADY_STOPPED = ("meeting_already_stopped", grpc.StatusCode.FAILED_PRECONDITION)
    JOB_ALREADY_RUNNING = ("job_already_running", grpc.StatusCode.FAILED_PRECONDITION)
    CONSENT_REQUIRED = ("consent_required", grpc.StatusCode.FAILED_PRECONDITION)

    # Permission errors → PERMISSION_DENIED
    WORKSPACE_ACCESS_DENIED = ("workspace_access_denied", grpc.StatusCode.PERMISSION_DENIED)

    # Resource errors → RESOURCE_EXHAUSTED
    RATE_LIMITED = ("rate_limited", grpc.StatusCode.RESOURCE_EXHAUSTED)

    # Internal errors → INTERNAL
    PROVIDER_ERROR = ("provider_error", grpc.StatusCode.INTERNAL)
    DATABASE_ERROR = ("database_error", grpc.StatusCode.INTERNAL)

    def __init__(self, code: str, grpc_status: grpc.StatusCode) -> None:
        self._code = code
        self._grpc_status = grpc_status

    @property
    def code(self) -> str:
        return self._code

    @property
    def grpc_status(self) -> grpc.StatusCode:
        return self._grpc_status


class DomainError(Exception):
    """Base class for domain errors."""

    def __init__(
        self,
        error_code: ErrorCode,
        message: str,
        *,
        details: dict[str, object] | None = None,
    ) -> None:
        super().__init__(message)
        self.error_code = error_code
        self.message = message
        self.details = details or {}

    @property
    def grpc_status(self) -> grpc.StatusCode:
        return self.error_code.grpc_status


class NotFoundError(DomainError):
    """Resource not found."""

    def __init__(self, resource_type: str, resource_id: str) -> None:
        error_code = ErrorCode[f"{resource_type.upper()}_NOT_FOUND"]
        super().__init__(
            error_code,
            f"{resource_type} '{resource_id}' not found",
            details={"resource_type": resource_type, "resource_id": resource_id},
        )


class ValidationError(DomainError):
    """Request validation failed."""

    def __init__(self, field: str, reason: str) -> None:
        super().__init__(
            ErrorCode.INVALID_REQUEST,
            f"Validation failed for '{field}': {reason}",
            details={"field": field, "reason": reason},
        )
```

**File**: `src/noteflow/grpc/_mixins/errors.py` (update)

```python
"""gRPC error handling utilities."""

from __future__ import annotations

from typing import TYPE_CHECKING

import grpc

from noteflow.domain.errors import DomainError

if TYPE_CHECKING:
    pass


async def handle_domain_error(
    context: grpc.aio.ServicerContext,
    error: DomainError,
) -> None:
    """Set gRPC error from domain error."""
    await context.abort(error.grpc_status, error.message)


def domain_error_interceptor(func):
    """Decorator to convert domain errors to gRPC errors."""
    async def wrapper(self, request, context):
        try:
            return await func(self, request, context)
        except DomainError as e:
            await handle_domain_error(context, e)
    return wrapper
```

### Task 2: OpenTelemetry Instrumentation + Structured Logging

**Objective**: Standardize traces/metrics/logs and include correlation IDs in LogBuffer.

**Files**:
- `src/noteflow/infrastructure/observability/otel.py` (new)
- `src/noteflow/infrastructure/logging/context.py` (new logging filter to inject trace/span IDs)
- `src/noteflow/infrastructure/logging/log_buffer.py` (extend to include trace_id/span_id fields)

```python
"""OpenTelemetry setup for NoteFlow."""

from __future__ import annotations

from opentelemetry import metrics, trace
from opentelemetry.instrumentation.grpc import GrpcInstrumentorServer
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.metrics import MeterProvider


def configure_observability(service_name: str = "noteflow") -> None:
    """Initialize trace + metrics providers and instrument gRPC."""
    resource = Resource.create({"service.name": service_name})
    trace.set_tracer_provider(TracerProvider(resource=resource))
    metrics.set_meter_provider(MeterProvider(resource=resource))
    GrpcInstrumentorServer().instrument()
```

### Task 3: Observability Port + Usage Events

**Objective**: Emit usage events from spans/metrics via an application-layer port.

**Files**:
- `src/noteflow/application/observability/ports.py` (new)
- `src/noteflow/infrastructure/observability/usage.py` (OTel sink)

```python
"""Usage events derived from OpenTelemetry spans."""

from __future__ import annotations

from opentelemetry import trace

from noteflow.application.observability.ports import UsageEventSink


class OtelUsageEventSink(UsageEventSink):
    """Attach usage events to the current span."""

    def record(self, event: str, **attributes: object) -> None:
        span = trace.get_current_span()
        for key, value in attributes.items():
            span.set_attribute(key, value)
        span.add_event(event)
```

### Task 4: Persist Usage Metadata + Wire Events

**Update**: `src/noteflow/application/services/summarization_service.py`

Populate `Summary.tokens_used` / `Summary.latency_ms` and emit a usage event after successful summarization:

```python
service_result = await provider.summarize(request)
service_result.summary.tokens_used = service_result.tokens_used
service_result.summary.latency_ms = service_result.latency_ms

self._usage_events.record(
    "summarization.completed",
    meeting_id=str(meeting_id),
    tokens_input=service_result.tokens_used,
    tokens_output=None,
    latency_ms=service_result.latency_ms,
    provider=service_result.provider_name,
    model=service_result.model_name,
    success=True,
)
```

---

## Acceptance Criteria

### Functional

- [x] All gRPC errors use central error taxonomy
- [x] Logs include trace_id + span_id when available
- [x] Usage events emitted via UsageEventSink and attached to OTel spans
- [x] Summary tokens/latency persisted for analytics
- [x] Existing `GetRecentLogs` RPC returns structured logs

### Technical

- [x] No duplicate error handling patterns in mixins
- [x] Correlation IDs propagate through gRPC + async operations
- [x] OTel instrumentation is non-blocking (fire-and-forget)

### Quality Gates

- [x] `pytest tests/domain/test_errors.py` passes (34 tests)
- [x] `ruff check src/noteflow` passes
- [x] `basedpyright` passes (0 errors, 12 warnings for optional deps)
- [x] Meets `docs/sprints/QUALITY_STANDARDS.md` (lint + test quality thresholds)

---

## Files Created/Modified

| File | Change Type | Notes |
|------|-------------|-------|
| `src/noteflow/domain/errors.py` | Created | ErrorCode enum + DomainError hierarchy (6 error classes) |
| `src/noteflow/application/observability/__init__.py` | Created | Package exports |
| `src/noteflow/application/observability/ports.py` | Created | UsageEvent dataclass + UsageEventSink protocol + NullUsageEventSink |
| `src/noteflow/infrastructure/observability/__init__.py` | Created | Package exports |
| `src/noteflow/infrastructure/observability/otel.py` | Created | OTel setup with graceful degradation when not installed |
| `src/noteflow/infrastructure/observability/usage.py` | Created | LoggingUsageEventSink, OtelUsageEventSink, factory function |
| `src/noteflow/infrastructure/logging/log_buffer.py` | Modified | Added trace_id/span_id to LogEntry, OTel context extraction in handler |
| `src/noteflow/grpc/_mixins/errors.py` | Modified | Added handle_domain_error(), domain_error_handler decorator |
| `src/noteflow/application/services/summarization_service.py` | Modified | Persist tokens/latency, emit usage events |
| `pyproject.toml` | Modified | Added `observability` optional dependency group |
| `tests/domain/test_errors.py` | Created | 34 tests for error taxonomy |
| `tests/infrastructure/observability/__init__.py` | Created | Test package |
| `tests/infrastructure/observability/test_usage.py` | Created | 12 tests for usage sinks |
| `tests/infrastructure/observability/test_log_buffer.py` | Created | 12 tests for LogBuffer tracing

---

## Rollback Plan

1. **Error taxonomy issues**: Revert to inline abort calls
2. **Logging issues**: Disable structured logging, use standard
3. **Usage tracking issues**: Disable OTel usage events

---

## Post-Sprint

- ~~Add usage event persistence to database~~ ✅ Implemented (2025-12-30)
- ~~Add usage aggregation queries~~ ✅ Implemented (2025-12-30)
- Integrate with Analytics UI (Sprint 23)

---

## Completion Summary (2025-12-30)

### Implementation Highlights

1. **Central Error Taxonomy**: Created `domain/errors.py` with 18 error codes mapping to gRPC status codes. Includes specialized error classes: `NotFoundError`, `ValidationError`, `StateError`, `ProviderError`, `ConsentRequiredError`, `DatabaseRequiredError`.

2. **OpenTelemetry Infrastructure**: Optional dependency pattern with graceful degradation. When OTel is not installed, all functions return no-op implementations. Supports tracer/meter providers and gRPC instrumentation.

3. **Usage Event Stream**: Three sink implementations:
   - `NullUsageEventSink`: No-op for testing/disabled scenarios
   - `LoggingUsageEventSink`: Logs events for development
   - `OtelUsageEventSink`: Attaches events to OTel spans

4. **Correlation ID Propagation**: `LogBufferHandler` automatically extracts `trace_id` and `span_id` from OTel context and includes them in `LogEntry` records.

5. **Usage Metadata Persistence**: `SummarizationService.summarize()` now populates `tokens_used` and `latency_ms` on Summary entities before persistence.

### Test Coverage

- **58 new tests** across 3 test files
- All tests passing with 0.46s execution time
- Quality gates satisfied (ruff, basedpyright)

### Dependencies Added

```toml
[project.optional-dependencies]
observability = [
    "opentelemetry-api>=1.28",
    "opentelemetry-sdk>=1.28",
    "opentelemetry-instrumentation-grpc>=0.49b",
    "opentelemetry-exporter-otlp>=1.28",
]
```

Install with: `pip install -e ".[observability]"`
