Agent + Thread Map (pyPiBot)

This document maps the active agents/threads in the current runtime and the data flows between them. It mirrors the wiring in main.py and the thread loops in the controllers/monitors. The realtime agent thread is the primary agentic flow, coordinating audio in/out, tool execution, and event injections.

Architecture context: This map focuses on runtime/lifecycle orchestration. For layer ownership and roadmap context (runtime vs perception/memory vs arbitration/governance), see docs/architecture/theo_cognitive_stack.md (see the Fast Layer Triage Matrix for quick placement checks), docs/architecture/decision_arbitration_authority_map.md for implemented arbitration authority seams, and docs/architecture/quiet_intent.md for quiet-intent ownership/authority boundaries, and docs/architecture/continuity_presence.md for continuity/presence ownership and boundary constraints. and the architecture index at docs/architecture/README.md.

Top-Level Runtime Orchestration

This section describes runtime plumbing, not higher-order decision policy. Cross-check proposed changes against the cognitive stack roadmap before moving logic between lifecycle and upper cognition layers.

The entrypoint in main.py builds and launches the following components:

Realtime API agent (async loop): The primary agent thread that connects to OpenAI realtime, manages the websocket event loop, and coordinates tool execution, audio input, audio output, and event injection. Tool calls are routed through the governance layer, which builds structured action packets, enforces tiered approvals/autonomy windows, and blocks tool use on stop-word emergency phrases. It exposes is_ready_for_injections for other threads to gate event delivery. It also tracks orchestration phases (sense/plan/act/reflect/idle) during the response lifecycle and emits phase-transition logs for observability. On interaction-state transitions, it also refreshes Quiet Intent (consultative posture-bias snapshot) and emits deduped quiet-intent diagnostics; this does not alter arbitration, governance, or execution authority seams.
Governance layer: Builds action packets for tool calls, applies tool tier policy (read-only vs. reversible vs. stateful/credentialed), enforces autonomy windows, and decides whether approval is required before execution.
Event bus (shared queue): Thread-safe queue that collects sensor events and orders them by priority for injection into the realtime session.
Motion control loop thread: A background loop that drives the servo controller and executes queued gesture/motion actions.
Vision loop thread: The camera controller captures frames, detects scene changes, and sends images into the realtime agent when ready.
IMU monitor thread: Samples IMU data, derives motion events, and emits event callbacks into the main runtime.
Battery monitor thread: Samples the ADS1015 voltage, derives battery events, and emits callbacks to the runtime.
Ops orchestrator thread: Runs a heartbeat loop that executes health probes, debounces health states, enforces rolling budgets, and emits health snapshots/alerts onto the event bus.
Event injector thread: Drains the shared event bus, applies cooldown/TTL checks, and injects events into the realtime session when the websocket is ready. The injector is instantiated during RealtimeAPI construction and is started when RealtimeAPI.run() begins.

main.py initializes RealtimeAPI and starts the sensor/service background threads before entering the async realtime loop, while the EventInjector thread itself is started by RealtimeAPI.run() at loop startup.

System Map (Visual)

Runtime/Lifecycle Ownership

flowchart TD
    MAIN[main.py] --> CFG[ConfigController]
    MAIN --> STOR[StorageController]
    MAIN --> MM[MemoryManager]
    MM --> EW[MemoryEmbeddingWorker.start]

    MAIN --> RT["RealtimeAPI (required)"]
    RT --> EB[EventBus]
    RT --> EI["EventInjector.start in run()"]
    RT --> WS[WebSocket loops]
    RT --> TOOLS["Tool dispatch + Governance"]

    MAIN --> MOT[MotionController.start_control_loop]
    MAIN --> CAM[CameraController.start_vision_loop]
    MAIN --> IMU[ImuMonitor.start_loop]
    MAIN --> BAT[BatteryMonitor.start_loop]
    MAIN --> OPS[OpsOrchestrator.start_loop]
    MAIN --> SYSCTX[SystemContextCoordinator.start]

    IMU --> EB
    BAT --> EB
    OPS --> EB
    EI --> RT

    MAIN --> SHUT[Shutdown sequence]
    SHUT --> SYSCTX_STOP[system_context_coordinator.stop]
    SHUT --> EW_STOP[embedding_worker.stop]
    SHUT --> OPS_STOP[ops_orchestrator.stop_loop]
    SHUT --> CAM_STOP[camera.stop_vision_loop]
    SHUT --> MOT_STOP[motion.stop_control_loop]
    SHUT --> IMU_STOP["imu.stop_loop + unregister"]
    SHUT --> BAT_STOP["battery.stop_loop + unregister"]

Layered Dependency Map

flowchart LR
    subgraph AI
      RT[ai/realtime_api.py]
      GOV[ai/governance.py]
      TOOLS[ai/tools.py]
      EBUS[ai/event_bus.py]
    end

    subgraph Services
      OPS[services/ops_orchestrator.py]
      IMU[services/imu_monitor.py]
      BAT[services/battery_monitor.py]
      MM[services/memory_manager.py]
      RS[services/research/*]
    end

    subgraph Storage
      SC[storage/controller.py]
      FAC[storage/factories.py]
      RB[storage/research_budget.py]
      MEM[storage/memories.py]
      UP[storage/user_profiles.py]
    end

    subgraph HardwareMotion
      HW[hardware/*]
      MOT[motion/*]
      INT[interaction/*]
    end

    RT --> GOV
    RT --> TOOLS
    RT --> INT
    RT --> MOT
    RT --> SC
    RT --> RS

    TOOLS --> Services
    TOOLS --> SC

    MM --> FAC
    FAC --> MEM
    FAC --> UP
    FAC --> RB
    RB --> SC

    OPS --> MM
    OPS --> MOT
    OPS --> Services

    IMU --> EBUS
    BAT --> EBUS
    OPS --> EBUS

    IMU --> HW
    BAT --> HW
    MOT --> HW

    %% Cross-boundary edges
    Services -.imports ai event bus.-> EBUS
    MOT -.imports storage controller.-> SC

Detailed Runtime Flow

flowchart LR
    subgraph Runtime["main.py runtime"]
        MAIN[main.py orchestration]
    end

    subgraph Realtime["Realtime agent flow (async)"]
        WS["RealtimeAPI.run() websocket loop"]
        MIC["AsyncMicrophone<br/>PyAudio callback + queue"]
        PLAYER["AudioPlayer<br/>playback thread"]
        GOV["GovernanceLayer<br/>action packets + approvals"]
        TOOLS["Tool execution<br/>function_map"]
        BUS["EventBus"]
        INJECT["EventInjector thread"]
        STATE[InteractionStateManager]
        ORCH["OrchestrationState<br/>sense/plan/act/reflect"]
        REFLECTOR["ReflectionCoordinator<br/>async task"]
    end

    subgraph Sensors["Sensor/monitor threads"]
        IMU[IMU monitor thread]
        BAT[Battery monitor thread]
        OPS[Ops orchestrator thread]
    end

    subgraph Vision["Vision thread"]
        CAM[CameraController vision loop]
    end

    subgraph Motion["Motion loop"]
        MOTION[MotionController control loop]
    end

    MAIN --> WS
    MAIN --> CAM
    MAIN --> MOTION
    MAIN --> IMU
    MAIN --> BAT
    MAIN --> OPS

    MIC --> WS
    WS --> PLAYER
    WS --> GOV
    GOV --> TOOLS
    WS --> BUS
    BUS --> INJECT
    INJECT --> WS
    WS --> STATE
    WS --> ORCH
    ORCH --> REFLECTOR

    IMU --> BUS
    BAT --> BUS
    OPS --> BUS

    CAM --> WS
    WS --> CAM

    STATE --> MOTION

Flow Details

1. Realtime Agent Thread (Audio In/Out + Tools)

Audio in is captured by AsyncMicrophone via PyAudio’s callback, which buffers frames into a queue; the realtime loop drains that queue and sends audio frames to the websocket as input_audio_buffer.append events.

Audio out is driven by websocket events (response.output_audio.delta), which are accumulated and streamed to AudioPlayer for playback. When playback finishes, the microphone is resumed and the agent can accept new audio input.

Tool execution is driven by realtime function call events (response.function_call_arguments.done). The realtime agent builds a structured action packet (what/why/impact/rollback/cost/confidence/alternatives), asks for approval when required, and enforces autonomy windows before mapping the call through function_map and returning results as function_call_output items. Stop words immediately cancel pending actions and pause tool execution for a cooldown interval.

2. Vision Thread → Realtime Agent

The camera controller’s vision loop captures low-res luma frames, detects motion or scene changes, and when a change is detected captures a full-resolution image. It then queues the image to the realtime agent for injection, using send_image_to_assistant() when the realtime loop is ready; otherwise, it buffers images until the agent is ready.

3. IMU + Battery Monitor Threads → Realtime Agent

The IMU and battery monitors run independent loops and publish structured events to the shared EventBus. main.py registers handlers that convert sensor events into bus payloads, and the realtime agent’s EventInjector thread drains the bus when ready, applying cooldown/TTL logic before injecting messages into the websocket session.

4. Ops Orchestrator → Event Bus

The ops orchestrator runs a periodic tick loop to execute health probes, calculate debounced health status, emit health snapshots, and publish alerts or budget warnings to the shared event bus. This provides a low-frequency operations heartbeat alongside the sensor loops and is started/stopped by main.py just like the other background services.

5. Motion Loop and Realtime State Hooks

Motion control is a continuous loop running in its own thread. The realtime agent uses an InteractionStateManager to interpret listening/speaking states and can emit gesture actions (e.g., nods) by pushing actions into the motion controller queue, assuming the control loop is running.

6. Orchestration + Reflection Lifecycle

The realtime loop now tracks orchestration phases with an OrchestrationState helper. Phases transition as the websocket events progress: speech start and injected text push the state into sense, response.created moves to plan, function calls enter act, and response completion marks reflect before returning to idle. Transitions are logged for runtime diagnostics.

Reflection generation is handled by ReflectionCoordinator, which is configured by reflection_enabled and reflection_min_interval_s and runs an async task so the main loop stays responsive. Each reflection captures the last user input, assistant reply, tool calls, and response metadata, then stores a JSON payload into StorageController (with min-interval and in-flight task guards). The latest lessons are also pulled from ReflectionManager during session initialization to seed instructions for the next turn.

Quick Reference: Threads + Responsibilities

Thread/Agent	Owner	Purpose	Key Inputs	Key Outputs
Realtime agent (async loop)	`ai/RealtimeAPI`	Websocket session, audio IO, tool execution	Mic audio, vision/battery/IMU messages	Audio playback, tool calls, state changes
Audio input thread	`AsyncMicrophone`	Capture mic audio into buffer	Mic hardware	Audio frames to realtime
Audio output thread	`AudioPlayer`	Playback assistant audio	Realtime output audio	Speaker output, playback completion callback
Vision thread	`CameraController`	Detect scene changes and push images	Camera frames	Image injections to realtime
Motion control loop	`MotionController`	Servo motion execution	Gesture/action queue	Servo position updates
IMU monitor	`ImuMonitor`	Sample IMU data, emit events	IMU sensor	Motion events to main/realtime
Battery monitor	`BatteryMonitor`	Sample voltage, emit events	ADS1015 sensor	Battery events to main/realtime
Ops orchestrator	`OpsOrchestrator`	Health probes, heartbeats, alerts, budgets	Realtime API + system probes	Health snapshots + alert events
Event bus	`ai/EventBus`	Thread-safe queue of pending realtime events	Sensor handler events	Prioritized events for injection
Event injector thread	`ai/EventInjector`	Flush queued bus events when realtime ready	EventBus entries	Messages to realtime

FilesExpand file tree

agent-map.md

Latest commit

History