headquarter/apps/api/src/services/instance/.pi-map.md

apps/api/src/services/instance

dir: apps/api/src/services/instance

index: apps/api/src/services/instance/.pi-map.index.md

role

Provides infrastructure for managing tool instance lifecycles, health monitoring, and event-driven communication within the API service.

files

• __init__.py | Exports the public API for the instance lifecycle services module. | dep: src.services.instance.event_bus, src.services.instance.health_monitor, src.services.instance.lifecycle_hooks
• event_bus.py | Implements a singleton in-memory typed event bus with publish/subscribe pattern for instance lifecycle and health events, supporting both sync and async callbacks with exception isolation. | exp: class:InstanceEventBus, method:init(self) → None, method:new(cls) → "InstanceEventBus", call:super().new, method:_reset_for_testing(self) → None, call:self._subscribers.clear, method:subscribe(self, event_type: str, callback: EventCallback) → Callable[[], None], call:str, call:uuid.uuid4, call:self._subscribers[event_type].append, call:self.unsubscribe, method:unsubscribe(self, event_type: str, callback_id: str) → None, method:unsubscribe_all(self, event_type: str) → None, call:self._subscribers.pop, method:publish(self, event_type: str, payload: InstanceEventPayload) → None, call:callbacks.extend, call:self._subscribers.get, call:inspect.iscoroutinefunction, call:callback, call:payload.get, call:logger.exception | dep: asyncio, inspect, logging, uuid, collections.abc, typing
• health_monitor.py | Background health monitor that polls Docker container and tunnel health for tool instances, updating database state and publishing events on status changes. | exp: class:HealthSnapshot, class:HealthMonitor, method:init(self, event_bus: InstanceEventBus) → None, method:start(self) → None, call:self._task.done, call:asyncio.get_running_loop, call:loop.create_task, call:self._poll_loop, method:stop(self) → None, call:self._task.done, call:self._task.cancel, call:self._last_known_state.clear, method:_poll_loop(self) → None, call:asyncio.sleep, call:self._run_check_cycle, call:logger.exception, method:run_check_cycle(self) → None, call:SessionLocal, call:session.execute, call:select(ToolInstance).where, call:ToolInstance.status.in, call:result.scalars().all, call:self._check_instance, method:_check_instance(self, session: AsyncSession, instance: ToolInstance) → None, call:logger.debug, call:get_container_status, call:logger.exception, call:str, call:get_correlation_id, call:check_tunnel_health, call:tunnel_result.get, call:HealthSnapshot, call:self._last_known_state.get, call:self._derive_status, call:self._snapshots_equal, call:self._handle_state_change, method:_derive_status(self, snapshot: HealthSnapshot, previous: HealthSnapshot | None, current_status: str | None) → str, method:_snapshots_equal(self, a: HealthSnapshot, b: HealthSnapshot) → bool, method:_handle_state_change(self, session: AsyncSession, instance: ToolInstance, previous: HealthSnapshot | None, snapshot: HealthSnapshot, new_status: str) → None, call:HealthCheck, call:session.add, call:session.commit, call:get_correlation_id, call:str, call:datetime.now(timezone.utc).isoformat, call:self._event_bus.publish, call:notification_service.create_notification, call:logger.exception | dep: asyncio, logging, uuid, dataclasses, datetime, sqlalchemy, sqlalchemy.ext.asyncio, src.database, src.models, src.services.shared.correlation, src.services.docker, src.services.shared.tunnel, src.services.instance.event_bus, src.services.shared.notification_service
• lifecycle_hooks.py | Publishes tool instance lifecycle events to an event bus, writes audit rows to the database, and conditionally sends notifications for errors and ready-state transitions. | exp: func:derive_title(event_type: str) → str, call:mapping.get, call:event_type.replace("instance.", "").replace("", " ").title, func:_should_notify(event_type: str, status: str | None) → bool, func:_build_payload(event_type: str, instance: ToolInstance, status, message, metadata) → InstanceEventPayload, call:str, call:datetime.now(timezone.utc).isoformat, call:get_correlation_id, func:_write_audit_row(session: AsyncSession, instance: ToolInstance, event_type: str, created_by, status, message, metadata) → InstanceEvent, call:InstanceEvent, call:event_type.replace, call:session.add, call:session.commit, func:publish_lifecycle_event(event_bus: InstanceEventBus, session: AsyncSession, instance: ToolInstance, event_type: str, created_by, status, message, metadata) → None, call:_build_payload, call:_write_audit_row, call:event_bus.publish, call:_should_notify, call:_derive_title, call:notification_service.create_notification, call:logger.exception, call:payload.get | dep: logging, uuid, datetime, sqlalchemy.ext.asyncio, src.models, src.services.shared.correlation, src.services.instance.event_bus, src.services.shared.notification_service

arch

Event-driven architecture using a singleton in-memory pub/sub event bus with typed events, background polling workers, and lifecycle hooks that coordinate database updates, audit logging, and notifications through exception-isolated sync/async callbacks.

tags

call:self., instance, src, services, event, health, call:logger.exception, check

symbols

• InstanceEventBus
• HealthSnapshot
• HealthMonitor
• init
• new
• _reset_for_testing
• subscribe
• unsubscribe

workflows

• change instance behavior read: __init__.py, event_bus.py, health_monitor.py

dirty