orchestrata_ultimate.kernel.audit.logger

Reference › orchestrata_ultimate › kernel › audit › logger

The audit logging service of the kernel, SPEC-001 story US-008 (REQ-001-US-008-020 and the logging, query and export requirements) with the fire-and-forget path from story US-010 (REQ-001-US-010-018). AuditLogger accepts typed AuditEntry records, buffers them, and writes them through the configured writers, optionally via a Redis queue. audit_bootstrap builds it, and pipeline_escalation uses it to record routing and escalation. The entry model is in model.

The constructor creates a file writer and a graph writer according to the config switches, a bounded buffer, and resolves the Redis host and port. start connects the Redis queue (when enabled and importable), then launches a worker task and a flush task; stop cancels both, flushes the buffer and closes the writers and the queue. log first tries to enqueue the entry in Redis and returns; if that is unavailable it appends to the buffer and writes a batch to all writers when the buffer is full. _worker_loop dequeues entries and writes each to every writer, _flush_loop flushes on the configured interval, and _flush drains the buffer. Writer errors are logged and never raised.

Typed helpers build entries through _create_entry: log_message_routing, log_policy_query, log_escalation, log_session_lifecycle, log_resource_event, log_workflow_event and log_volk_lifecycle. Query methods read the log files (including gzip archives): query_by_session, query_by_volk, query_by_event_type, query_by_time_range, query_by_correlation_id and search_logs. aggregate_logs, export_logs (JSON or CSV, default window seven days), generate_audit_report, backup_logs and restore_logs complete the surface.

At a glance

Class diagram

Dependency diagram

API

Classes

AuditLogger

Decorators: traces(spec='001', story='US-008', reqs=['REQ-001-US-008-020']) · Line 44

Main audit logging service (US-008).

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/kernel/audit/logger.py

# mypy: ignore-errors
"""Main audit logging service with buffered writes and query capabilities."""

import asyncio
import gzip
import json
import shutil
import uuid
from collections import deque
from collections.abc import Callable, Iterable
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.redis_factory import (
    resolve_redis_host,
    resolve_redis_port,
)
from orchestrata_ultimate.kernel.audit._common import (
    REDIS_AVAILABLE,
    _logger,
    _utc_now,
)
from orchestrata_ultimate.kernel.audit.model import (
    AuditEntry,
    AuditEventType,
    AuditLogConfig,
    AuditOutcome,
)
from orchestrata_ultimate.kernel.audit.queue import RedisAuditQueue
from orchestrata_ultimate.kernel.audit.writers import (
    AuditLogWriter,
    FileAuditWriter,
    GraphAuditWriter,
)


@traces(
    spec="001",
    story="US-008",
    reqs=["REQ-001-US-008-020"],
)
class AuditLogger:
    """Main audit logging service (US-008).

    Provides comprehensive audit logging with:
    - Multi-destination writing (REQ-001-US-008-020)
    - Buffered async writes
    - Query capabilities (REQ-001-US-008-021 to REQ-001-US-008-030)
    """

    @traces(spec="001", story="US-008", reqs=["REQ-001-US-008-020"])
    def __init__(
        self,
        config: AuditLogConfig | None = None,
        redis_host: str | None = None,
        redis_port: int | None = None,
        redis_db: int = 15,
        enable_redis_queue: bool = True,
        redis_queue_key: str | None = None,
    ) -> None:
        self.config = config or AuditLogConfig()
        self._writers: list[AuditLogWriter] = []
        self._buffer: deque[AuditEntry] = deque(maxlen=self.config.buffer_size)
        self._lock = asyncio.Lock()
        self._running = False
        self._flush_task: asyncio.Task | None = None

        # Redis queue for fire-and-forget audit logging
        self._redis_queue: RedisAuditQueue | None = None
        self._worker_task: asyncio.Task | None = None
        self._enable_redis_queue = enable_redis_queue and REDIS_AVAILABLE
        self._redis_host = (
            redis_host if redis_host is not None else resolve_redis_host()
        )
        self._redis_port = (
            redis_port if redis_port is not None else resolve_redis_port()
        )
        self._redis_db = redis_db
        self._redis_queue_key = redis_queue_key

        # Initialize writers
        if self.config.enable_file:
            self._writers.append(FileAuditWriter(self.config))
        if self.config.enable_graph:
            self._writers.append(GraphAuditWriter(self.config))

    async def start(self) -> None:
        """Start the audit logger."""
        if self._running:
            return
        self._running = True

        # Initialize Redis queue if enabled
        if self._enable_redis_queue:
            self._redis_queue = RedisAuditQueue(
                redis_host=self._redis_host,
                redis_port=self._redis_port,
                redis_db=self._redis_db,
                queue_key=self._redis_queue_key,
            )
            await self._redis_queue.connect()

            # Start background worker to process queue
            self._worker_task = asyncio.create_task(self._worker_loop())

        self._flush_task = asyncio.create_task(self._flush_loop())
        _logger.info("Audit logger started")

    async def stop(self) -> None:
        """Stop the audit logger."""
        self._running = False

        # Cancel worker task
        if self._worker_task:
            self._worker_task.cancel()
            try:
                await self._worker_task
            except asyncio.CancelledError:
                _logger.debug("audit worker task cancelled")

        # Cancel flush task
        if self._flush_task:
            self._flush_task.cancel()
            try:
                await self._flush_task
            except asyncio.CancelledError:
                _logger.debug("audit flush task cancelled")

        # Flush remaining entries
        await self._flush()

        # Close writers
        for writer in self._writers:
            await writer.close()

        # Close Redis queue
        if self._redis_queue:
            await self._redis_queue.close()

        _logger.info("Audit logger stopped")

    @traces(spec="001", story="US-010", reqs=["REQ-001-US-010-018"])
    async def _flush_loop(self) -> None:
        """Background task to flush buffer periodically."""
        while self._running:
            try:
                await asyncio.sleep(self.config.flush_interval_seconds)
                await self._flush()
            except asyncio.CancelledError:
                break
            except Exception as e:
                _logger.error(f"Error in flush loop: {e}")

    @traces(spec="001", story="US-010", reqs=["REQ-001-US-010-018"])
    async def _worker_loop(self) -> None:
        """Background worker to process Redis queue."""
        while self._running:
            try:
                if self._redis_queue is None:
                    await asyncio.sleep(1.0)
                    continue

                # Dequeue entry with timeout
                entry = await self._redis_queue.dequeue(timeout=1.0)
                if entry is None:
                    continue

                # Write to all writers
                for writer in self._writers:
                    try:
                        await writer.write(entry)
                    except Exception as e:
                        _logger.error(f"Error writing audit entry from queue: {e}")

            except asyncio.CancelledError:
                break
            except Exception as e:
                _logger.error(f"Error in worker loop: {e}")

    @traces(spec="001", story="US-008", reqs=["REQ-001-US-008-020"])
    async def _flush(self) -> None:
        """Flush buffered entries to writers."""
        async with self._lock:
            if not self._buffer:
                return

            entries = list(self._buffer)
            self._buffer.clear()

        for writer in self._writers:
            try:
                await writer.write_batch(entries)
            except Exception as e:
                _logger.error(f"Error flushing to writer {type(writer).__name__}: {e}")

    @traces(
        spec="001",
        story="US-008",
        reqs=["REQ-001-US-008-008", "REQ-001-US-008-009", "REQ-001-US-008-010"],
    )
    def _create_entry(
        self,
        event_type: AuditEventType,
        session_id: str,
        volk_id: str,
        correlation_id: str | None = None,
        outcome: AuditOutcome | None = None,
        data: dict[str, Any] | None = None,
        reason: str | None = None,
        duration_ms: int | None = None,
    ) -> AuditEntry:
        """Create an audit entry with defaults."""
        return AuditEntry(
            entry_id=str(uuid.uuid4()),
            event_type=event_type,
            timestamp=_utc_now(),
            session_id=session_id,
            correlation_id=correlation_id or str(uuid.uuid4()),
            volk_id=volk_id,
            outcome=outcome,
            data=data or {},
            reason=reason,
            duration_ms=duration_ms,
        )

    @traces(spec="001", story="US-010", reqs=["REQ-001-US-010-018"])
    async def log(self, entry: AuditEntry) -> None:
        """Log an audit entry (fire-and-forget via Redis queue)."""
        # If Redis queue is available, use fire-and-forget pattern. enqueue()
        # returns False when Redis is unreachable; in that case fall through to
        # the in-memory buffer / file path so the entry is not silently lost.
        if self._redis_queue and await self._redis_queue.enqueue(entry):
            return

        # Fallback to buffer if Redis is not available
        async with self._lock:
            self._buffer.append(entry)

            # Flush if buffer is full
            if len(self._buffer) >= self.config.buffer_size:
                entries = list(self._buffer)
                self._buffer.clear()

                for writer in self._writers:
                    try:
                        await writer.write_batch(entries)
                    except Exception as e:
                        _logger.error(f"Error writing batch: {e}")

    # ===== Event Logging Methods (REQ-001-US-008-001 to REQ-001-US-008-007) =====

    @traces(spec="001", story="US-008", reqs=["REQ-001-US-008-001"])
    async def log_message_routing(
        self,
        session_id: str,
        volk_id: str,
        message_id: str,
        from_volk: str,
        to_volk: str,
        route_decision: str,
        correlation_id: str | None = None,
        duration_ms: int | None = None,
    ) -> None:
        """Log message routing decision (REQ-001-US-008-001)."""
        entry = self._create_entry(
            event_type=AuditEventType.MESSAGE_ROUTING,
            session_id=session_id,
            volk_id=volk_id,
            correlation_id=correlation_id,
            outcome=AuditOutcome.SUCCESS,
            data={
                "message_id": message_id,
                "from_volk": from_volk,
                "to_volk": to_volk,
                "route_decision": route_decision,
            },
            duration_ms=duration_ms,
        )
        await self.log(entry)

    @traces(spec="001", story="US-008", reqs=["REQ-001-US-008-002"])
    async def log_policy_query(
        self,
        session_id: str,
        volk_id: str,
        query: str,
        result: str,
        cached: bool,
        correlation_id: str | None = None,
        duration_ms: int | None = None,
    ) -> None:
        """Log policy query and result (REQ-001-US-008-002)."""
        entry = self._create_entry(
            event_type=AuditEventType.POLICY_QUERY,
            session_id=session_id,
            volk_id=volk_id,
            correlation_id=correlation_id,
            outcome=AuditOutcome.SUCCESS,
            data={
                "query": query,
                "result": result,
                "cached": cached,
            },
            duration_ms=duration_ms,
        )
        await self.log(entry)

    @traces(
        spec="001",
        story="US-003",
        reqs=[
            "REQ-001-US-003-030",
            "REQ-001-US-005-007",
            "REQ-001-US-005-019",
            "REQ-001-US-005-024",
            "REQ-001-US-005-028",
            "REQ-001-US-008-003",
        ],
    )
    async def log_escalation(
        self,
        session_id: str,
        volk_id: str,
        issue: str,
        severity: str,
        escalation_path: list[str],
        resolved: bool,
        resolution: str | None = None,
        correlation_id: str | None = None,
    ) -> None:
        """Log escalation event (REQ-001-US-008-003)."""
        entry = self._create_entry(
            event_type=AuditEventType.ESCALATION,
            session_id=session_id,
            volk_id=volk_id,
            correlation_id=correlation_id,
            outcome=AuditOutcome.SUCCESS if resolved else AuditOutcome.ESCALATED,
            data={
                "issue": issue,
                "severity": severity,
                "escalation_path": escalation_path,
                "resolved": resolved,
                "resolution": resolution,
            },
        )
        await self.log(entry)

    @traces(
        spec="001",
        story="US-004",
        reqs=["REQ-001-US-004-010", "REQ-001-US-004-029", "REQ-001-US-008-004"],
    )
    async def log_session_lifecycle(
        self,
        session_id: str,
        volk_id: str,
        event: str,
        context: dict[str, Any],
        correlation_id: str | None = None,
    ) -> None:
        """Log session lifecycle event (REQ-001-US-008-004)."""
        event_type_map = {
            "start": AuditEventType.SESSION_START,
            "end": AuditEventType.SESSION_END,
            "pause": AuditEventType.SESSION_PAUSE,
            "resume": AuditEventType.SESSION_RESUME,
        }
        entry = self._create_entry(
            event_type=event_type_map.get(event, AuditEventType.SESSION_START),
            session_id=session_id,
            volk_id=volk_id,
            correlation_id=correlation_id,
            outcome=AuditOutcome.SUCCESS,
            data={"event": event, "context": context},
        )
        await self.log(entry)

    @traces(
        spec="001",
        story="US-007",
        reqs=[
            "REQ-001-US-007-008",
            "REQ-001-US-008-005",
        ],
    )
    async def log_resource_event(
        self,
        session_id: str,
        volk_id: str,
        resource_type: str,
        resource_id: str,
        action: str,
        amount: float | None = None,
        correlation_id: str | None = None,
    ) -> None:
        """Log resource allocation/deallocation (REQ-001-US-008-005)."""
        event_type = (
            AuditEventType.RESOURCE_ALLOCATION
            if action == "allocate"
            else AuditEventType.RESOURCE_DEALLOCATION
        )
        entry = self._create_entry(
            event_type=event_type,
            session_id=session_id,
            volk_id=volk_id,
            correlation_id=correlation_id,
            outcome=AuditOutcome.SUCCESS,
            data={
                "resource_type": resource_type,
                "resource_id": resource_id,
                "action": action,
                "amount": amount,
            },
        )
        await self.log(entry)

    @traces(
        spec="001",
        story="US-006",
        reqs=["REQ-001-US-006-020", "REQ-001-US-008-006"],
    )
    async def log_workflow_event(
        self,
        session_id: str,
        volk_id: str,
        workflow_id: str,
        event: str,
        step: str | None = None,
        step_result: str | None = None,
        correlation_id: str | None = None,
        duration_ms: int | None = None,
    ) -> None:
        """Log workflow execution step (REQ-001-US-008-006)."""
        event_type_map = {
            "start": AuditEventType.WORKFLOW_START,
            "step": AuditEventType.WORKFLOW_STEP,
            "end": AuditEventType.WORKFLOW_END,
        }
        entry = self._create_entry(
            event_type=event_type_map.get(event, AuditEventType.WORKFLOW_STEP),
            session_id=session_id,
            volk_id=volk_id,
            correlation_id=correlation_id,
            outcome=AuditOutcome.SUCCESS,
            data={
                "workflow_id": workflow_id,
                "event": event,
                "step": step,
                "step_result": step_result,
            },
            duration_ms=duration_ms,
        )
        await self.log(entry)

    @traces(spec="001", story="US-008", reqs=["REQ-001-US-008-007"])
    async def log_volk_lifecycle(
        self,
        session_id: str,
        volk_id: str,
        event: str,
        volk_alias: str,
        capabilities: list[str] | None = None,
        protocols: list[str] | None = None,
        correlation_id: str | None = None,
    ) -> None:
        """Log Volk registration/deregistration (REQ-001-US-008-007)."""
        event_type_map = {
            "register": AuditEventType.VOLK_REGISTER,
            "deregister": AuditEventType.VOLK_DEREGISTER,
            "heartbeat": AuditEventType.VOLK_HEARTBEAT,
        }
        entry = self._create_entry(
            event_type=event_type_map.get(event, AuditEventType.VOLK_REGISTER),
            session_id=session_id,
            volk_id=volk_id,
            correlation_id=correlation_id,
            outcome=AuditOutcome.SUCCESS,
            data={
                "event": event,
                "volk_alias": volk_alias,
                "capabilities": capabilities or [],
                "protocols": protocols or [],
            },
        )
        await self.log(entry)

    # ===== Query Methods (REQ-001-US-008-021 to REQ-001-US-008-030) =====

    @traces(
        spec="001",
        story="US-008",
        reqs=["REQ-001-US-008-021"],
    )
    async def query_by_session(
        self,
        session_id: str,
        limit: int = 100,
    ) -> list[AuditEntry]:
        """Query logs by session ID (REQ-001-US-008-021)."""
        entries = []

        # Query from file
        if self.config.enable_file:
            entries.extend(
                await self._query_file_logs(
                    filter_fn=lambda e: e.session_id == session_id,
                    limit=limit,
                )
            )

        return entries[:limit]

    @traces(spec="001", story="US-008", reqs=["REQ-001-US-008-022"])
    async def query_by_volk(
        self,
        volk_id: str,
        limit: int = 100,
    ) -> list[AuditEntry]:
        """Query logs by Volk alias (REQ-001-US-008-022)."""
        entries = []

        if self.config.enable_file:
            entries.extend(
                await self._query_file_logs(
                    filter_fn=lambda e: e.volk_id == volk_id,
                    limit=limit,
                )
            )

        return entries[:limit]

    @traces(spec="001", story="US-008", reqs=["REQ-001-US-008-023"])
    async def query_by_event_type(
        self,
        event_type: AuditEventType,
        limit: int = 100,
    ) -> list[AuditEntry]:
        """Query logs by event type (REQ-001-US-008-023)."""
        entries = []

        if self.config.enable_file:
            entries.extend(
                await self._query_file_logs(
                    filter_fn=lambda e: e.event_type == event_type,
                    limit=limit,
                )
            )

        return entries[:limit]

    @traces(spec="001", story="US-008", reqs=["REQ-001-US-008-024"])
    async def query_by_time_range(
        self,
        start: datetime,
        end: datetime | None = None,
        limit: int = 100,
    ) -> list[AuditEntry]:
        """Query logs by time range (REQ-001-US-008-024)."""
        end = end or _utc_now()
        entries = []

        if self.config.enable_file:
            entries.extend(
                await self._query_file_logs(
                    filter_fn=lambda e: start <= e.timestamp <= end,
                    limit=limit,
                )
            )

        return entries[:limit]

    @traces(
        spec="001",
        story="US-008",
        reqs=["REQ-001-US-008-025"],
    )
    async def query_by_correlation_id(
        self,
        correlation_id: str,
        limit: int = 100,
    ) -> list[AuditEntry]:
        """Query logs by correlation ID (REQ-001-US-008-025)."""
        entries = []

        if self.config.enable_file:
            entries.extend(
                await self._query_file_logs(
                    filter_fn=lambda e: e.correlation_id == correlation_id,
                    limit=limit,
                )
            )

        return entries[:limit]

    @traces(spec="001", story="US-008", reqs=["REQ-001-US-008-026"])
    async def search_logs(
        self,
        search_text: str,
        limit: int = 100,
    ) -> list[AuditEntry]:
        """Full-text search in logs (REQ-001-US-008-026)."""
        search_lower = search_text.lower()
        entries = []

        if self.config.enable_file:
            entries.extend(
                await self._query_file_logs(
                    filter_fn=lambda e: (
                        search_lower in json.dumps(e.data).lower()
                        or bool(e.reason and search_lower in e.reason.lower())
                    ),
                    limit=limit,
                )
            )

        return entries[:limit]

    @traces(
        spec="001",
        story="US-008",
        reqs=["REQ-001-US-008-027"],
    )
    async def aggregate_logs(
        self,
        group_by: str,
        start: datetime | None = None,
        end: datetime | None = None,
    ) -> dict[str, int]:
        """Aggregate logs by field (REQ-001-US-008-027)."""
        start = start or (_utc_now() - timedelta(days=7))
        end = end or _utc_now()

        aggregation: dict[str, int] = {}

        entries = await self.query_by_time_range(start, end, limit=10000)

        for entry in entries:
            if group_by == "event_type":
                key = entry.event_type.value
            elif group_by == "volk_id":
                key = entry.volk_id
            elif group_by == "session_id":
                key = entry.session_id
            elif group_by == "outcome":
                key = entry.outcome.value if entry.outcome else "none"
            else:
                key = str(entry.data.get(group_by, "unknown"))

            aggregation[key] = aggregation.get(key, 0) + 1

        return aggregation

    @traces(
        spec="001",
        story="US-008",
        reqs=["REQ-001-US-008-017", "REQ-001-US-008-029"],
    )
    async def export_logs(
        self,
        format: str,
        start: datetime | None = None,
        end: datetime | None = None,
        output_path: Path | None = None,
    ) -> str:
        """Export logs in specified format (REQ-001-US-008-017, REQ-001-US-008-029)."""
        start = start or (_utc_now() - timedelta(days=7))
        end = end or _utc_now()

        entries = await self.query_by_time_range(start, end, limit=100000)

        if format == "json":
            content = json.dumps([e.to_dict() for e in entries], indent=2)
        elif format == "csv":
            lines = ["entry_id,event_type,timestamp,session_id,volk_id,outcome,reason"]
            for e in entries:
                lines.append(
                    f"{e.entry_id},{e.event_type.value},{e.timestamp.isoformat()},"
                    f"{e.session_id},{e.volk_id},{e.outcome.value if e.outcome else ''},"
                    f'"{e.reason or ""}"'
                )
            content = "\n".join(lines)
        else:
            raise ValueError(f"Unsupported format: {format}")

        if output_path:
            output_path.write_text(content)

        return content

    @traces(
        spec="001",
        story="US-008",
        reqs=["REQ-001-US-008-030"],
    )
    async def generate_audit_report(
        self,
        start: datetime,
        end: datetime | None = None,
    ) -> dict[str, Any]:
        """Generate comprehensive audit report (REQ-001-US-008-030)."""
        end = end or _utc_now()

        entries = await self.query_by_time_range(start, end, limit=100000)

        # Aggregate statistics
        by_event_type = await self.aggregate_logs("event_type", start, end)
        by_outcome = await self.aggregate_logs("outcome", start, end)
        by_volk = await self.aggregate_logs("volk_id", start, end)

        # Calculate metrics
        total_entries = len(entries)
        unique_sessions = len(set(e.session_id for e in entries))
        unique_volks = len(set(e.volk_id for e in entries))

        # Find errors and escalations
        errors = [e for e in entries if e.event_type == AuditEventType.ERROR]
        escalations = [e for e in entries if e.event_type == AuditEventType.ESCALATION]

        return {
            "report_generated": _utc_now().isoformat(),
            "period": {
                "start": start.isoformat(),
                "end": end.isoformat(),
            },
            "summary": {
                "total_entries": total_entries,
                "unique_sessions": unique_sessions,
                "unique_volks": unique_volks,
                "error_count": len(errors),
                "escalation_count": len(escalations),
            },
            "by_event_type": by_event_type,
            "by_outcome": by_outcome,
            "by_volk": by_volk,
            "recent_errors": [e.to_dict() for e in errors[-10:]],
            "recent_escalations": [e.to_dict() for e in escalations[-10:]],
        }

    @traces(spec="001", story="US-008", reqs=["REQ-001-US-008-026"])
    async def _query_file_logs(
        self,
        filter_fn: Callable[[AuditEntry], bool],
        limit: int = 100,
    ) -> list[AuditEntry]:
        """Query logs from file storage."""
        entries: list[AuditEntry] = []

        # Get all log files sorted by modification time (newest first)
        log_files = sorted(
            self.config.log_dir.glob("audit_*.jsonl*"),
            key=lambda p: p.stat().st_mtime,
            reverse=True,
        )

        for log_file in log_files:
            if len(entries) >= limit:
                break

            try:
                opener = (
                    (lambda p: gzip.open(p, "rt", encoding="utf-8"))
                    if log_file.suffix == ".gz"
                    else (lambda p: open(p, encoding="utf-8"))
                )
                with opener(log_file) as f:
                    self._collect_matching_entries(f, filter_fn, limit, entries)
            except Exception as e:
                _logger.error(f"Error reading log file {log_file}: {e}")

        return entries

    @traces(spec="001", story="US-008", reqs=["REQ-001-US-008-026"])
    @staticmethod
    def _collect_matching_entries(
        lines: Iterable[str],
        filter_fn: Callable[[AuditEntry], bool],
        limit: int,
        entries: list[AuditEntry],
    ) -> None:
        """Append entries from ``lines`` that pass ``filter_fn`` up to ``limit``.

        Malformed lines (bad JSON or missing fields) are skipped.
        """
        for line in lines:
            if len(entries) >= limit:
                break
            try:
                entry = AuditEntry.from_dict(json.loads(line.strip()))
            except json.JSONDecodeError, KeyError:
                continue
            if filter_fn(entry):
                entries.append(entry)

    # ===== Backup and Restore (REQ-001-US-008-019) =====

    @traces(
        spec="001",
        story="US-008",
        reqs=["REQ-001-US-008-019"],
    )
    async def backup_logs(self, backup_path: Path) -> bool:
        """Backup all audit logs to specified path."""
        try:
            backup_path.mkdir(parents=True, exist_ok=True)

            for log_file in self.config.log_dir.glob("audit_*.jsonl*"):
                shutil.copy2(str(log_file), str(backup_path / log_file.name))

            _logger.info(f"Audit logs backed up to {backup_path}")
            return True
        except Exception as e:
            _logger.error(f"Failed to backup logs: {e}")
            return False

    @traces(
        spec="001",
        story="US-008",
        reqs=["REQ-001-US-008-019"],
    )
    async def restore_logs(self, backup_path: Path) -> bool:
        """Restore audit logs from backup."""
        try:
            for log_file in backup_path.glob("audit_*.jsonl*"):
                shutil.copy2(str(log_file), str(self.config.log_dir / log_file.name))

            _logger.info(f"Audit logs restored from {backup_path}")
            return True
        except Exception as e:
            _logger.error(f"Failed to restore logs: {e}")
            return False