orchestrata_ultimate.orchestrator.graph_audit

Reference › orchestrata_ultimate › orchestrator › graph_audit

The durable audit trail for governance. GraphAuditTrail receives events from the engine and the governance package and writes them to the knowledge graph without ever blocking or breaking the caller. Its importers are engine, intent_history and the governance modules governance.audit, governance.violations and the escalation handler. It depends on core.graph_factory, core.redis_factory, the knowledge_graph package and its canonical schema, and traces to spec 003 story US-009.

The write path is a queue. Each log_* method (log_event, log_policy_decision, log_session_start, log_session_end, log_tool_call, log_violation, log_escalation, log_resolution, log_timeout) calls _enqueue, which pushes a JSON event onto a Redis list; with no Redis client the event is counted as dropped (get_dropped_events_count). start creates the Redis client when none was injected (from REDIS_HOST, REDIS_PORT and REDIS_DB through the factory) and launches _consumer_loop, a background task that pops events and calls _write_event. That method maps the event type to a graph label, takes the session id (or the escalation id) as anchor, stamps the label's canonical merge key from the canonical schema, writes the node and links it to its Session node, creating the session lazily. Unknown event types and an unavailable graph return false with a warning. stop, close and aclose shut the consumer down.

The data types are AuditEventType (violation, escalation, resolution, timeout, event, policy decision, session start and end, tool call, volk lifecycle, error, compliance check, intent recorded), DecisionOutcome, GraphAuditConfig, RedisConfig, AuditEvent and PolicyDecision. Read methods are get_escalations, get_violations, get_resolutions, query_audit_trail and health_check.

At a glance

Class diagram

Dependency diagram

API

Classes

AuditEventType (enum)

Bases: Enum · Decorators: traces(spec='003', story='US-009', reqs=['REQ-003-US-009-001']) · Line 69

Types of audit events.

DecisionOutcome (enum)

Bases: Enum · Decorators: traces(spec='003', story='US-009', reqs=['REQ-003-US-009-001']) · Line 88

Governance decision outcomes.

GraphAuditConfig (dataclass)

Decorators: dataclass, traces(spec='003', story='US-009', reqs=['REQ-003-US-009-001']) · Line 99

Graph connection configuration.

RedisConfig (dataclass)

Decorators: dataclass, traces(spec='003', story='US-009', reqs=['REQ-003-US-009-001']) · Line 118

Redis connection configuration.

AuditEvent (dataclass)

Decorators: dataclass, traces(spec='003', story='US-009', reqs=['REQ-003-US-009-001']) · Line 133

Audit event to be recorded in graph.

PolicyDecision (dataclass)

Decorators: dataclass, traces(spec='003', story='US-009', reqs=['REQ-003-US-009-001']) · Line 152

Record of a governance decision.

GraphAuditTrail

Decorators: traces(spec='003', story='US-010', reqs=['REQ-003-US-010-006', 'REQ-003-US-010-018', 'REQ-003-US-003-010', 'REQ-003-US-… · Line 178

Manages audit trail in graph for Orchestrata.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/orchestrator/graph_audit.py

# mypy: ignore-errors
"""Graph audit trail for Orchestrata.

Handles cold storage for audit trail, relationships, and compliance.
Based on INFRASTRUCTURE.md specifications.

ASYNC REWRITE: Uses AsyncGraphDatabase + Redis queue for durable fire-and-forget logging.
"""

# Spec: 008 - Knowledge Graph

from __future__ import annotations

import asyncio
import json
import logging
import os
import uuid
from collections.abc import Awaitable
from dataclasses import dataclass, field
from datetime import UTC, datetime
from enum import Enum
from typing import TYPE_CHECKING, Any, cast

import redis.asyncio as redis

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.graph import (
    AsyncGraphRepository,
    GraphRepositoryConfig,
    create_async_graph_repository,
)
from orchestrata_ultimate.core.graph_factory import resolve_kg_database, resolve_uri
from orchestrata_ultimate.core.redis_factory import (
    resolve_redis_host,
    resolve_redis_port,
)
from orchestrata_ultimate.knowledge_graph import canonical_schema as cs

if TYPE_CHECKING:
    from orchestrata_ultimate.governance.escalation_handler import EscalationResult

# Module-level logger
_logger = logging.getLogger(__name__)

# Redis queue key for audit events
QUEUE_KEY = "orchestrata:graph:audit:queue"

_DEFAULT_PROJECT_SLUG = "orchestrata-ultimate"


@traces(spec="003", story="US-009", reqs=["REQ-003-US-009-001"])
def _project_slug() -> str:
    """Resolve the root ``:Project`` slug this writer's hub edge MERGEs onto.

    Read at call time, never at import (Key Rule / RUL-003): the test suite and
    the CLI both override ``ORCHESTRATA_PROJECT_ID`` per process.
    """
    return os.environ.get("ORCHESTRATA_PROJECT_ID", _DEFAULT_PROJECT_SLUG)


@traces(spec="003", story="US-009", reqs=["REQ-003-US-009-001"])
def _utc_now() -> datetime:
    """Get current UTC time with timezone info."""
    return datetime.now(UTC)


@traces(spec="003", story="US-009", reqs=["REQ-003-US-009-001"])
class AuditEventType(Enum):
    """Types of audit events."""

    VIOLATION = "violation"
    ESCALATION = "escalation"
    RESOLUTION = "resolution"
    TIMEOUT = "timeout"
    EVENT = "event"
    POLICY_DECISION = "policy_decision"
    SESSION_START = "session_start"
    SESSION_END = "session_end"
    TOOL_CALL = "tool_call"
    VOLK_LIFECYCLE = "volk_lifecycle"
    ERROR = "error"
    COMPLIANCE_CHECK = "compliance_check"
    INTENT_RECORDED = "intent_recorded"


@traces(spec="003", story="US-009", reqs=["REQ-003-US-009-001"])
class DecisionOutcome(Enum):
    """Governance decision outcomes."""

    ALLOWED = "allowed"
    DENIED = "denied"
    ESCALATED = "escalated"
    PENDING = "pending"


@dataclass
@traces(spec="003", story="US-009", reqs=["REQ-003-US-009-001"])
class GraphAuditConfig:
    """Graph connection configuration."""

    uri: str = field(default_factory=resolve_uri)
    # No credentials: NEO4J_USERNAME/NEO4J_PASSWORD were retired with the Neo4j
    # containers on 2026-07-27 and every remaining endpoint negotiates no auth.
    username: str = ""
    password: str = ""
    # "orchestrata" is not a database on this cluster — see the identical
    # defect already recorded in kernel/audit/model.py. Resolved from
    # KG_DATABASE so an unset config raises instead of silently opening (and
    # writing to) a database of the wrong name.
    database: str = field(default_factory=resolve_kg_database)
    max_connection_lifetime: int = 3600
    max_connection_pool_size: int = 50


@dataclass
@traces(spec="003", story="US-009", reqs=["REQ-003-US-009-001"])
class RedisConfig:
    """Redis connection configuration."""

    # No literal endpoint default (RUL-003/RUL-011): "localhost:6379" would
    # connect to nothing when REDIS_HOST/PORT aren't the target and the
    # resulting refusal reads as an outage instead of the config gap it is.
    host: str = field(default_factory=resolve_redis_host)
    port: int = field(default_factory=resolve_redis_port)
    db: int = 0
    password: str | None = None
    socket_timeout: float = 5.0


@dataclass
@traces(spec="003", story="US-009", reqs=["REQ-003-US-009-001"])
class AuditEvent:
    """Audit event to be recorded in graph."""

    event_type: AuditEventType
    session_id: str
    volk_id: str
    timestamp: datetime
    data: dict[str, Any]
    outcome: DecisionOutcome | None = None
    reason: str | None = None
    correlation_id: str | None = None


# Type alias for queue operations
AuditEntry = AuditEvent


@dataclass
@traces(spec="003", story="US-009", reqs=["REQ-003-US-009-001"])
class PolicyDecision:
    """Record of a governance decision."""

    decision_id: str
    session_id: str
    volk_id: str
    intent: str
    tool: str
    outcome: DecisionOutcome
    reason: str
    timestamp: datetime
    prolog_query: str | None = None
    prolog_result: str | None = None
    cached: bool = False


@traces(
    spec="003",
    story="US-010",
    reqs=[
        "REQ-003-US-010-006",
        "REQ-003-US-010-018",
        "REQ-003-US-003-010",
        "REQ-003-US-003-022",
    ],
)
class GraphAuditTrail:
    """Manages audit trail in graph for Orchestrata.

    Uses Redis queue for durable async logging with background consumer.
    """

    @traces(spec="003", story="US-010", reqs=["REQ-003-US-010-018"])
    def __init__(
        self,
        config: GraphAuditConfig | None = None,
        redis_client: redis.Redis | None = None,
        redis_config: RedisConfig | None = None,
    ):
        """Initialize graph audit trail.

        Args:
            config: Graph configuration. Uses defaults if not provided.
            redis_client: Optional Redis client for queueing events. When not
                provided, a real async client is created in ``start()`` from
                ``redis_config`` (or environment defaults), so audit events are
                durably enqueued rather than silently dropped.
            redis_config: Optional Redis connection configuration used to build
                the client when ``redis_client`` is not injected.
        """
        self.config = config or GraphAuditConfig()
        self._repo: AsyncGraphRepository | None = None
        self._redis: redis.Redis | None = redis_client
        # When no explicit RedisConfig is given, the internally-created client
        # reads connection details from the environment (REDIS_HOST/PORT/DB)
        # via redis_factory, mirroring fact_store.
        self._redis_config = redis_config
        # Whether _redis was created internally (and must be closed by us).
        self._owns_redis = False
        self._consumer_task: asyncio.Task | None = None
        self._running = False
        self._closed = False
        self._dropped_events = 0

    @traces(spec="003", story="US-010", reqs=["REQ-003-US-010-018"])
    async def start(self) -> None:
        """Initialize driver and start background consumer."""
        if self._running:
            return
        self._repo = create_async_graph_repository(
            GraphRepositoryConfig(
                uri=self.config.uri,
                auth=(self.config.username, self.config.password),
            )
        )
        # Wire a real async Redis client when none was injected so audit
        # events are actually enqueued instead of dropped (the no-redis path
        # silently increments _dropped_events).
        if self._redis is None:
            self._redis = self._create_redis_client()
            self._owns_redis = self._redis is not None
        self._running = True
        self._consumer_task = asyncio.create_task(self._consumer_loop())

    @traces(spec="003", story="US-010", reqs=["REQ-003-US-010-018"])
    def _create_redis_client(self) -> redis.Redis | None:
        """Create an async Redis client from config / environment defaults.

        Mirrors fact_store via the shared redis_factory so connection handling
        is consistent. Returns None if the client cannot be constructed.
        """
        try:
            from orchestrata_ultimate.core.redis_factory import create_async_redis

            cfg = self._redis_config
            return create_async_redis(
                host=cfg.host if cfg else None,
                port=cfg.port if cfg else None,
                db=cfg.db if cfg else None,
                password=cfg.password if cfg else None,
                decode_responses=True,
            )
        except Exception as e:  # pragma: no cover - defensive
            _logger.error(f"Failed to create Redis client for audit trail: {e}")
            return None

    @traces(spec="003", story="US-010", reqs=["REQ-003-US-010-018"])
    async def stop(self) -> None:
        """Stop consumer and close connections, draining queue first."""
        self._running = False
        if self._consumer_task:
            self._consumer_task.cancel()
            try:
                await self._consumer_task
            except asyncio.CancelledError:
                _logger.debug("Graph audit consumer task cancelled")
        # Drain remaining queue items
        if self._redis:
            while True:
                # redis-py's command mixins are shared between the sync and
                # async clients, so `rpop()` is stub-typed as the union
                # `Awaitable[T] | T`; `self._redis` here is always the async
                # client, so it always actually returns an awaitable.
                result = await cast("Awaitable[Any]", self._redis.rpop(QUEUE_KEY))
                if not result:
                    break
                try:
                    event = json.loads(result)
                    await self._write_event(event)
                except Exception as e:
                    _logger.error(f"Error draining queue: {e}")
        if self._repo:
            await self._repo.close()
            self._repo = None
        # Close the Redis client only if we created it; never close a
        # caller-injected client.
        if self._owns_redis and self._redis is not None:
            try:
                await self._redis.aclose()
            except Exception as e:  # pragma: no cover - defensive
                _logger.debug(f"Error closing audit Redis client: {e}")
            self._redis = None
            self._owns_redis = False

    @traces(spec="003", story="US-010", reqs=["REQ-003-US-010-006"])
    async def _enqueue(self, event_type: str, data: dict) -> bool:
        """Push event to Redis queue (fire-and-forget)."""
        if not self._redis:
            self._dropped_events += 1
            return False
        event = {
            "event_type": event_type,
            "timestamp": _utc_now().isoformat(),
            "data": data,
        }
        # Stub deficiency (see stop()'s rpop cast).
        await cast("Awaitable[Any]", self._redis.lpush(QUEUE_KEY, json.dumps(event)))
        return True

    @traces(spec="003", story="US-010", reqs=["REQ-003-US-010-006"])
    async def _consumer_loop(self) -> None:
        """Background task: drain Redis queue, write to graph."""
        if not self._redis:
            return
        while self._running:
            try:
                # Stub deficiency (see stop()'s rpop cast).
                result = await cast(
                    "Awaitable[Any]", self._redis.brpop(QUEUE_KEY, timeout=1)
                )
                if result:
                    _, event_json = result
                    event = json.loads(event_json)
                    await self._write_event(event)
            except asyncio.CancelledError:
                break
            except Exception as e:
                _logger.error(f"Consumer error: {e}")
                await asyncio.sleep(1)

    # Map event_type string → graph node label.
    _EVENT_LABELS: dict[str, str] = {
        "violation": "Violation",
        "escalation": "Escalation",
        "resolution": "Resolution",
        "timeout": "Timeout",
        "event": "Event",
        "policy_decision": "PolicyDecision",
        "session_start": "SessionStart",
        "session_end": "SessionEnd",
        "tool_call": "ToolCall",
        "volk_lifecycle": "VolkLifecycle",
        "error": "Error",
        "compliance_check": "ComplianceCheck",
    }

    @traces(spec="003", story="US-010", reqs=["REQ-003-US-010-006"])
    async def _write_event(self, event: dict[str, Any]) -> bool:
        """Write event to graph and link to its Session.

        Creates the event node with data, timestamp, and session_id
        as first-class properties, then links to the Session node
        (creating it lazily via MERGE if it doesn't exist yet).

        Returns:
            True if the write landed, False if the graph is unavailable,
            the event type is unrecognized, or the write itself failed.
            Non-fatal either way — the audit trail is a side-channel, never
            a reason to break the caller.
        """
        if not self._repo:
            return False
        event_type = event.get("event_type")
        label = self._EVENT_LABELS.get(event_type) if event_type is not None else None
        if not label:
            _logger.warning("Unknown event type: %s", event_type)
            return False

        data = event.get("data", {})
        data_json = json.dumps(data)
        ts = event.get("timestamp")
        raw_session_id = data.get("session_id", "")
        # When the emitter provides no session_id (e.g. log_timeout/log_resolution
        # only carry escalation_id), derive a session from escalation_id so the
        # event is still reachable via the canonical IN_SESSION + HAS_SESSION
        # path. This avoids minting zero-degree orphans and avoids inventing a
        # non-canonical HAS_AUDIT_EVENT edge (RUL-016).
        session_id = raw_session_id or data.get("escalation_id", "")
        if not session_id:
            _logger.warning(
                "Audit event %s has no session_id or escalation_id; "
                "writing without session anchor",
                event_type,
            )
        # Without the canonical key (canonical_schema.CANONICAL_MERGE_KEYS)
        # these nodes fail the kg_schema conformance check as unkeyed
        # pollution. Most event labels key on `id`, but not all — Resolution
        # keys on `resolution_id` (the escalation resolver's natural id), and
        # writing audit Resolutions keyed only on `id` split the label across
        # two identities (D3, plans/hub-link-plan.md). Stamp the label's
        # declared key alongside `id` whenever it differs.
        event_id = str(uuid.uuid4())
        key_prop = cs.CANONICAL_MERGE_KEYS.get(label, "id")
        key_fragment = f", {key_prop}: $id" if key_prop != "id" else ""
        project = _project_slug()

        try:
            if session_id:
                await self._repo.execute_write(
                    f"CREATE (n:{label} {{id: $id{key_fragment}, data: $data, "
                    f"  timestamp: $ts, session_id: $sid}}) "
                    f"WITH n "
                    f"MERGE (s:{cs.LABEL_SESSION} {{session_id: $sid}}) "
                    f"MERGE (n)-[:IN_SESSION]->(s) "
                    f"WITH s "
                    f"MERGE (p:{cs.LABEL_PROJECT} {{slug: $project}}) "
                    f"MERGE (p)-[:{cs.REL_HAS_SESSION}]->(s)",
                    {
                        "id": event_id,
                        "data": data_json,
                        "ts": ts,
                        "sid": session_id,
                        "project": project,
                    },
                )
            else:
                await self._repo.execute_write(
                    f"CREATE (n:{label} {{id: $id{key_fragment}, "
                    f"data: $data, timestamp: $ts}})",
                    {"id": event_id, "data": data_json, "ts": ts},
                )
            return True
        except Exception as e:
            _logger.error("Failed to write %s event: %s", event_type, e)
            return False

    @traces(
        spec="003",
        story="US-010",
        reqs=[
            "REQ-003-US-010-006",
            "REQ-003-US-010-018",
            "REQ-003-US-010-024",
        ],
    )
    async def log_violation(self, violation: dict) -> None:
        """Log a violation event."""
        await self._enqueue("violation", violation)

    @traces(spec="003", story="US-010", reqs=["REQ-003-US-010-024"])
    async def log_escalation(self, result: EscalationResult) -> None:
        """Log an escalation event."""
        data = {
            "escalation_id": result.escalation_id,
            "conflict_id": result.conflict.conflict_id,
            "conflict_type": result.conflict.conflict_type.value,
            "resolved": getattr(result, "resolved", result.resolved_at is not None),
            "resolution": result.resolution,
            "total_time_ms": getattr(result, "total_time_ms", 0),
        }
        await self._enqueue("escalation", data)

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-005-024"])
    async def log_resolution(self, escalation_id: str, resolution: Any) -> None:
        """Log a resolution event."""
        data = {
            "escalation_id": escalation_id,
            "resolution_text": resolution.resolution_text,
            "resolved_by": resolution.resolved_by,
        }
        await self._enqueue("resolution", data)

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-025"])
    async def log_timeout(self, escalation_id: str) -> None:
        """Log a timeout event."""
        await self._enqueue("timeout", {"escalation_id": escalation_id})

    @traces(
        spec="003",
        story="US-010",
        reqs=[
            "REQ-003-US-010-006",
            "REQ-003-US-010-010",
            "REQ-003-US-010-018",
            "REQ-003-US-010-023",
            "REQ-003-US-010-036",
            "REQ-003-US-010-040",
        ],
    )
    async def log_event(self, event: AuditEvent) -> bool:
        """Log an audit event asynchronously."""
        return await self._enqueue(
            "event",
            {
                "event_type": event.event_type.value,
                "session_id": event.session_id,
                "volk_id": event.volk_id,
                "timestamp": event.timestamp.isoformat(),
                "data": event.data,
                "outcome": event.outcome.value if event.outcome else None,
                "reason": event.reason,
                "correlation_id": event.correlation_id,
            },
        )

    @traces(
        spec="003",
        story="US-003",
        reqs=[
            "REQ-003-US-003-010",
            "REQ-003-US-003-022",
            "REQ-003-US-003-034",
            "REQ-003-US-005-005",
        ],
    )
    async def log_policy_decision(self, decision: PolicyDecision) -> None:
        """Log policy decision event."""
        await self._enqueue(
            "policy_decision",
            {
                "decision_id": decision.decision_id,
                "session_id": decision.session_id,
                "volk_id": decision.volk_id,
                "intent": decision.intent,
                "tool": decision.tool,
                "outcome": decision.outcome.value,
                "reason": decision.reason,
                "timestamp": decision.timestamp.isoformat(),
                "prolog_query": decision.prolog_query,
                "prolog_result": decision.prolog_result,
                "cached": decision.cached,
            },
        )

    @traces(spec="003", story="US-010", reqs=["REQ-003-US-010-006"])
    async def log_session_start(
        self, session_id: str, volk_id: str, goals: list[str], context: dict
    ) -> None:
        """Log session start event."""
        await self._enqueue(
            "session_start",
            {
                "session_id": session_id,
                "volk_id": volk_id,
                "goals": goals,
                "context": context,
            },
        )

    @traces(spec="003", story="US-010", reqs=["REQ-003-US-010-006"])
    async def log_session_end(
        self, session_id: str, volk_id: str, outcome: str, stats: dict
    ) -> None:
        """Log session end event."""
        await self._enqueue(
            "session_end",
            {
                "session_id": session_id,
                "volk_id": volk_id,
                "outcome": outcome,
                "stats": stats,
            },
        )

    @traces(spec="003", story="US-010", reqs=["REQ-003-US-010-006"])
    async def log_tool_call(
        self,
        session_id: str,
        volk_id: str,
        tool: str,
        args: dict,
        result: Any,
        duration_ms: int,
    ) -> None:
        """Log tool call event."""
        await self._enqueue(
            "tool_call",
            {
                "session_id": session_id,
                "volk_id": volk_id,
                "tool": tool,
                "args": args,
                "result": str(result),
                "duration_ms": duration_ms,
            },
        )

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-024"])
    async def get_escalations(self, limit: int = 100) -> list[dict]:
        """Query escalation nodes from graph."""
        if not self._repo:
            return []
        rows = await self._repo.execute_read(
            "MATCH (e:Escalation) RETURN e ORDER BY e.timestamp DESC LIMIT $limit",
            {"limit": limit},
        )
        return [json.loads(r["e"]["data"]) for r in rows]

    @traces(spec="007", story="FR-PGOV-007", reqs=["REQ-007-FR7-008"])
    async def get_violations(self, limit: int = 100) -> list[dict]:
        """Query violation nodes from graph."""
        if not self._repo:
            return []
        rows = await self._repo.execute_read(
            "MATCH (v:Violation) RETURN v ORDER BY v.timestamp DESC LIMIT $limit",
            {"limit": limit},
        )
        return [json.loads(r["v"]["data"]) for r in rows]

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-024"])
    async def get_resolutions(self, limit: int = 100) -> list[dict]:
        """Query resolution nodes from graph."""
        if not self._repo:
            return []
        rows = await self._repo.execute_read(
            "MATCH (r:Resolution) RETURN r ORDER BY r.resolved_at DESC LIMIT $limit",
            {"limit": limit},
        )
        resolutions = []
        for r in rows:
            node = dict(r["r"])
            if "data" in node:
                resolutions.append(json.loads(node["data"]))
            else:
                resolutions.append(node)
        return resolutions

    @traces(
        spec="003",
        story="US-010",
        reqs=["REQ-003-US-010-022", "REQ-003-US-010-037"],
    )
    async def query_audit_trail(self, filters: dict) -> list[dict]:
        """Query audit trail with filters."""
        if not self._repo:
            return []
        # Build query based on filters
        node_type = filters.get("event_type", "Event")
        session_id = filters.get("session_id")
        limit = filters.get("limit", 100)

        query = f"MATCH (n:{node_type})"
        params = {"limit": limit}

        if session_id:
            query += " WHERE n.data CONTAINS $session_id"
            params["session_id"] = session_id

        query += " RETURN n ORDER BY n.timestamp DESC LIMIT $limit"

        rows = await self._repo.execute_read(query, params)
        return [json.loads(r["n"]["data"]) for r in rows]

    async def health_check(self) -> bool:
        """Check graph connection health."""
        if self._closed or not self._repo:
            return False
        try:
            await self._repo.execute_read("RETURN 1")
            return True
        except Exception as e:
            _logger.error(f"Graph health check failed: {e}")
            return False

    @traces(spec="003", story="US-010", reqs=["REQ-003-US-010-023"])
    def get_dropped_events_count(self) -> int:
        """Get the count of dropped events due to Redis unavailability.

        Returns:
            Number of dropped events.
        """
        return self._dropped_events

    @traces(spec="003", story="US-010", reqs=["REQ-003-US-010-018"])
    async def close(self) -> None:
        """Close graph connections."""
        self._closed = True
        await self.stop()

    @traces(spec="003", story="US-010", reqs=["REQ-003-US-010-018"])
    async def aclose(self) -> None:
        """Async close alias for close()."""
        await self.close()