Monorepo for live camera perception (Ultralytics YOLO + MiDaS monocular depth) exposed over a FastAPI WebSocket, plus a Next.js client that draws detections and optionally runs a Gemini-based “household tidying” agent that proposes human-executable actions and gated text-to-speech.

This document is a technical reference for schemas, in-memory state, concurrency, environment variables, and how each module fits together.

flowchart LR
  subgraph browser [Browser / Next.js]
    Cam[Camera + canvas capture]
    WS[WebSocket client]
    HTTP[fetch POST agent/step]
    TTS[Web Speech or OpenAI TTS]
  end
  subgraph backend [Python backend]
    Dec[decode_image_bytes]
    VP[VisionPipeline YOLO + MiDaS]
    PP[postprocess_frame_for_agent]
    WSrv[/ws/infer]
    Pub[publish_latest LATEST_STATE]
    Agent[run_agent Gemini]
    Gate[compute_voice_gate]
    Step[/v1/agent/step]
  end
  Cam -->|binary WebP/JPEG + fmt byte| WS
  WS --> WSrv
  WSrv --> Dec --> VP --> PP
  PP --> Pub
  Step --> Agent
  Pub --> Step
  Step --> Gate
  HTTP --> Step
  Gate --> HTTP
  HTTP --> TTS

Separation of concerns

• Vision path is tuned for low latency: decode → YOLO + depth → JSON to the browser; in parallel, CPU postprocess builds WebP + grounded JSON and atomically publishes LATEST_STATE for the session.
• Agent path is slow (Gemini round-trip). It never blocks the WebSocket loop beyond copying a snapshot; rate limiting and “busy” flags prevent overlapping calls per session.
• TTS is a client concern; the server only returns a voice object and accepts is_tts_playing so the voice gate can defer non-urgent speech while audio plays.

• Path: /ws/infer
• Query parameters (read in inference_server.py):
  • model: oiv7 (default) or coco → selects YOLO weights preset.
  • session_id: opaque string; default server-side session is "default" if missing or blank. Must match POST /v1/agent/step?session_id= so LATEST_STATE lines up.

The server accepts raw WebP/JPEG bytes or an optional leading format byte:

The Next.js client sends [fmt_u8][image_bytes] (CameraOverlay.tsx: FMT_JPEG = 1, FMT_WEBP = 2).

Decode implementation: decode_image_bytes in inference_server.py (TurboJPEG for JPEG when available; Pillow for WebP).

• A process-wide asyncio.Lock() (_infer_lock) ensures one GPU pipeline infer() at a time across all WebSocket clients (avoids GPU contention).
• Decode and CPU postprocess run in the VISION thread pool.

1. Receive bytes → decode to BGR numpy.
2. Under _infer_lock: VisionPipeline.infer(frame) → wire payload out (see §4).
3. Without holding _infer_lock: postprocess_frame_for_agent(frame, out) → (webp_bytes, grounded).
4. await publish_latest(session_id, webp_bytes, out, grounded) — updates LATEST_STATE before replying on the socket (avoids client races where agent POST returns 409).
5. await ws.send_json(out) — same structure as §4.

Produced by _detections_payload in vision_pipeline.py. Sent to the browser as the WebSocket text JSON message.

// Conceptual TypeScript shape (matches frontend InferResponse)
type InferResponse = {
  w: number;
  h: number;
  detections: Detection[];
  error?: string; // only on decode/pipeline failure paths
};

type Detection = {
  label: string;      // class name from YOLO
  conf: number;       // 0..1
  x1: number; y1: number; x2: number; y2: number;  // normalized 0..1
  cx: number; cy: number;                          // box center, normalized
  rel_depth: number;  // MiDaS scalar at integer pixel (u,v) of center
};

Depth semantics: MiDaS outputs are used as a relative ordering signal (see YOLO webcam docstring: larger values ≈ closer in this setup). They are not metric depth. The agent prompt describes cz as a rough step-equivalent after scaling (§5).

On failure, the server may send { error, w: 0, h: 0, detections: [] }.

Built by ground_detections_payload in grounding.py from the same out dict the client receives (not from WebP).

Each element:

{
  "class": "cup",
  "cx": 0.512,
  "cy": 0.403,
  "cz": 1.234,
  "sector": "left" | "center" | "right",
  "v_band": "upper" | "middle" | "lower",
  "conf": 0.87
}

• cz = rel_depth / K where K is AGENT_DEPTH_SCALE_K (default 1.0). There is no per-frame min/max normalization — calibration is your responsibility.
• sector / v_band use configurable thresholds: AGENT_SECTOR_* env vars (defaults split at ~1/3 and ~2/3 of normalized x/y).
• Filtering: drop detections with conf < AGENT_DETECTION_MIN_CONF (default 0.3), sort by conf descending, keep up to AGENT_TOP_N_DETECTIONS (default 20).
• Optional allowlist: AGENT_CLASS_ALLOWLIST — comma-separated lowercase names; if set, only those classes pass.

prioritize_grounded_for_model in agent_attention.py:

• If task_anchor is empty → returns a copy of full grounded list (no reorder).
• Else:
  1. person, cat, dog always appear first (by conf), regardless of anchor.
  2. Up to AGENT_ANCHOR_TOP_MATCH detections whose class matches anchor tokens (substring/token overlap on class string).
  3. Up to AGENT_ANCHOR_TOP_REST other detections (by conf), with a twist: if inferred_held_object is set, prefer non-held classes first in the “rest” slots so surfaces (table, bin, …) surface before the held blob.

Deduping preserves order on (class, cx, cy, cz).

All session data lives in _sessions: dict[str, _SessionBrainState] under a single asyncio.Lock() (_lock). There is no disk persistence of session brain state (except Gemini logs).

apply_inferred_held_after_actions walks this tick’s actions in order:

• place or drop → clears held.
• pick_up → if args.target is a non-empty string, sets held to that string (truncated to AGENT_HELD_OBJECT_MAX_LEN).

This is not vision-based truth; the prompt tells the model to verify visually.

merge_task_anchor_from_model (world_state.py):

• Model output "" → keep current anchor.
• CLEAR (case-insensitive; outer quotes stripped) → clear anchor.
• Otherwise → sanitize (alphanumerics + ASCII spaces only, collapse whitespace); if empty after sanitize, keep current; else replace, truncated to AGENT_TASK_ANCHOR_MAX_LEN.

get_memory_for_prompt builds:

• turn_log_excerpt: last AGENT_MEMORY_TURNS turns, each as one JSON object per line (thought, instruction, actions, anchor_after), total chars capped by AGENT_MEMORY_MAX_CHARS (tail-truncated if needed).
• seconds_since_last_step: time.monotonic() - last_step_monotonic if a prior step exists.

try_begin_agent_work:

• Rejects if agent_busy → HTTP 429 reason busy.
• Rejects if now_ms - last_agent_ms < AGENT_MIN_INTERVAL_MS → 429 reason throttle.
• Else sets agent_busy = True.

finish_agent_work: clears busy; on success updates last_agent_ms.

POST /v1/agent/step (inference_server.py):

1. copy_snapshot(session_id) — if None → 409 (“No perception state… send frames first”).
2. try_begin_agent_work — may 429.
3. Missing GEMINI_API_KEY → 503, busy cleared.
4. get_memory_for_prompt, prioritize_grounded_for_model on snap.grounded.
5. run_agent(snap.frame_webp, grounded_for_model, …) in AGENT thread pool (executors.py).
6. On success: update_memory_after_agent_success, compute_voice_gate, commit_voice_gate_result, finish_agent_work(success=True).

Query params: goal, last_outcome (optional strings, forwarded into the user message).

Body (JSON, optional): { "is_tts_playing": boolean } — forwarded to compute_voice_gate.

The system instruction is assembled in _system_instruction() in agent_gemini.py: mission, role, anchor/holding rules, ACTION_REGISTRY_PROMPT, and JSON-only output requirement.

Includes, in order (when applicable):

• Grounded objects (JSON array): — JSON dump of attention-filtered grounded list.
• Stored task_anchor
• Session tool-inference — inferred held object line.
• Recent turn log — excerpt.
• Recent action labels + Last step action labels
• Motion cooldown hint — if within AGENT_MOTION_COOLDOWN_HINT_SEC of last step and last labels included a motion action (move_*, turn_*), nudge toward wait / not repeating motion without evidence.
• Current goal, Last outcome / human feedback
• Closing instruction to respond with JSON.

Image side: types.Part.from_bytes(..., mime_type="image/webp").

Defined in agent_tools.py (Pydantic, used as response_schema):

Legacy: if the model returns say instead of thought, a validator maps say → thought.

ActionName: move_forward, move_backward, turn_left, turn_right, pick_up, drop, place, look_around, wait.

args_json examples (from prompt/registry):

• move_forward / move_backward: {"steps": 1}
• turn_left / turn_right: {"degrees": 30}
• pick_up: {"target": "<class string>"} — must match grounded class
• place: {"target": "...", "near": "<class string>"} — near must be grounded class
• drop: {}
• look_around: {}
• wait: {} or {"seconds": 2}

Success JSON (inference_server.py):

HTTP errors:

• 409 — no LATEST_STATE for session.
• 429 — busy or throttle.
• 503 — no API key.
• 502 — Gemini failure (Gemini error: …).

append_gemini_log(session_id, entry) writes JSON Lines (.jsonl). Filename pattern: gemini_{sanitized_session}_{utc_timestamp}.jsonl.

Each line may include: UTC timestamp, model id, full user_text, frame_webp_bytes length, grounded snapshot, memory fields, gemini_response metadata, parsed agent_response, or error.

compute_voice_gate decides should_speak, supersede, and phase using anchor transitions, motor fingerprint, dwell, min interval, and is_tts_playing.

motor_fingerprint_for_actions: for every action except wait, canonicalize args_json (sorted keys JSON) and build sorted strings name:{...} joined by |. Empty if only waits.

1. Anchor changed (anchor_before != anchor_after) → should_speak, supersede = True, commit reason anchor.
2. Else if fingerprint non-empty and ≠ voice_last_motor_fingerprint → should_speak, supersede = False, commit motor.
3. Else if dwell: same anchor, voice_last_speak_monotonic > 0, and elapsed ≥ VOICE_MIN_DWELL_SEC → should_speak (unless normalized TTS line equals last spoken line — then suppressed).
4. is_tts_playing and should_speak and not supersede → force should_speak = False (urgent barge-in still allowed via supersede).

VOICE_MIN_INTERVAL_SEC (default 2): if should_speak, not supersede, and last speak was too recent → suppress.

The speak string in the payload is the line intended for TTS when gating allows; when not speaking, the gate may echo voice_last_speak_text for UI stability.

commit_voice_gate_result (in world_state.py) runs only when should_speak; updates voice_last_speak_text, voice_last_speak_monotonic, and on commit_reason == "motor" stores voice_last_motor_fingerprint.

CameraOverlayMount.tsx loads CameraOverlay with next/dynamic { ssr: false } so browser-only APIs stay client-only.

On mount, crypto.randomUUID() (fallback sid-{Date.now()}) — same id appended to /ws/infer and /v1/agent/step.

NEXT_PUBLIC_INFERENCE_WS_URL (default ws://127.0.0.1:8000/ws/infer). Query params added at connect: model, session_id.

requestAnimationFrame drives tick() when streaming:

• Skips when document.visibilityState === 'hidden'.
• Ping-pong with busyRef: one frame in flight at a time.
• Canvas toBlob WebP at quality 0.5 if supported, else JPEG 0.72.
• Prepends FMT_WEBP / FMT_JPEG byte.

inferReady: set true after first JSON message with w > 0 (used to avoid agent POST before state exists).

When streaming && sessionId && inferReady:

• HTTP base: NEXT_PUBLIC_AGENT_HTTP_URL or derived from WebSocket URL (ws→http, wss→https).
• POST .../v1/agent/step?session_id= every 750ms (gapMs), with at most one request in flight (agentStepInFlightRef).
• Body: { is_tts_playing } from subscribeTtsPlaying / getIsTtsPlaying() in agentTts.ts.
• 429 is ignored silently (throttle).
• TTS: if voice.should_speak, uses voice.speak with speakInstruction(line, { supersede }); non-supersede path delays slightly (CLIENT_TTS_MIN_GAP_MS = 250) to stagger from server VOICE_MIN_INTERVAL_SEC.

unlockAudioFromUserGesture() and resumeAudioContextAfterUserGesture() run on Start camera (same user gesture) for Safari/iOS: silent Web Speech utterance + AudioContext.resume().

NEXT_PUBLIC_TTS_ENGINE, NEXT_PUBLIC_OPENAI_TTS_VOICE, server OPENAI_API_KEY for /api/tts. Optional NEXT_PUBLIC_TTS_DEBUG.

See backend/.env.example for the canonical list. Additional / noteworthy:

See frontend-remop/.env.example.

You need two processes: Python inference server + Next.js dev server.

cd backend
python -m venv .venv
source .venv/bin/activate   # Windows: .venv\Scripts\activate
pip install -r requirements.txt
cp .env.example .env   # set GEMINI_API_KEY for agent; optional CORS, tuning
python -m uvicorn inference_server:app --host 0.0.0.0 --port 8000

Shortcut: ./serve.sh (uses venv python if present).

Health: http://127.0.0.1:8000/health

Troubleshooting ModuleNotFoundError: No module named 'torch': use python -m uvicorn inside the activated venv, not a global uvicorn binary.

PyTurboJPEG (optional JPEG speedup): requires libjpeg-turbo on the system; otherwise OpenCV decode is used.

cd frontend-remop
cp .env.example .env.local   # optional: NEXT_PUBLIC_INFERENCE_WS_URL, OPENAI_API_KEY
npm install
npm run dev

Open http://localhost:3000. Use Start camera (required for Safari / user-gesture rules).

Use wss:// and HTTPS in production for camera APIs outside localhost; align INFERENCE_CORS_ORIGINS.

cd backend && source .venv/bin/activate && python yolo26_depth_webcam.py

First run downloads YOLO and MiDaS weights; timm is pinned for MiDaS v3.1 compatibility.

This project is licensed under the MIT License—see LICENSE.