alex/headquarter
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'dev' of ssh://git.commumedia.org:2222/alex/headquarter into dev

Browse Source
This commit is contained in:
Alex Blank
2026-05-29 10:35:47 +02:00
parent d2b1c132d1 2682e0268c
commit 569876538a
46 changed files with 6599 additions and 1380 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/src/api/__init__.py
+2 -1
View File
@@ -1,4 +1,5 @@
from src.api.auth import router as auth_router
from src.api.events import router as events_router
from src.api.users import router as users_router
__all__ = ["auth_router", "users_router"]
__all__ = ["auth_router", "events_router", "users_router"]
apps/api/src/api/events.py
+80
View File
@@ -0,0 +1,80 @@
"""SSE streaming endpoint for instance events."""
import asyncio
import contextlib
import json
import uuid
from collections.abc import AsyncGenerator
from fastapi import APIRouter, Depends, HTTPException, Request, status
from fastapi.responses import StreamingResponse
from src.auth.dependencies import get_current_user_id
from src.services.event_bus import InstanceEventBus, InstanceEventPayload
router = APIRouter(prefix="/events", tags=["events"])
# In-memory connection counter per user (single-process assumption)
_connection_counts: dict[uuid.UUID, int] = {}
MAX_CONNECTIONS_PER_USER = 5
@router.get("/stream")
async def events_stream(
request: Request,
user_id: uuid.UUID = Depends(get_current_user_id),
) -> StreamingResponse:
"""Stream instance events via Server-Sent Events.
Enforces a maximum of 5 concurrent connections per user.
"""
current = _connection_counts.get(user_id, 0)
if current >= MAX_CONNECTIONS_PER_USER:
raise HTTPException(
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
detail="Too many SSE connections",
)
_connection_counts[user_id] = current + 1
async def event_generator() -> AsyncGenerator[str, None]:
event_bus = InstanceEventBus()
queue: asyncio.Queue[InstanceEventPayload] = asyncio.Queue(maxsize=100)
async def on_event(payload: InstanceEventPayload) -> None:
try:
queue.put_nowait(payload)
except asyncio.QueueFull:
# Drop oldest event to make room
with contextlib.suppress(asyncio.QueueEmpty):
queue.get_nowait()
with contextlib.suppress(asyncio.QueueFull):
queue.put_nowait(payload)
unsubscribe = event_bus.subscribe("*", on_event)
try:
while True:
try:
payload = await asyncio.wait_for(queue.get(), timeout=30.0)
yield f"event: {payload['event']}\ndata: {json.dumps(payload)}\n\n"
except asyncio.TimeoutError:
yield ":ping\n\n"
except asyncio.CancelledError:
# Client disconnected
raise
finally:
unsubscribe()
_connection_counts[user_id] = max(0, _connection_counts.get(user_id, 1) - 1)
if _connection_counts[user_id] == 0:
_connection_counts.pop(user_id, None)
return StreamingResponse(
event_generator(),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no",
},
)
apps/api/src/api/tool_instances.py
+149
View File
@@ -28,6 +28,8 @@ from src.auth.dependencies import (
get_current_user_id,
get_db_session,
)
from src.services.event_bus import InstanceEventBus
from src.services.lifecycle_hooks import publish_lifecycle_event
from src.models.config_profile import ConfigProfile
from src.models.git_repository import GitRepository
from src.models.project import Project
@@ -77,6 +79,7 @@ from src.services.readiness_probe import execute_probe
from src.services.ssh_keys import cleanup_ssh_key_files, prepare_ssh_key_files
logger = logging.getLogger(__name__)
_event_bus = InstanceEventBus()
async def _resolve_git_mounts(
@@ -958,6 +961,15 @@ services:
session.add(instance)
await session.commit()
await session.refresh(instance)
await publish_lifecycle_event(
event_bus=_event_bus,
session=session,
instance=instance,
event_type="instance.created",
created_by=user_id,
status="pending",
message="Instance created",
)
return {
"id": str(instance.id),
@@ -1502,6 +1514,15 @@ async def start_instance(
instance.status = "starting"
instance.last_started_at = datetime.now()
await session.commit()
await publish_lifecycle_event(
event_bus=_event_bus,
session=session,
instance=instance,
event_type="instance.started",
created_by=user_id,
status="starting",
message="Container starting...",
)
logger.debug("Instance %s: verifying container startup...", instance.id)
startup_result = wait_for_container_running(
@@ -1519,6 +1540,19 @@ async def start_instance(
instance.status = "error"
await session.commit()
await publish_lifecycle_event(
event_bus=_event_bus,
session=session,
instance=instance,
event_type="instance.error",
created_by=user_id,
status="error",
message=error_msg,
metadata={
"exit_code": startup_result["exit_code"],
"error_type": "container",
},
)
logger.error(
"Instance %s container startup failed after %.1fs: %s\nLogs:\n%s",
instance.id,
@@ -1608,6 +1642,16 @@ async def start_instance(
if not success:
instance.status = "unhealthy"
await session.commit()
await publish_lifecycle_event(
event_bus=_event_bus,
session=session,
instance=instance,
event_type="instance.health_changed",
created_by=user_id,
status="unhealthy",
message="Readiness probe failed",
metadata={"probe_output": "\n".join(probe_logs)},
)
logger.error(
"Readiness probe failed for instance %s after %ds: %s",
instance.id,
@@ -1624,6 +1668,16 @@ async def start_instance(
instance.status = "running"
await session.commit()
await publish_lifecycle_event(
event_bus=_event_bus,
session=session,
instance=instance,
event_type="instance.health_changed",
created_by=user_id,
status="running",
message="Container running",
metadata={"previous_status": "starting"},
)
logger.info("Instance %s is now running", instance.id)
# Get tool type for default port
@@ -1632,6 +1686,15 @@ async def start_instance(
logger.error("Tool type %s not found", instance.tool_type_id)
instance.status = "error"
await session.commit()
await publish_lifecycle_event(
event_bus=_event_bus,
session=session,
instance=instance,
event_type="instance.error",
created_by=user_id,
status="error",
message=f"Tool type '{instance.tool_type_id}' not found",
)
return {
"status": "error",
"error": f"Tool type '{instance.tool_type_id}' not found",
@@ -1757,6 +1820,15 @@ async def stop_instance(
instance.public_url = None
instance.tunnel_id = None
await session.commit()
await publish_lifecycle_event(
event_bus=_event_bus,
session=session,
instance=instance,
event_type="instance.stopped",
created_by=user_id,
status="stopped",
message="Instance stopped",
)
return {"status": instance.status}
@@ -1893,6 +1965,15 @@ async def restart_instance(
instance.public_url = None
await session.commit()
await publish_lifecycle_event(
event_bus=_event_bus,
session=session,
instance=instance,
event_type="instance.restarted",
created_by=user_id,
status="running",
message="Instance restarted",
)
return {"status": instance.status, "url": instance.url}
instance.status = "error"
@@ -1979,6 +2060,15 @@ async def delete_instance(
shutil.rmtree(instance_dir)
await publish_lifecycle_event(
event_bus=_event_bus,
session=session,
instance=instance,
event_type="instance.deleted",
created_by=user_id,
status="deleted",
message="Instance deleted",
)
await session.delete(instance)
await session.commit()
@@ -2195,6 +2285,65 @@ async def check_instance_tunnel_health(
return response
@router.get(
"/{project_id}/repositories/{repo_id}/instances/{instance_id}/events",
summary="Get instance events history",
description="Get lifecycle event history for a tool instance.",
)
async def get_instance_events(
project_id: uuid.UUID,
repo_id: uuid.UUID,
instance_id: uuid.UUID,
limit: int = 50,
user_id: uuid.UUID = Depends(get_current_user_id),
session: AsyncSession = Depends(get_db_session),
) -> list[dict]:
"""Get lifecycle event history for an instance.
Args:
project_id: UUID of the project.
repo_id: UUID of the repository.
instance_id: UUID of the instance.
limit: Maximum number of events to return (default: 50).
user_id: ID of the authenticated user.
session: Database session.
Returns:
List of event dictionaries.
"""
from sqlalchemy import select
from src.models.instance_event import InstanceEvent
_user = await _get_user(session, user_id)
_project = await _get_owned_project(project_id, user_id, session)
instance = await session.get(ToolInstance, instance_id)
if instance is None or instance.repository_id != repo_id:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="instance not found"
)
result = await session.execute(
select(InstanceEvent)
.where(InstanceEvent.instance_id == instance_id)
.order_by(InstanceEvent.created_at.desc())
.limit(limit)
)
rows = result.scalars().all()
return [
{
"id": str(row.id),
"event_type": row.event_type,
"status": row.status,
"message": row.message,
"metadata": row.event_metadata,
"created_at": row.created_at.isoformat() if row.created_at else None,
}
for row in rows
]
@router.get(
"/{project_id}/repositories/{repo_id}/instances/{instance_id}/proxy/{path:path}",
summary="Proxy to instance",
apps/api/src/logging_config.py
+41 -8
View File
@@ -1,15 +1,52 @@
"""Structured JSON logging configuration."""
import json
import logging
import sys
import time
import traceback
from typing import Callable
from collections.abc import Callable
from fastapi import Request, Response
from starlette.middleware.base import BaseHTTPMiddleware
from src.services.correlation import get_correlation_id
logger = logging.getLogger(__name__)
class CorrelationIdFilter(logging.Filter):
"""Inject correlation_id into every log record from context var."""
def filter(self, record: logging.LogRecord) -> bool:
record.correlation_id = get_correlation_id() # type: ignore[attr-defined]
return True
class JSONFormatter(logging.Formatter):
"""Emit log records as single-line JSON."""
def format(self, record: logging.LogRecord) -> str:
log_obj: dict = {
"timestamp": self.formatTime(record),
"level": record.levelname,
"logger": record.name,
"message": record.getMessage(),
"correlation_id": getattr(record, "correlation_id", None),
}
# Optional extra fields
for key in ("instance_id", "event_type"):
value = getattr(record, key, None)
if value is not None:
log_obj[key] = value
if record.exc_info:
log_obj["exception"] = self.formatException(record.exc_info)
return json.dumps(log_obj, default=str)
def formatTime(self, record: logging.LogRecord, datefmt: str | None = None) -> str:
return time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime(record.created))
class RequestLoggingMiddleware(BaseHTTPMiddleware):
"""Log all HTTP requests with timing and status codes."""
@@ -17,7 +54,6 @@ class RequestLoggingMiddleware(BaseHTTPMiddleware):
start_time = time.time()
client_host = request.client.host if request.client else "unknown"
# Log the incoming request
logger.info(
"→ Request: %s %s (client: %s)",
request.method,
@@ -29,7 +65,6 @@ class RequestLoggingMiddleware(BaseHTTPMiddleware):
response = await call_next(request)
duration = time.time() - start_time
# Log the response
logger.info(
"← Response: %s %s → %d (%dms)",
request.method,
@@ -69,15 +104,13 @@ class ExceptionLoggingMiddleware(BaseHTTPMiddleware):
def configure_logging(level: int = logging.INFO) -> None:
"""Configure structured logging for the application."""
formatter = logging.Formatter(
fmt="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
"""Configure structured JSON logging for the application."""
formatter = JSONFormatter()
# Console handler
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setFormatter(formatter)
console_handler.addFilter(CorrelationIdFilter())
# Configure root logger
root_logger = logging.getLogger()
apps/api/src/main.py
+23
View File
@@ -9,6 +9,7 @@ from fastapi.staticfiles import StaticFiles
from src.api.auth import router as auth_router
from src.api.dashboard import router as dashboard_router
from src.api.events import router as events_router
from src.api.git_repositories import router as git_repositories_router
from src.api.health import router as health_router
from src.api.projects import router as projects_router
@@ -30,6 +31,9 @@ from src.logging_config import (
RequestLoggingMiddleware,
configure_logging,
)
from src.services.correlation import CorrelationIdMiddleware
from src.services.event_bus import InstanceEventBus
from src.services.health_monitor import HealthMonitor
# Configure logging early
log_level = os.getenv("LOG_LEVEL", "INFO").upper()
@@ -54,6 +58,7 @@ app.add_middleware(
allow_headers=["*"],
)
app.add_middleware(CorrelationIdMiddleware)
app.add_middleware(RequestLoggingMiddleware)
app.add_middleware(ExceptionLoggingMiddleware)
@@ -103,6 +108,11 @@ async def validation_exception_handler(request: Request, exc: RequestValidationE
)
# Global services
_event_bus = InstanceEventBus()
_health_monitor = HealthMonitor(_event_bus)
@app.on_event("startup")
async def on_startup():
logger.info("Starting up Headquarter API...")
@@ -115,9 +125,21 @@ async def on_startup():
sys.exit(1)
# Start background health monitor
_health_monitor.start()
logger.info("Health monitor started")
logger.info("Startup complete.")
@app.on_event("shutdown")
async def on_shutdown():
logger.info("Shutting down Headquarter API...")
_health_monitor.stop()
logger.info("Health monitor stopped")
logger.info("Shutdown complete.")
app.include_router(health_router)
app.include_router(auth_router)
app.include_router(dashboard_router)
@@ -133,4 +155,5 @@ app.include_router(tool_instances_router)
app.include_router(sessions_router)
app.include_router(instance_proxy_router)
app.include_router(terminal_router)
app.include_router(events_router)
app.mount("/uploads", StaticFiles(directory="uploads"), name="uploads")
apps/api/src/models/__init__.py
+4
View File
@@ -1,6 +1,8 @@
from src.models.base import Base
from src.models.config_profile import ConfigProfile, ConfigProfileInclude
from src.models.git_repository import GitRepository
from src.models.health_check import HealthCheck
from src.models.instance_event import InstanceEvent
from src.models.project import Project
from src.models.ssh_key import SSHKey
from src.models.terminal_session import TerminalSessionModel
@@ -15,6 +17,8 @@ __all__ = [
"ConfigProfile",
"ConfigProfileInclude",
"GitRepository",
"HealthCheck",
"InstanceEvent",
"Project",
"SSHKey",
"TerminalSessionModel",
apps/api/src/models/health_check.py
+30
View File
@@ -0,0 +1,30 @@
"""SQLAlchemy model for health check snapshots."""
import uuid
from datetime import datetime
from sqlalchemy import Boolean, DateTime, ForeignKey, Integer, String, Text, Uuid, func
from sqlalchemy.orm import Mapped, mapped_column
from src.models.base import Base, UUIDPrimaryKeyMixin
class HealthCheck(UUIDPrimaryKeyMixin, Base):
__tablename__ = "health_checks"
instance_id: Mapped[uuid.UUID] = mapped_column(
Uuid(as_uuid=True),
ForeignKey("tool_instances.id", ondelete="CASCADE"),
nullable=False,
)
container_status: Mapped[str | None] = mapped_column(String(50), nullable=True)
container_healthy: Mapped[bool | None] = mapped_column(Boolean, nullable=True)
tunnel_healthy: Mapped[bool | None] = mapped_column(Boolean, nullable=True)
exit_code: Mapped[int | None] = mapped_column(Integer, nullable=True)
probe_status: Mapped[str | None] = mapped_column(String(50), nullable=True)
probe_output: Mapped[str | None] = mapped_column(Text, nullable=True)
checked_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
server_default=func.now(),
nullable=False,
)
apps/api/src/models/instance_event.py
+39
View File
@@ -0,0 +1,39 @@
"""SQLAlchemy model for instance lifecycle event audit rows."""
import uuid
from datetime import datetime
from typing import Any
from sqlalchemy import DateTime, ForeignKey, JSON, String, Text, Uuid, func
from sqlalchemy.orm import Mapped, mapped_column
from src.models.base import Base, UUIDPrimaryKeyMixin
class InstanceEvent(UUIDPrimaryKeyMixin, Base):
__tablename__ = "instance_events"
instance_id: Mapped[uuid.UUID] = mapped_column(
Uuid(as_uuid=True),
ForeignKey("tool_instances.id", ondelete="CASCADE"),
nullable=False,
)
event_type: Mapped[str] = mapped_column(String(50), nullable=False)
status: Mapped[str | None] = mapped_column(String(50), nullable=True)
message: Mapped[str | None] = mapped_column(Text, nullable=True)
created_by: Mapped[uuid.UUID | None] = mapped_column(
Uuid(as_uuid=True),
ForeignKey("users.id", ondelete="SET NULL"),
nullable=True,
)
event_metadata: Mapped[dict[str, Any]] = mapped_column(
"metadata",
JSON,
nullable=False,
default=dict,
)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
server_default=func.now(),
nullable=False,
)
apps/api/src/services/correlation.py
+32
View File
@@ -0,0 +1,32 @@
"""Async correlation ID context variable and helpers."""
import contextvars
import uuid
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
CORRELATION_ID: contextvars.ContextVar[str] = contextvars.ContextVar("correlation_id")
def get_correlation_id() -> str:
"""Return the current correlation ID or generate a new UUID."""
try:
return CORRELATION_ID.get()
except LookupError:
return str(uuid.uuid4())
class CorrelationIdMiddleware(BaseHTTPMiddleware):
"""Set correlation ID from X-Request-ID header or generate a new UUID."""
async def dispatch(self, request: Request, call_next):
request_id = request.headers.get("X-Request-ID")
correlation_id = request_id or str(uuid.uuid4())
token = CORRELATION_ID.set(correlation_id)
try:
response = await call_next(request)
response.headers["X-Request-ID"] = correlation_id
return response
finally:
CORRELATION_ID.reset(token)
apps/api/src/services/event_bus.py
+97
View File
@@ -0,0 +1,97 @@
"""In-memory typed event bus for instance lifecycle and health events."""
import asyncio
import inspect
import logging
import uuid
from collections.abc import Awaitable, Callable
from typing import Any
logger = logging.getLogger(__name__)
InstanceEventPayload = dict[str, Any]
EventCallback = Callable[[InstanceEventPayload], Awaitable[None] | None] # noqa: UP044
class InstanceEventBus:
"""Singleton in-memory event bus with typed pub/sub and exception isolation."""
_instance: "InstanceEventBus | None" = None
_lock: asyncio.Lock = asyncio.Lock()
def __init__(self) -> None:
self._subscribers: dict[str, list[tuple[str, EventCallback]]] = {}
def __new__(cls) -> "InstanceEventBus":
if cls._instance is None:
cls._instance = super().__new__(cls)
cls._instance._subscribers = {}
return cls._instance
def _reset_for_testing(self) -> None:
"""Clear all subscribers. For test use only."""
self._subscribers.clear()
def subscribe(
self,
event_type: str,
callback: EventCallback,
) -> Callable[[], None]:
"""Register a callback for an event type.
Args:
event_type: The event type to subscribe to.
callback: A sync or async callable that receives the payload.
Returns:
An unsubscribe function.
"""
if event_type not in self._subscribers:
self._subscribers[event_type] = []
callback_id = str(uuid.uuid4())
self._subscribers[event_type].append((callback_id, callback))
def unsubscribe() -> None:
self.unsubscribe(event_type, callback_id)
return unsubscribe
def unsubscribe(self, event_type: str, callback_id: str) -> None:
"""Remove a specific callback by ID."""
if event_type in self._subscribers:
self._subscribers[event_type] = [
(cid, cb)
for cid, cb in self._subscribers[event_type]
if cid != callback_id
]
if not self._subscribers[event_type]:
del self._subscribers[event_type]
def unsubscribe_all(self, event_type: str) -> None:
"""Remove all subscribers for an event type."""
self._subscribers.pop(event_type, None)
async def publish(self, event_type: str, payload: InstanceEventPayload) -> None:
"""Deliver payload to all subscribers of event_type.
Also delivers to subscribers registered under the wildcard "*".
Exceptions from individual subscribers are caught and logged;
delivery continues to remaining subscribers.
"""
callbacks: list[tuple[str, EventCallback]] = []
callbacks.extend(self._subscribers.get(event_type, []))
callbacks.extend(self._subscribers.get("*", []))
for _callback_id, callback in callbacks:
try:
if inspect.iscoroutinefunction(callback):
await callback(payload)
else:
callback(payload)
except Exception:
correlation_id = payload.get("correlation_id", "unknown")
logger.exception(
"Event subscriber failed for %s",
event_type,
extra={"correlation_id": correlation_id},
)
apps/api/src/services/health_monitor.py
+219
View File
@@ -0,0 +1,219 @@
"""Background health monitor that polls container and tunnel health."""
import asyncio
import logging
import uuid
from dataclasses import dataclass
from datetime import datetime, timezone
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from src.database import SessionLocal
from src.models.health_check import HealthCheck
from src.models.tool_instance import ToolInstance
from src.services.correlation import get_correlation_id
from src.services.docker import check_tunnel_health, get_container_status
from src.services.event_bus import InstanceEventBus, InstanceEventPayload
logger = logging.getLogger(__name__)
@dataclass
class HealthSnapshot:
"""In-memory snapshot of an instance's health state."""
container_status: str | None = None
container_healthy: bool | None = None
tunnel_healthy: bool | None = None
exit_code: int | None = None
class HealthMonitor:
"""Polls container and tunnel health, publishing events on state changes."""
POLL_INTERVAL_SECONDS: float = 15.0
_MONITORED_STATUSES: set[str] = {"starting", "running", "unhealthy"}
def __init__(self, event_bus: InstanceEventBus) -> None:
self._event_bus = event_bus
self._task: asyncio.Task | None = None
self._last_known_state: dict[uuid.UUID, HealthSnapshot] = {}
def start(self) -> None:
"""Idempotent start of the background polling task."""
if self._task is not None and not self._task.done():
return
try:
loop = asyncio.get_running_loop()
self._task = loop.create_task(self._poll_loop())
except RuntimeError:
pass
def stop(self) -> None:
"""Cancel the background task and clear state."""
if self._task is not None and not self._task.done():
self._task.cancel()
self._last_known_state.clear()
self._task = None
async def _poll_loop(self) -> None:
"""Main polling loop."""
while True:
try:
await asyncio.sleep(self.POLL_INTERVAL_SECONDS)
await self._run_check_cycle()
except asyncio.CancelledError:
break
except Exception:
logger.exception("Health monitor poll loop error")
async def _run_check_cycle(self) -> None:
"""Check all monitored instances in one cycle."""
async with SessionLocal() as session:
result = await session.execute(
select(ToolInstance).where(
ToolInstance.status.in_(self._MONITORED_STATUSES)
)
)
instances = result.scalars().all()
for instance in instances:
async with SessionLocal() as session:
await self._check_instance(session, instance)
async def _check_instance(
self,
session: AsyncSession,
instance: ToolInstance,
) -> None:
"""Check a single instance and handle state transitions."""
try:
container_info = get_container_status(instance.container_id or "")
except Exception:
logger.exception(
"Health check failed for instance %s",
instance.id,
extra={
"instance_id": str(instance.id),
"correlation_id": get_correlation_id(),
},
)
return
container_status = container_info["status"]
exit_code = container_info["exit_code"]
container_healthy = (
container_info["health"] == "healthy" if container_info["health"] else None
)
tunnel_healthy: bool | None = None
if instance.public_url and container_status == "running":
try:
tunnel_result = check_tunnel_health(instance.public_url)
tunnel_healthy = tunnel_result.get("healthy", False)
except Exception:
logger.exception(
"Tunnel health check failed for instance %s",
instance.id,
extra={
"instance_id": str(instance.id),
"correlation_id": get_correlation_id(),
},
)
tunnel_healthy = False
snapshot = HealthSnapshot(
container_status=container_status,
container_healthy=container_healthy,
tunnel_healthy=tunnel_healthy,
exit_code=exit_code,
)
previous = self._last_known_state.get(instance.id)
# Determine new status
new_status = self._derive_status(snapshot)
# If first check or state changed
if previous is None or not self._snapshots_equal(previous, snapshot):
await self._handle_state_change(
session, instance, previous, snapshot, new_status
)
self._last_known_state[instance.id] = snapshot
def _derive_status(self, snapshot: HealthSnapshot) -> str:
"""Derive instance status from health snapshot."""
if snapshot.container_status != "running":
return "error"
if snapshot.tunnel_healthy is False:
return "unhealthy"
return "running"
def _snapshots_equal(self, a: HealthSnapshot, b: HealthSnapshot) -> bool:
"""Compare two snapshots for equality."""
return (
a.container_status == b.container_status
and a.container_healthy == b.container_healthy
and a.tunnel_healthy == b.tunnel_healthy
and a.exit_code == b.exit_code
)
async def _handle_state_change(
self,
session: AsyncSession,
instance: ToolInstance,
previous: HealthSnapshot | None,
snapshot: HealthSnapshot,
new_status: str,
) -> None:
"""Update DB, insert health check, and publish event."""
previous_status = instance.status
# Update instance status
instance.status = new_status
if new_status == "error":
instance.last_stopped_at = datetime.now(timezone.utc)
# Insert health check row
health_check = HealthCheck(
instance_id=instance.id,
container_status=snapshot.container_status,
container_healthy=snapshot.container_healthy,
tunnel_healthy=snapshot.tunnel_healthy,
exit_code=snapshot.exit_code,
probe_status=None,
probe_output=None,
)
session.add(health_check)
await session.commit()
# Build event payload
correlation_id = get_correlation_id()
metadata: dict = {"previous_status": previous_status}
if snapshot.exit_code is not None:
metadata["exit_code"] = snapshot.exit_code
metadata["error_type"] = "container"
if instance.public_url:
metadata["tunnel_url"] = instance.public_url
if new_status == "error":
event_type = "instance.error"
message = f"Container failed with status {snapshot.container_status}"
if snapshot.exit_code is not None:
message += f" (exit code: {snapshot.exit_code})"
else:
event_type = "instance.health_changed"
message = f"Container is now {new_status}"
payload: InstanceEventPayload = {
"event": event_type,
"instance_id": str(instance.id),
"status": new_status,
"message": message,
"metadata": metadata,
"timestamp": datetime.now(timezone.utc).isoformat(),
"correlation_id": correlation_id,
}
await self._event_bus.publish(event_type, payload)
apps/api/src/services/lifecycle_hooks.py
+98
View File
@@ -0,0 +1,98 @@
"""Lifecycle hook helpers for instrumenting tool instance transitions."""
import uuid
from datetime import datetime, timezone
from sqlalchemy.ext.asyncio import AsyncSession
from src.models.instance_event import InstanceEvent
from src.models.tool_instance import ToolInstance
from src.services.correlation import get_correlation_id
from src.services.event_bus import InstanceEventBus, InstanceEventPayload
def _build_payload(
event_type: str,
instance: ToolInstance,
status: str | None = None,
message: str | None = None,
metadata: dict | None = None,
) -> InstanceEventPayload:
"""Construct a standard event payload."""
return {
"event": event_type,
"instance_id": str(instance.id),
"status": status or instance.status,
"message": message,
"metadata": metadata or {},
"timestamp": datetime.now(timezone.utc).isoformat(),
"correlation_id": get_correlation_id(),
}
async def _write_audit_row(
session: AsyncSession,
instance: ToolInstance,
event_type: str,
created_by: uuid.UUID | None = None,
status: str | None = None,
message: str | None = None,
metadata: dict | None = None,
) -> InstanceEvent:
"""Persist an instance_events audit row."""
row = InstanceEvent(
instance_id=instance.id,
event_type=event_type.replace("instance.", ""),
status=status or instance.status,
message=message,
created_by=created_by,
event_metadata=metadata or {},
)
session.add(row)
await session.commit()
return row
async def publish_lifecycle_event(
event_bus: InstanceEventBus,
session: AsyncSession,
instance: ToolInstance,
event_type: str,
created_by: uuid.UUID | None = None,
status: str | None = None,
message: str | None = None,
metadata: dict | None = None,
) -> None:
"""Publish a lifecycle event and write an audit row after DB commit.
Args:
event_bus: The global event bus.
session: Active async DB session.
instance: The affected tool instance.
event_type: One of instance.created, instance.started, etc.
created_by: User ID for user-initiated actions; None for system.
status: Optional status override.
message: Optional human-readable message.
metadata: Optional extra metadata.
"""
payload = _build_payload(
event_type=event_type,
instance=instance,
status=status,
message=message,
metadata=metadata,
)
# Write audit row
await _write_audit_row(
session=session,
instance=instance,
event_type=event_type,
created_by=created_by,
status=status or instance.status,
message=message,
metadata=metadata,
)
# Publish to bus
await event_bus.publish(event_type, payload)