Runtime

Runtime State Management

The runtime uses a 7-state machine to prevent duplicate operations and coordinate state transitions.

flowchart LR
    Start(("start")) --> Idle["idle"]
    Idle -->|wake| Wake["wake_detected"]
    Wake -->|listen| Listening["listening"]
    Listening -->|transcribe| Transcribing["transcribing"]
    Transcribing -->|think| Thinking["thinking"]
    Thinking -->|speak| Speaking["speaking"]
    Speaking -->|cooldown| Cooldown["cooldown"]
    Cooldown -->|reset| Idle
    Error["error"] -->|recover| Idle

State Guard Methods

Guard	Returns
`should_ignore_wakeword()`	True if mode is not `idle` or current time is before `ignore_wakeword_until`.
`can_start_thinking()`	True if mode is `idle` or `transcribing`.
`snapshot()`	Full runtime state dictionary with mode, flags, timestamps, text, and counters.
`mark_heartbeat()`	Increments `heartbeat_count` and updates `last_heartbeat_at`.
`heartbeat_alive()`	True if heartbeat updated within the last 10 seconds.
`uptime_seconds()`	Seconds since runtime started.

State Side Effects

Mode	Side effects
`idle`	Clears TTS, LLM, capture flags, current transcript, current response, and previous error.
`listening`	Sets `capture_active = True`.
`transcribing`	Sets `capture_active = False`.
`thinking`	Sets `llm_active = True` and stores `current_transcript`.
`speaking`	Sets `llm_active = False`, `tts_active = True`, and stores `current_response`.
`cooldown`	Sets `tts_active = False` and `ignore_wakeword_until = now + cooldown_seconds`.
`error`	Stores `last_error`.

Async Event System

The AsyncEventBus is the core dispatch mechanism for active audio, STT, cognition, and TTS flow. It uses a priority queue so critical events are processed first while preserving FIFO order within the same priority level.

Level	Constant	Category
0	`PRIORITY_CRITICAL`	Critical/shutdown events.
1	`PRIORITY_AUDIO`	Wake word, capture, STT.
2	`PRIORITY_CONVERSATION`	Conversation-level events.
3	`PRIORITY_COGNITION`	LLM/response generation.
5	`PRIORITY_BACKGROUND`	Default/background tasks.

Queue Behavior

Parameter	Value	Meaning
Max queue size	100	Queue full triggers `QueueFull`, logs warning, and increments `events_dropped`.
Listener timeout	30 seconds	Enforced per listener; triggers fallback TTS for transcript/audio/response events.
Concurrent listeners	All	Listeners for the same event run in parallel with `asyncio.gather()`.

Key Event Metrics

{
    "events_received": int,
    "events_emitted": int,
    "events_dispatched": int,
    "events_completed": int,
    "events_failed": int,
    "events_partial_failure": int,
    "events_unhandled": int,
    "events_dropped": int,
    "events_no_listeners": int,
    "listener_timeouts": int,
    "listener_errors": int,
    "current_queue_size": int,
    "queue_peak": int,
    "avg_queue_wait_time": float,
    "avg_event_processing_time": float,
    "avg_listener_processing_time": float
}

No-listener events increment events_no_listeners and events_unhandled, and they are also counted as events_completed. Queue metrics currently use current_queue_size and queue_peak.

Event Persistence

Cosmo persists certain events to SQLite through EventRepository for a durable audit trail. Not all events are persisted; the current implementation persists specific event types to the events table.

CREATE TABLE events (
    id INTEGER PRIMARY KEY,
    type TEXT,
    payload TEXT,
    created_at TIMESTAMP
);

from cosmo.data.database.repositories.event_repository import event_repository

event_repository.emit_event(
    type="user_spoke",
    payload={
        "text": "hello cosmo",
        "confidence": 0.95,
        "duration_ms": 1200
    }
)

Diagnostics System

DiagnosticsManager provides full and compact runtime snapshots. The compact snapshot powers the local status command and the WebUI/SSE stream.

{
    "timestamp": str,
    "system": dict,
    "mode": str,
    "previous_mode": str,
    "tts_active": bool,
    "llm_active": bool,
    "capture_active": bool,
    "conversation_size": int,
    "queue_size": int,
    "events_received": int,
    "events_completed": int,
    "events_no_listeners": int,
    "events_failed": int,
    "listener_timeouts": int,
    "listener_errors": int,
    "last_error": str | None,
    "uptime_seconds": int,
    "uptime_human": str,
    "heartbeat_count": int,
    "last_heartbeat_at": str | None,
    "heartbeat_alive": bool,
    "database": dict,
    "vision": dict
}

Vision Snapshot

The compact diagnostics payload includes a normalized vision snapshot even when the camera is unavailable or the vision manager raises an error.

{
    "enabled": bool,
    "camera_active": bool,
    "camera_available": bool,
    "camera_index": int | None,
    "width": int | None,
    "height": int | None,
    "grayscale": bool | None,
    "started_at": str | None,
    "last_error": str | None,
    "last_brightness": float | None,
    "image_quality": str,
    "image_metrics": {
        "brightness_mean": float,
        "brightness_std": float,
        "dark_ratio": float,
        "bright_ratio": float,
        "overexposed_ratio": float,
        "blur_score": float,
        "backlit_score": float,
        "image_quality": str,
        "face_ready": bool
    },
    "face_ready": bool,
    "face_detection": dict,
    "last_frame_at": str | None,
    "last_snapshot_path": str | None,
    "has_frame": bool,
    "auto_capture": bool,
    "capture_interval": int | float | None
}

Face Detection Snapshot

{
    "enabled": bool,
    "detection_ready": bool,
    "skipped": bool,
    "skip_reason": str | None,
    "face_detected": bool,
    "face_count": int,
    "raw_face_count": int,
    "filtered_out_count": int,
    "faces": list,
    "raw_faces": list,
    "rejected_faces": list,
    "largest_face": dict | None,
    "last_error": str | None
}

A heartbeat background task increments heartbeat_count and is used to monitor runtime liveness.