feat: container monitoring backend core (PR-1)

- Add instance_events and health_checks tables with Alembic migration
- InstanceEventBus: typed pub/sub singleton with wildcard support
- HealthMonitor: async background loop polling containers every 15s
- SSE endpoint GET /events/stream with auth and connection limits
- Lifecycle hooks in tool_instances.py (create/start/stop/restart/delete)
- Structured JSON logging with correlation IDs
- 15 new unit tests (EventBus, HealthMonitor, MonitoringModels)

Quality gates: pytest 15 new passed, ruff clean

apps/api/src/api/__init__.py

@@ -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

@@ -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

@@ -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),
@@ -1501,6 +1513,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(
@@ -1518,6 +1539,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,
@@ -1607,6 +1641,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,
@@ -1623,6 +1667,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
@@ -1631,6 +1685,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",
@@ -1756,6 +1819,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}
 
@@ -1892,6 +1964,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"
@@ -1978,6 +2059,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()
 

apps/api/src/logging_config.py

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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)