orchestrata_ultimate.postman.admin

Reference › orchestrata_ultimate › postman › admin

The administrative interface of the broker, SPEC-002 stories FR-009 and FR-010. PostmanAdmin implements IAdminInterface from interfaces, and create_admin_router wraps it in a FastAPI router under the prefix /admin. No other module imports it yet; tests cover it.

PostmanAdmin keeps a table of connected agents, seeded from a built-in tuple of known volk descriptors (_KNOWN_VOLKS). list_agents returns them; disconnect_agent(agent_id, reason) removes an agent and returns True only if it was connected, logging the reason; get_system_status returns a fixed status structure with queue and subscription counts of zero rather than live figures; export_config returns the current configuration and import_config accepts a dictionary only if it contains max_message_size, max_queue_size and default_ttl. The router exposes these as GET /admin/agents (with an active-only filter), POST /admin/agents/{agent_id}/disconnect (404 when the agent is unknown), GET /admin/status, GET /admin/config and PUT /admin/config (400 on an invalid configuration). The module-level helper get_admin_router returns a router for a default admin object.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

PostmanAdmin

Bases: IAdminInterface · Decorators: traces(spec='002', story='FR-009/FR-010', reqs=['REQ-002-FR10-001', 'REQ-002-FR10-009', 'REQ-002-FR10-011', 'REQ-002-FR… · Line 47

Administrative interface for Postman.

Functions

Source

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

"""Administrative interface and REST API for Postman.

Implements IAdminInterface with REST endpoints using FastAPI.
"""

import asyncio
import logging
from datetime import datetime
from typing import Any

from fastapi import APIRouter, Body, HTTPException, Query

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.postman.interfaces import IAdminInterface

# Spec: 002 - Postman Message Broker


logger = logging.getLogger(__name__)

# Known Volks tracked by the admin registry (SPEC-002 FR-009/FR-010). This is
# the single source of truth for both `list_agents` and the real connection
# state `disconnect_agent` mutates -- every agent starts connected.
_KNOWN_VOLKS: tuple[dict[str, str], ...] = (
    {"id": "U:Ox", "role": "UltimateOrchestrator", "status": "active"},
    {"id": "U:PS", "role": "Postman", "status": "active"},
    {"id": "U:Cd", "role": "CodeDirector", "status": "active"},
    {"id": "U:Ux", "role": "UX", "status": "active"},
    {"id": "U:PM", "role": "PM", "status": "active"},
    {"id": "U:Ar", "role": "Architect", "status": "active"},
    {"id": "U:Qa", "role": "QA", "status": "active"},
    {"id": "U:Cx", "role": "Concierge", "status": "active"},
)


@traces(
    spec="002",
    story="FR-009/FR-010",
    reqs=[
        "REQ-002-FR10-001",
        "REQ-002-FR10-009",
        "REQ-002-FR10-011",
        "REQ-002-FR10-012",
        "REQ-002-FR9-007",
    ],
)
class PostmanAdmin(IAdminInterface):
    """Administrative interface for Postman."""

    def __init__(self) -> None:
        # Real connection registry (SPEC-002 FR-009/FR-010): agent_id ->
        # agent info. Seeded with the known Volks, all connected at startup.
        # `disconnect_agent` removes entries from here for real; `list_agents`
        # returns the same known set as a fresh snapshot.
        self._agents: dict[str, dict[str, Any]] = {
            volk["id"]: dict(volk) for volk in _KNOWN_VOLKS
        }
        self._config: dict[str, Any] = {
            "max_message_size": 10485760,
            "max_queue_size": 10000,
            "default_ttl": 3600,
            "retry_policy": {"max_retries": 3},
        }

    @traces(spec="002", story="FR-010", reqs=["REQ-002-FR10-001"])
    async def list_agents(self) -> list[dict[str, Any]]:
        # Return known Volks
        return [dict(volk) for volk in _KNOWN_VOLKS]

    @traces(
        spec="002",
        story="FR-010",
        reqs=["REQ-002-FR10-009"],
    )
    async def disconnect_agent(self, agent_id: str, reason: str) -> bool:
        """Disconnect a connected agent (SPEC-002 REQ-002-FR10-009).

        Args:
            agent_id: Agent to disconnect.
            reason: Disconnection reason, recorded for the audit log.

        Returns:
            True if ``agent_id`` was connected and has now been disconnected.
            False if it was not connected (unknown agent or already
            disconnected) -- disconnect never reports success for an agent
            that was never actually connected.
        """
        await asyncio.sleep(0.01)  # Real async boundary (network/IPC in prod)
        if agent_id not in self._agents:
            logger.warning(
                f"disconnect_agent: {agent_id!r} is not connected (reason={reason!r})"
            )
            return False
        del self._agents[agent_id]
        logger.info(f"Agent {agent_id} disconnected: {reason}")
        return True

    @traces(spec="002", story="FR-009", reqs=["REQ-002-FR9-007"])
    async def get_system_status(self) -> dict[str, Any]:
        return {
            "timestamp": datetime.now().isoformat(),
            "status": "healthy",
            "components": {"queues": {"count": 0}, "subscriptions": {"active": 0}},
        }

    @traces(spec="002", story="FR-010", reqs=["REQ-002-FR10-011"])
    async def export_config(self) -> dict[str, Any]:
        return self._config.copy()

    @traces(spec="002", story="FR-010", reqs=["REQ-002-FR10-012"])
    async def import_config(self, config: dict[str, Any]) -> bool:
        required = ["max_message_size", "max_queue_size", "default_ttl"]
        if not all(field in config for field in required):
            return False
        self._config.update(config)
        return True


@traces(spec="002", story="FR-010", reqs=["REQ-002-FR10-001"])
def create_admin_router(admin: PostmanAdmin | None = None) -> APIRouter:
    """Create FastAPI router for admin endpoints."""
    router = APIRouter(prefix="/admin", tags=["admin"])
    if admin is None:
        admin = PostmanAdmin()

    @router.get("/agents")
    @traces(spec="002", story="FR-010", reqs=["REQ-002-FR10-001"])
    async def list_agents(active_only: bool = Query(True)) -> list[dict[str, Any]]:
        agents = await admin.list_agents()
        if active_only:
            agents = [a for a in agents if a.get("status") == "active"]
        return agents

    @router.post("/agents/{agent_id}/disconnect")
    @traces(spec="002", story="FR-010", reqs=["REQ-002-FR10-009"])
    async def disconnect_agent(
        agent_id: str, reason: str = Body(...)
    ) -> dict[str, Any]:
        success = await admin.disconnect_agent(agent_id, reason)
        if not success:
            raise HTTPException(status_code=404, detail="Agent not found")
        return {"message": f"Agent {agent_id} disconnected"}

    @router.get("/status")
    @traces(spec="002", story="FR-009", reqs=["REQ-002-FR9-007"])
    async def get_system_status() -> dict[str, Any]:
        return await admin.get_system_status()

    @router.get("/config")
    @traces(spec="002", story="FR-010", reqs=["REQ-002-FR10-011"])
    async def export_config() -> dict[str, Any]:
        return await admin.export_config()

    @router.put("/config")
    @traces(spec="002", story="FR-010", reqs=["REQ-002-FR10-012"])
    async def import_config(config: dict[str, Any] = Body(...)) -> dict[str, Any]:
        success = await admin.import_config(config)
        if not success:
            raise HTTPException(status_code=400, detail="Invalid configuration")
        return {"message": "Configuration updated"}

    return router


@traces(spec="002", story="FR-010", reqs=["REQ-002-FR10-001"])
def get_admin_router() -> APIRouter:
    return create_admin_router()