# LinkedStorm Implementation Plan > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. **Goal:** Build and deploy LinkedStorm — 3 Docker containers (pipeline, api, cron) that capture household speech via browser WebSocket, transcribe with Whisper, identify speakers with resemblyzer, and serve a searchable daily log dashboard. **Architecture:** Browser tabs stream 250ms audio chunks over WebSocket to `linkedstorm-pipeline` (port 8300), which runs Silero VAD, calls Whisper STT + resemblyzer concurrently, matches voice embeddings against SQLite, and writes utterances. `linkedstorm-api` (port 8310) serves the FastAPI dashboard. `linkedstorm-cron` runs nightly Ollama summaries. All three mount the same NAS volume for SQLite + WAV clips. **Tech Stack:** Python 3.12, FastAPI, uvicorn, SQLite WAL+FTS5, resemblyzer, silero-vad, httpx, livekit-api, pytest, pytest-asyncio, vanilla JS, Docker Compose. **Code repo:** `http://192.168.86.100:3003/jeremy/linkedstorm` (clone to `/mnt/c/Users/Jeremy/projects/linkedstorm`) **Compose repo:** `/mnt/c/Users/Jeremy/projects/docker/compose/hf-pdocker-01/linkedstorm/` --- ## File Map ``` /mnt/c/Users/Jeremy/projects/linkedstorm/ ├── pyproject.toml ├── pytest.ini ├── .env.example ├── app/ │ ├── shared/ │ │ ├── __init__.py │ │ ├── config.py # Pydantic Settings — all env vars │ │ ├── database.py # SQLite thread-local connection, schema init │ │ └── models.py # Dataclasses: Room, Speaker, Utterance, etc. │ ├── pipeline/ │ │ ├── __init__.py │ │ ├── main.py # FastAPI app: ws://:8300/ws/{room_id} + /internal/rooms/{id}/mute │ │ ├── ingestion.py # RoomVADBuffer: per-room VAD accumulator │ │ ├── processor.py # process_utterance: Whisper + resemblyzer + DB write │ │ └── recognition.py # embed_audio, find_best_speaker_match, pack/unpack embedding │ ├── api/ │ │ ├── __init__.py │ │ ├── main.py # FastAPI app: mounts all routes, serves static/ │ │ ├── routes/ │ │ │ ├── __init__.py │ │ │ ├── utterances.py # GET/POST /api/utterances, /tag, /dismiss, /clip │ │ │ ├── speakers.py # GET/POST /api/speakers │ │ │ ├── rooms.py # GET /api/rooms, POST /api/rooms/{id}/mute │ │ │ ├── search.py # GET /api/search?q= │ │ │ └── rag.py # POST /api/ask, GET /api/summary/{date} │ │ └── static/ │ │ ├── index.html # Dashboard SPA (5 tabs) │ │ ├── capture.html # Minimal capture page (MediaRecorder + WS) │ │ └── app.js # Dashboard JS │ └── cron/ │ ├── __init__.py │ ├── main.py # Sleep-loop entrypoint: runs all jobs at midnight │ ├── daily_summary.py # Fetch utterances → Ollama → daily_summaries row │ ├── cluster_unknowns.py # Cluster unknown embeddings into Unknown #N groups │ └── purge_clips.py # Delete WAV files older than CLIP_RETENTION_DAYS ├── tests/ │ ├── conftest.py # Shared fixtures: in-memory SQLite settings │ ├── test_database.py │ ├── test_recognition.py │ ├── test_ingestion.py │ ├── test_processor.py │ ├── test_cron.py │ └── routes/ │ ├── conftest.py # TestClient fixture │ ├── test_utterances.py │ ├── test_speakers.py │ ├── test_rooms.py │ ├── test_search.py │ └── test_rag.py ├── pipeline/ │ └── Dockerfile ├── api/ │ └── Dockerfile └── cron/ └── Dockerfile /mnt/c/Users/Jeremy/projects/docker/ ├── compose/hf-pdocker-01/linkedstorm/ │ ├── docker-compose.yml │ └── .env.example └── stacks/traefik/dynamic/linkedstorm.yml ``` --- ## Task 1: Clone repo and scaffold **Files:** `pyproject.toml`, `pytest.ini`, `.env.example`, all `__init__.py` files - [ ] **Clone the linkedstorm repo locally** ```bash git clone http://192.168.86.100:3003/jeremy/linkedstorm /mnt/c/Users/Jeremy/projects/linkedstorm cd /mnt/c/Users/Jeremy/projects/linkedstorm ``` - [ ] **Create `pyproject.toml`** ```toml [build-system] requires = ["setuptools>=68"] build-backend = "setuptools.backends.legacy:build" [project] name = "linkedstorm" version = "0.1.0" requires-python = ">=3.12" dependencies = [ "fastapi==0.115.0", "uvicorn[standard]==0.30.6", "pydantic-settings==2.4.0", "httpx==0.27.2", "websockets==13.1", "resemblyzer==0.1.1.dev0", "silero-vad==5.1.2", "torch==2.4.1", "numpy==1.26.4", "livekit-api==0.7.2", "scipy==1.14.1", ] [project.optional-dependencies] dev = [ "pytest==8.3.3", "pytest-asyncio==0.24.0", "pytest-mock==3.14.0", "httpx==0.27.2", ] [tool.setuptools.packages.find] where = ["."] include = ["app*"] ``` - [ ] **Create `pytest.ini`** ```ini [pytest] asyncio_mode = auto testpaths = tests ``` - [ ] **Create `.env.example`** ```bash DB_PATH=/data/linkedstorm.db CLIPS_DIR=/data/clips SPEACHES_URL=http://192.168.86.150:8000 OLLAMA_URL=http://192.168.86.150:11434 OLLAMA_MODEL=qwen2.5:14b LIVEKIT_URL=ws://192.168.86.150:7880 LIVEKIT_API_KEY=devkey LIVEKIT_API_SECRET=devsecret PIPELINE_INTERNAL_URL=http://linkedstorm-pipeline:8300 CLIP_RETENTION_DAYS=30 SPEAKER_KNOWN_THRESHOLD=0.85 SPEAKER_AMBIGUOUS_THRESHOLD=0.60 ``` - [ ] **Create all `__init__.py` files** ```bash mkdir -p app/shared app/pipeline app/api/routes app/api/static app/cron mkdir -p tests/routes pipeline api cron touch app/__init__.py app/shared/__init__.py app/pipeline/__init__.py touch app/api/__init__.py app/api/routes/__init__.py touch app/cron/__init__.py touch tests/__init__.py tests/routes/__init__.py ``` - [ ] **Install dev dependencies** ```bash pip install -e ".[dev]" ``` - [ ] **Commit** ```bash git add . git commit -m "feat: scaffold project structure" git push origin master ``` --- ## Task 2: Shared config and models **Files:** `app/shared/config.py`, `app/shared/models.py` - [ ] **Write `app/shared/config.py`** ```python from pydantic_settings import BaseSettings class Settings(BaseSettings): db_path: str = "/data/linkedstorm.db" clips_dir: str = "/data/clips" speaches_url: str = "http://192.168.86.150:8000" ollama_url: str = "http://192.168.86.150:11434" ollama_model: str = "qwen2.5:14b" livekit_url: str = "ws://192.168.86.150:7880" livekit_api_key: str = "devkey" livekit_api_secret: str = "devsecret" pipeline_internal_url: str = "http://linkedstorm-pipeline:8300" clip_retention_days: int = 30 speaker_known_threshold: float = 0.85 speaker_ambiguous_threshold: float = 0.60 vad_silence_ms: int = 600 vad_min_speech_ms: int = 500 model_config = {"env_file": ".env", "env_file_encoding": "utf-8"} def get_settings() -> Settings: return Settings() ``` - [ ] **Write `app/shared/models.py`** ```python from dataclasses import dataclass from datetime import datetime from typing import Optional @dataclass class Room: id: int name: str device_label: Optional[str] is_active: bool last_seen: Optional[datetime] created_at: datetime @dataclass class Speaker: id: int name: str created_at: datetime updated_at: datetime @dataclass class VoiceEmbedding: id: int speaker_id: int embedding: bytes # packed float32 audio_sample: Optional[bytes] created_at: datetime @dataclass class Utterance: id: int room_id: int speaker_id: Optional[int] transcript: str embedding: Optional[bytes] match_status: str # known | unknown | ambiguous | unreviewed match_confidence: Optional[float] start_time: datetime end_time: datetime audio_clip_path: Optional[str] livekit_room: Optional[str] created_at: datetime @dataclass class DailySummary: id: int date: str # YYYY-MM-DD summary_text: str generated_at: datetime ``` - [ ] **Commit** ```bash git add app/shared/ git commit -m "feat: add shared config and models" git push origin master ``` --- ## Task 3: Database layer (TDD) **Files:** `app/shared/database.py`, `tests/conftest.py`, `tests/test_database.py` - [ ] **Write failing tests first — `tests/conftest.py`** ```python import pytest import tempfile import os from app.shared.config import Settings from app.shared.database import init_schema, get_db @pytest.fixture def settings(tmp_path): db_file = str(tmp_path / "test.db") clips_dir = str(tmp_path / "clips") os.makedirs(clips_dir, exist_ok=True) s = Settings( db_path=db_file, clips_dir=clips_dir, speaches_url="http://localhost:9999", ollama_url="http://localhost:9999", livekit_api_key="test", livekit_api_secret="test", ) init_schema(s) return s @pytest.fixture def db(settings): return get_db(settings) ``` - [ ] **Write `tests/test_database.py`** ```python def test_schema_creates_rooms_table(db): rows = db.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='rooms'").fetchall() assert len(rows) == 1 def test_schema_creates_utterances_fts(db): rows = db.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='utterances_fts'").fetchall() assert len(rows) == 1 def test_insert_and_fetch_room(db): db.execute("INSERT INTO rooms (name) VALUES (?)", ("Kitchen",)) db.commit() row = db.execute("SELECT name FROM rooms WHERE name='Kitchen'").fetchone() assert row["name"] == "Kitchen" def test_insert_speaker_and_embedding(db): db.execute("INSERT INTO speakers (name) VALUES (?)", ("Jeremy",)) db.commit() speaker = db.execute("SELECT id FROM speakers WHERE name='Jeremy'").fetchone() db.execute( "INSERT INTO voice_embeddings (speaker_id, embedding) VALUES (?, ?)", (speaker["id"], b"\x00\x01\x02\x03"), ) db.commit() emb = db.execute("SELECT speaker_id FROM voice_embeddings").fetchone() assert emb["speaker_id"] == speaker["id"] def test_fts_trigger_on_utterance_insert(db): db.execute("INSERT INTO rooms (name) VALUES (?)", ("Living Room",)) db.commit() room = db.execute("SELECT id FROM rooms WHERE name='Living Room'").fetchone() db.execute( "INSERT INTO utterances (room_id, transcript, match_status, start_time, end_time) VALUES (?,?,?,?,?)", (room["id"], "hello world test", "unreviewed", "2026-01-01 10:00:00", "2026-01-01 10:00:03"), ) db.commit() results = db.execute("SELECT rowid FROM utterances_fts WHERE transcript MATCH 'hello'").fetchall() assert len(results) == 1 ``` - [ ] **Run tests — expect FAIL (database.py does not exist)** ```bash cd /mnt/c/Users/Jeremy/projects/linkedstorm pytest tests/test_database.py -v ``` Expected: `ModuleNotFoundError: No module named 'app.shared.database'` - [ ] **Write `app/shared/database.py`** ```python import sqlite3 import threading from pathlib import Path from app.shared.config import Settings _local = threading.local() def get_db(settings: Settings) -> sqlite3.Connection: if not hasattr(_local, "conn") or _local.conn is None: conn = sqlite3.connect(settings.db_path, check_same_thread=False) conn.row_factory = sqlite3.Row conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA foreign_keys=ON") _local.conn = conn return _local.conn def init_schema(settings: Settings) -> None: Path(settings.clips_dir).mkdir(parents=True, exist_ok=True) db = get_db(settings) db.executescript(""" CREATE TABLE IF NOT EXISTS rooms ( id INTEGER PRIMARY KEY, name TEXT NOT NULL UNIQUE, device_label TEXT, is_active BOOLEAN DEFAULT 0, last_seen DATETIME, created_at DATETIME DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS speakers ( id INTEGER PRIMARY KEY, name TEXT NOT NULL, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS voice_embeddings ( id INTEGER PRIMARY KEY, speaker_id INTEGER NOT NULL REFERENCES speakers(id) ON DELETE CASCADE, embedding BLOB NOT NULL, audio_sample BLOB, created_at DATETIME DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS utterances ( id INTEGER PRIMARY KEY, room_id INTEGER NOT NULL REFERENCES rooms(id), speaker_id INTEGER REFERENCES speakers(id), transcript TEXT NOT NULL, embedding BLOB, match_status TEXT NOT NULL DEFAULT 'unreviewed', match_confidence REAL, start_time DATETIME NOT NULL, end_time DATETIME NOT NULL, audio_clip_path TEXT, livekit_room TEXT, created_at DATETIME DEFAULT CURRENT_TIMESTAMP ); CREATE VIRTUAL TABLE IF NOT EXISTS utterances_fts USING fts5( transcript, content=utterances, content_rowid=id ); CREATE TRIGGER IF NOT EXISTS utterances_ai AFTER INSERT ON utterances BEGIN INSERT INTO utterances_fts(rowid, transcript) VALUES (new.id, new.transcript); END; CREATE TRIGGER IF NOT EXISTS utterances_au AFTER UPDATE OF transcript ON utterances BEGIN INSERT INTO utterances_fts(utterances_fts, rowid, transcript) VALUES ('delete', old.id, old.transcript); INSERT INTO utterances_fts(rowid, transcript) VALUES (new.id, new.transcript); END; CREATE TRIGGER IF NOT EXISTS utterances_ad AFTER DELETE ON utterances BEGIN INSERT INTO utterances_fts(utterances_fts, rowid, transcript) VALUES ('delete', old.id, old.transcript); END; CREATE TABLE IF NOT EXISTS daily_summaries ( id INTEGER PRIMARY KEY, date DATE NOT NULL UNIQUE, summary_text TEXT NOT NULL, generated_at DATETIME DEFAULT CURRENT_TIMESTAMP ); """) db.commit() ``` - [ ] **Run tests — expect PASS** ```bash pytest tests/test_database.py -v ``` Expected: `4 passed` - [ ] **Commit** ```bash git add app/shared/database.py tests/conftest.py tests/test_database.py git commit -m "feat: add database schema with WAL mode and FTS5 triggers" git push origin master ``` --- ## Task 4: Voice recognition module (TDD) **Files:** `app/pipeline/recognition.py`, `tests/test_recognition.py` - [ ] **Write `tests/test_recognition.py`** ```python import numpy as np import struct import pytest from unittest.mock import patch, MagicMock def test_pack_unpack_roundtrip(): from app.pipeline.recognition import pack_embedding, unpack_embedding original = np.array([0.1, 0.2, 0.3, -0.5], dtype=np.float32) packed = pack_embedding(original) recovered = unpack_embedding(packed) np.testing.assert_allclose(recovered, original, rtol=1e-6) def test_cosine_similarity_identical(): from app.pipeline.recognition import cosine_similarity v = np.array([1.0, 0.0, 0.0], dtype=np.float32) assert cosine_similarity(v, v) == pytest.approx(1.0) def test_cosine_similarity_orthogonal(): from app.pipeline.recognition import cosine_similarity a = np.array([1.0, 0.0], dtype=np.float32) b = np.array([0.0, 1.0], dtype=np.float32) assert cosine_similarity(a, b) == pytest.approx(0.0) def test_cosine_similarity_zero_vector(): from app.pipeline.recognition import cosine_similarity a = np.array([0.0, 0.0], dtype=np.float32) b = np.array([1.0, 0.0], dtype=np.float32) assert cosine_similarity(a, b) == 0.0 def test_find_best_match_known(): from app.pipeline.recognition import find_best_speaker_match, pack_embedding ref = np.array([1.0, 0.0, 0.0], dtype=np.float32) query = np.array([0.98, 0.2, 0.0], dtype=np.float32) query /= np.linalg.norm(query) ref /= np.linalg.norm(ref) status, sid, conf = find_best_speaker_match( query, [(1, pack_embedding(ref))], known_threshold=0.85, ambiguous_threshold=0.60 ) assert status == "known" assert sid == 1 assert conf >= 0.85 def test_find_best_match_ambiguous(): from app.pipeline.recognition import find_best_speaker_match, pack_embedding ref = np.array([1.0, 0.0, 0.0], dtype=np.float32) # cos similarity ~0.707 (45 degrees) query = np.array([1.0, 1.0, 0.0], dtype=np.float32) query /= np.linalg.norm(query) ref /= np.linalg.norm(ref) status, sid, conf = find_best_speaker_match( query, [(1, pack_embedding(ref))], known_threshold=0.85, ambiguous_threshold=0.60 ) assert status == "ambiguous" assert sid == 1 def test_find_best_match_unknown(): from app.pipeline.recognition import find_best_speaker_match, pack_embedding ref = np.array([1.0, 0.0, 0.0], dtype=np.float32) query = np.array([0.0, 1.0, 0.0], dtype=np.float32) # orthogonal = 0 similarity status, sid, conf = find_best_speaker_match( query, [(1, pack_embedding(ref))], known_threshold=0.85, ambiguous_threshold=0.60 ) assert status == "unknown" assert sid is None def test_find_best_match_empty_embeddings(): from app.pipeline.recognition import find_best_speaker_match query = np.array([1.0, 0.0], dtype=np.float32) status, sid, conf = find_best_speaker_match(query, []) assert status == "unknown" assert sid is None def test_find_best_match_takes_max_per_speaker(): from app.pipeline.recognition import find_best_speaker_match, pack_embedding # Speaker 1 has two embeddings — one poor, one good ref_bad = np.array([0.0, 1.0, 0.0], dtype=np.float32) ref_good = np.array([1.0, 0.0, 0.0], dtype=np.float32) query = np.array([1.0, 0.0, 0.0], dtype=np.float32) stored = [(1, pack_embedding(ref_bad)), (1, pack_embedding(ref_good))] status, sid, conf = find_best_speaker_match( query, stored, known_threshold=0.85, ambiguous_threshold=0.60 ) assert status == "known" assert sid == 1 def test_embed_audio_calls_resemblyzer(mocker): from app.pipeline.recognition import embed_audio mock_wav = np.zeros(16000, dtype=np.float32) mock_emb = np.ones(256, dtype=np.float32) mocker.patch("app.pipeline.recognition.preprocess_wav", return_value=mock_wav) mock_encoder = MagicMock() mock_encoder.embed_utterance.return_value = mock_emb mocker.patch("app.pipeline.recognition.get_encoder", return_value=mock_encoder) result = embed_audio(b"\x00" * 100) assert result.shape == (256,) mock_encoder.embed_utterance.assert_called_once_with(mock_wav) ``` - [ ] **Run to confirm failure** ```bash pytest tests/test_recognition.py -v ``` Expected: `ModuleNotFoundError: No module named 'app.pipeline.recognition'` - [ ] **Write `app/pipeline/recognition.py`** ```python import struct import io import numpy as np from resemblyzer import VoiceEncoder, preprocess_wav _encoder: VoiceEncoder | None = None def get_encoder() -> VoiceEncoder: global _encoder if _encoder is None: _encoder = VoiceEncoder() return _encoder def embed_audio(wav_bytes: bytes) -> np.ndarray: """Extract 256-dim d-vector from raw WAV bytes.""" wav = preprocess_wav(io.BytesIO(wav_bytes)) return get_encoder().embed_utterance(wav) def pack_embedding(embedding: np.ndarray) -> bytes: n = len(embedding) return struct.pack(f"{n}f", *embedding) def unpack_embedding(blob: bytes) -> np.ndarray: n = len(blob) // 4 return np.array(struct.unpack(f"{n}f", blob), dtype=np.float32) def cosine_similarity(a: np.ndarray, b: np.ndarray) -> float: norm_a = np.linalg.norm(a) norm_b = np.linalg.norm(b) if norm_a == 0 or norm_b == 0: return 0.0 return float(np.dot(a, b) / (norm_a * norm_b)) def find_best_speaker_match( utterance_embedding: np.ndarray, stored_embeddings: list[tuple[int, bytes]], known_threshold: float = 0.85, ambiguous_threshold: float = 0.60, ) -> tuple[str, int | None, float | None]: """ Returns (match_status, speaker_id, confidence). Groups by speaker_id, takes MAX cosine similarity per speaker. """ if not stored_embeddings: return "unknown", None, None best_by_speaker: dict[int, float] = {} for speaker_id, blob in stored_embeddings: ref = unpack_embedding(blob) sim = cosine_similarity(utterance_embedding, ref) if speaker_id not in best_by_speaker or sim > best_by_speaker[speaker_id]: best_by_speaker[speaker_id] = sim best_id = max(best_by_speaker, key=best_by_speaker.__getitem__) best_sim = best_by_speaker[best_id] if best_sim >= known_threshold: return "known", best_id, best_sim elif best_sim >= ambiguous_threshold: return "ambiguous", best_id, best_sim else: return "unknown", None, best_sim ``` - [ ] **Run tests — expect PASS** ```bash pytest tests/test_recognition.py -v ``` Expected: `9 passed` - [ ] **Commit** ```bash git add app/pipeline/recognition.py tests/test_recognition.py git commit -m "feat: voice recognition — resemblyzer embed + cosine similarity matching" git push origin master ``` --- ## Task 5: VAD ingestion buffer (TDD) **Files:** `app/pipeline/ingestion.py`, `tests/test_ingestion.py` - [ ] **Write `tests/test_ingestion.py`** ```python import numpy as np import pytest from datetime import datetime, timezone from unittest.mock import patch, MagicMock def _make_chunk(samples: int = 4000, amplitude: float = 0.0) -> bytes: """Create a fake PCM int16 chunk.""" audio = (np.ones(samples) * amplitude * 32767).astype(np.int16) return audio.tobytes() @pytest.fixture def buf(mocker): """RoomVADBuffer with mocked VAD model.""" mocker.patch( "app.pipeline.ingestion.vad_prob", side_effect=lambda chunk: 0.9 if np.frombuffer(chunk, np.int16).max() > 100 else 0.1, ) from app.pipeline.ingestion import RoomVADBuffer return RoomVADBuffer(silence_threshold_ms=600, min_speech_ms=500, chunk_ms=250) def test_silence_produces_no_segment(buf): ts = datetime(2026, 1, 1, tzinfo=timezone.utc) for _ in range(10): result = buf.process_chunk(_make_chunk(amplitude=0.0), ts) assert result is None def test_speech_then_silence_produces_segment(buf): ts = datetime(2026, 1, 1, tzinfo=timezone.utc) # 4 speech chunks = 1 second of speech for _ in range(4): result = buf.process_chunk(_make_chunk(amplitude=0.5), ts) assert result is None # still accumulating # 3 silence chunks = 750ms silence (> 600ms threshold) segment = None for _ in range(3): segment = buf.process_chunk(_make_chunk(amplitude=0.0), ts) assert segment is not None assert isinstance(segment, tuple) audio_bytes, start_time = segment assert len(audio_bytes) > 0 def test_short_speech_discarded(buf): ts = datetime(2026, 1, 1, tzinfo=timezone.utc) # 1 speech chunk = 250ms (below 500ms min) buf.process_chunk(_make_chunk(amplitude=0.5), ts) # Now silence to trigger emission result = None for _ in range(3): result = buf.process_chunk(_make_chunk(amplitude=0.0), ts) assert result is None # discarded as too short def test_start_time_recorded(buf): ts = datetime(2026, 5, 1, 12, 0, 0, tzinfo=timezone.utc) buf.process_chunk(_make_chunk(amplitude=0.5), ts) for _ in range(3): result = buf.process_chunk(_make_chunk(amplitude=0.0), ts) # Should have started at ts # (segment would be discarded due to being too short in this test, # so test start_time tracking via internal state) # Reset and do a proper speech segment from app.pipeline.ingestion import RoomVADBuffer pass # covered by test_speech_then_silence_produces_segment ``` - [ ] **Run to confirm failure** ```bash pytest tests/test_ingestion.py -v ``` Expected: `ModuleNotFoundError: No module named 'app.pipeline.ingestion'` - [ ] **Write `app/pipeline/ingestion.py`** ```python import asyncio import numpy as np import torch from datetime import datetime, timezone from typing import Optional # Load Silero VAD model once at module level _vad_model = None _vad_utils = None def _load_vad(): global _vad_model, _vad_utils if _vad_model is None: _vad_model, _vad_utils = torch.hub.load( repo_or_dir="snakers4/silero-vad", model="silero_vad", force_reload=False, trust_repo=True, ) return _vad_model def vad_prob(chunk: bytes, sample_rate: int = 16000) -> float: """Return speech probability (0-1) for a PCM int16 chunk.""" model = _load_vad() audio = np.frombuffer(chunk, dtype=np.int16).astype(np.float32) / 32768.0 tensor = torch.FloatTensor(audio) with torch.no_grad(): prob = model(tensor, sample_rate).item() return prob class RoomVADBuffer: """ Accumulates audio chunks for a single room. Emits (audio_bytes, start_time) when a speech segment ends. """ SPEECH_THRESHOLD = 0.5 def __init__( self, silence_threshold_ms: int = 600, min_speech_ms: int = 500, chunk_ms: int = 250, sample_rate: int = 16000, ): self.silence_threshold_ms = silence_threshold_ms self.min_speech_ms = min_speech_ms self.chunk_ms = chunk_ms self.sample_rate = sample_rate self._buffer: list[bytes] = [] self._speaking = False self._silence_ms = 0 self._start_time: Optional[datetime] = None def process_chunk( self, chunk: bytes, timestamp: datetime ) -> Optional[tuple[bytes, datetime]]: """ Returns (audio_bytes, start_time) when a complete speech segment is ready. Returns None while accumulating or during silence. """ prob = vad_prob(chunk) if prob >= self.SPEECH_THRESHOLD: if not self._speaking: self._speaking = True self._start_time = timestamp self._buffer = [] self._buffer.append(chunk) self._silence_ms = 0 elif self._speaking: self._buffer.append(chunk) # include trailing silence self._silence_ms += self.chunk_ms if self._silence_ms >= self.silence_threshold_ms: audio = b"".join(self._buffer) start = self._start_time self._buffer = [] self._speaking = False self._silence_ms = 0 self._start_time = None # Calculate duration (int16 = 2 bytes per sample) duration_ms = (len(audio) / 2 / self.sample_rate) * 1000 if duration_ms >= self.min_speech_ms: return audio, start return None # Per-room registry: room_id -> (buffer, muted_flag) _rooms: dict[str, dict] = {} def get_or_create_room(room_id: str, silence_ms: int, min_speech_ms: int) -> dict: if room_id not in _rooms: _rooms[room_id] = { "buffer": RoomVADBuffer(silence_threshold_ms=silence_ms, min_speech_ms=min_speech_ms), "muted": False, } return _rooms[room_id] def mute_room(room_id: str) -> None: if room_id in _rooms: _rooms[room_id]["muted"] = True def unmute_room(room_id: str) -> None: if room_id in _rooms: _rooms[room_id]["muted"] = False def is_muted(room_id: str) -> bool: return _rooms.get(room_id, {}).get("muted", False) ``` - [ ] **Run tests — expect PASS (VAD is mocked)** ```bash pytest tests/test_ingestion.py -v ``` Expected: `3 passed` (the 4th test is a placeholder, it passes trivially) - [ ] **Commit** ```bash git add app/pipeline/ingestion.py tests/test_ingestion.py git commit -m "feat: VAD ingestion buffer — Silero VAD, per-room accumulation" git push origin master ``` --- ## Task 6: Processing pipeline (TDD) **Files:** `app/pipeline/processor.py`, `tests/test_processor.py` - [ ] **Write `tests/test_processor.py`** ```python import pytest import numpy as np from datetime import datetime, timezone from unittest.mock import AsyncMock, MagicMock, patch @pytest.fixture def proc_settings(settings): return settings @pytest.mark.asyncio async def test_process_utterance_known_speaker(mocker, proc_settings, db): # Setup: insert a room and speaker with embedding db.execute("INSERT INTO rooms (name) VALUES (?)", ("Kitchen",)) db.execute("INSERT INTO speakers (name) VALUES (?)", ("Jeremy",)) db.commit() room = db.execute("SELECT id FROM rooms WHERE name='Kitchen'").fetchone() speaker = db.execute("SELECT id FROM speakers WHERE name='Jeremy'").fetchone() from app.pipeline.recognition import pack_embedding ref_emb = np.array([1.0] + [0.0] * 255, dtype=np.float32) db.execute( "INSERT INTO voice_embeddings (speaker_id, embedding) VALUES (?,?)", (speaker["id"], pack_embedding(ref_emb)), ) db.commit() # Mock Whisper HTTP call mocker.patch( "app.pipeline.processor.transcribe_audio", new=AsyncMock(return_value="hello world"), ) # Mock resemblyzer — return a very similar embedding query_emb = np.array([0.99] + [0.0] * 255, dtype=np.float32) mocker.patch("app.pipeline.processor.embed_audio", return_value=query_emb) # Mock clip saving mocker.patch("app.pipeline.processor.save_clip", return_value="/data/clips/test.wav") from app.pipeline.processor import process_utterance start = datetime(2026, 1, 1, 10, 0, 0, tzinfo=timezone.utc) end = datetime(2026, 1, 1, 10, 0, 3, tzinfo=timezone.utc) utterance_id = await process_utterance( room_id=room["id"], audio_bytes=b"\x00" * 100, start_time=start, end_time=end, settings=proc_settings, db=db, ) assert utterance_id is not None row = db.execute("SELECT * FROM utterances WHERE id=?", (utterance_id,)).fetchone() assert row["transcript"] == "hello world" assert row["match_status"] == "known" assert row["speaker_id"] == speaker["id"] @pytest.mark.asyncio async def test_process_utterance_unknown_speaker(mocker, proc_settings, db): db.execute("INSERT INTO rooms (name) VALUES (?)", ("Office",)) db.commit() room = db.execute("SELECT id FROM rooms WHERE name='Office'").fetchone() mocker.patch( "app.pipeline.processor.transcribe_audio", new=AsyncMock(return_value="test transcript"), ) mocker.patch( "app.pipeline.processor.embed_audio", return_value=np.zeros(256, dtype=np.float32), ) mocker.patch("app.pipeline.processor.save_clip", return_value="/data/clips/test.wav") from app.pipeline.processor import process_utterance start = datetime(2026, 1, 1, 10, 0, 0, tzinfo=timezone.utc) end = datetime(2026, 1, 1, 10, 0, 2, tzinfo=timezone.utc) utterance_id = await process_utterance( room_id=room["id"], audio_bytes=b"\x00" * 100, start_time=start, end_time=end, settings=proc_settings, db=db, ) row = db.execute("SELECT * FROM utterances WHERE id=?", (utterance_id,)).fetchone() assert row["match_status"] == "unknown" assert row["speaker_id"] is None ``` - [ ] **Run to confirm failure** ```bash pytest tests/test_processor.py -v ``` Expected: `ModuleNotFoundError: No module named 'app.pipeline.processor'` - [ ] **Write `app/pipeline/processor.py`** ```python import asyncio import sqlite3 import wave import io from datetime import datetime, timezone from pathlib import Path from typing import Optional import httpx import numpy as np from app.shared.config import Settings from app.pipeline.recognition import embed_audio, find_best_speaker_match, pack_embedding async def transcribe_audio(wav_bytes: bytes, speaches_url: str) -> str: """POST audio to Speaches/Whisper, return transcript text.""" async with httpx.AsyncClient(timeout=30.0) as client: resp = await client.post( f"{speaches_url}/v1/audio/transcriptions", files={"file": ("audio.wav", wav_bytes, "audio/wav")}, data={"model": "Systran/faster-whisper-large-v3-turbo"}, ) resp.raise_for_status() return resp.json()["text"].strip() def save_clip(audio_bytes: bytes, clips_dir: str, utterance_id: int, start_time: datetime) -> str: """Save raw PCM bytes as a WAV file, return the path.""" date_str = start_time.strftime("%Y-%m-%d") clip_dir = Path(clips_dir) / date_str clip_dir.mkdir(parents=True, exist_ok=True) path = clip_dir / f"{utterance_id}.wav" with wave.open(str(path), "wb") as wf: wf.setnchannels(1) wf.setsampwidth(2) # int16 wf.setframerate(16000) wf.writeframes(audio_bytes) return str(path) async def process_utterance( room_id: int, audio_bytes: bytes, start_time: datetime, end_time: datetime, settings: Settings, db: sqlite3.Connection, ) -> Optional[int]: """ Full pipeline for one speech segment: 1. Whisper STT + resemblyzer embed (concurrent) 2. Speaker matching 3. SQLite write + WAV clip save Returns the new utterance row id, or None on failure. """ try: transcript, embedding = await asyncio.gather( transcribe_audio(audio_bytes, settings.speaches_url), asyncio.to_thread(embed_audio, audio_bytes), ) except Exception: return None # Load all stored embeddings for matching rows = db.execute("SELECT speaker_id, embedding FROM voice_embeddings").fetchall() stored = [(r["speaker_id"], r["embedding"]) for r in rows] match_status, speaker_id, confidence = find_best_speaker_match( embedding, stored, known_threshold=settings.speaker_known_threshold, ambiguous_threshold=settings.speaker_ambiguous_threshold, ) # Insert utterance row first to get the id cur = db.execute( """INSERT INTO utterances (room_id, speaker_id, transcript, embedding, match_status, match_confidence, start_time, end_time) VALUES (?,?,?,?,?,?,?,?)""", ( room_id, speaker_id, transcript, pack_embedding(embedding), match_status, confidence, start_time.isoformat(), end_time.isoformat(), ), ) db.commit() utterance_id = cur.lastrowid # Save WAV clip and update row with path clip_path = save_clip(audio_bytes, settings.clips_dir, utterance_id, start_time) db.execute( "UPDATE utterances SET audio_clip_path=? WHERE id=?", (clip_path, utterance_id), ) db.commit() # If known match, store embedding as an additional reference for this speaker if match_status == "known" and speaker_id is not None: db.execute( "INSERT INTO voice_embeddings (speaker_id, embedding) VALUES (?,?)", (speaker_id, pack_embedding(embedding)), ) db.commit() return utterance_id ``` - [ ] **Run tests — expect PASS** ```bash pytest tests/test_processor.py -v ``` Expected: `2 passed` - [ ] **Commit** ```bash git add app/pipeline/processor.py tests/test_processor.py git commit -m "feat: processing pipeline — Whisper + resemblyzer + speaker match + SQLite write" git push origin master ``` --- ## Task 7: Pipeline FastAPI app (WebSocket + internal HTTP) **Files:** `app/pipeline/main.py` *(No TDD — integration test requires running server; covered in Task 18 deployment verification)* - [ ] **Write `app/pipeline/main.py`** ```python import asyncio import sqlite3 from contextlib import asynccontextmanager from datetime import datetime, timezone import uvicorn from fastapi import FastAPI, WebSocket, WebSocketDisconnect, HTTPException from fastapi.responses import JSONResponse from app.shared.config import get_settings, Settings from app.shared.database import get_db, init_schema from app.pipeline.ingestion import get_or_create_room, mute_room, unmute_room, is_muted from app.pipeline.processor import process_utterance _settings: Settings | None = None def settings() -> Settings: global _settings if _settings is None: _settings = get_settings() return _settings @asynccontextmanager async def lifespan(app: FastAPI): s = settings() init_schema(s) yield app = FastAPI(lifespan=lifespan) @app.websocket("/ws/{room_name}") async def audio_ws(websocket: WebSocket, room_name: str): await websocket.accept() s = settings() db = get_db(s) # Upsert room in DB db.execute( "INSERT INTO rooms (name, is_active, last_seen) VALUES (?,1,CURRENT_TIMESTAMP) " "ON CONFLICT(name) DO UPDATE SET is_active=1, last_seen=CURRENT_TIMESTAMP", (room_name,), ) db.commit() room = db.execute("SELECT id FROM rooms WHERE name=?", (room_name,)).fetchone() room_id = room["id"] room_state = get_or_create_room(room_name, s.vad_silence_ms, s.vad_min_speech_ms) try: while True: data = await websocket.receive() if "bytes" in data: chunk = data["bytes"] if not is_muted(room_name): db.execute( "UPDATE rooms SET last_seen=CURRENT_TIMESTAMP WHERE id=?", (room_id,), ) now = datetime.now(timezone.utc) result = room_state["buffer"].process_chunk(chunk, now) if result is not None: audio_bytes, start_time = result end_time = datetime.now(timezone.utc) asyncio.create_task( process_utterance(room_id, audio_bytes, start_time, end_time, s, db) ) elif "text" in data: import json msg = json.loads(data["text"]) if msg.get("type") == "mute": mute_room(room_name) elif msg.get("type") == "unmute": unmute_room(room_name) except WebSocketDisconnect: db.execute( "UPDATE rooms SET is_active=0 WHERE id=?", (room_id,) ) db.commit() @app.post("/internal/rooms/{room_name}/mute") async def internal_mute(room_name: str): mute_room(room_name) return {"status": "muted", "room": room_name} @app.post("/internal/rooms/{room_name}/unmute") async def internal_unmute(room_name: str): unmute_room(room_name) return {"status": "unmuted", "room": room_name} @app.get("/health") async def health(): return {"status": "ok"} if __name__ == "__main__": uvicorn.run("app.pipeline.main:app", host="0.0.0.0", port=8300, reload=False) ``` - [ ] **Commit** ```bash git add app/pipeline/main.py git commit -m "feat: pipeline FastAPI app — WebSocket audio ingestion + internal mute API" git push origin master ``` --- ## Task 8: API routes — utterances (TDD) **Files:** `app/api/routes/utterances.py`, `tests/routes/conftest.py`, `tests/routes/test_utterances.py` - [ ] **Write `tests/routes/conftest.py`** ```python import pytest from fastapi.testclient import TestClient from app.shared.database import init_schema, get_db from app.shared.config import Settings @pytest.fixture def api_client(settings): from app.api.main import create_app app = create_app(settings) return TestClient(app) @pytest.fixture def seeded_db(db): """DB with a room and speaker pre-inserted.""" db.execute("INSERT INTO rooms (name) VALUES (?)", ("Kitchen",)) db.execute("INSERT INTO speakers (name) VALUES (?)", ("Jeremy",)) db.commit() return db ``` - [ ] **Write `tests/routes/test_utterances.py`** ```python from datetime import datetime def test_list_utterances_empty(api_client): resp = api_client.get("/api/utterances") assert resp.status_code == 200 assert resp.json() == [] def test_list_utterances_with_data(api_client, seeded_db): room = seeded_db.execute("SELECT id FROM rooms WHERE name='Kitchen'").fetchone() seeded_db.execute( "INSERT INTO utterances (room_id, transcript, match_status, start_time, end_time) " "VALUES (?,?,?,?,?)", (room["id"], "hello world", "unknown", "2026-01-01 10:00:00", "2026-01-01 10:00:03"), ) seeded_db.commit() resp = api_client.get("/api/utterances") assert resp.status_code == 200 data = resp.json() assert len(data) == 1 assert data[0]["transcript"] == "hello world" assert data[0]["room_name"] == "Kitchen" def test_filter_utterances_by_date(api_client, seeded_db): room = seeded_db.execute("SELECT id FROM rooms WHERE name='Kitchen'").fetchone() seeded_db.execute( "INSERT INTO utterances (room_id, transcript, match_status, start_time, end_time) VALUES (?,?,?,?,?)", (room["id"], "today speech", "unknown", "2026-05-27 10:00:00", "2026-05-27 10:00:03"), ) seeded_db.execute( "INSERT INTO utterances (room_id, transcript, match_status, start_time, end_time) VALUES (?,?,?,?,?)", (room["id"], "yesterday speech", "unknown", "2026-05-26 10:00:00", "2026-05-26 10:00:03"), ) seeded_db.commit() resp = api_client.get("/api/utterances?date=2026-05-27") data = resp.json() assert len(data) == 1 assert data[0]["transcript"] == "today speech" def test_tag_utterance_existing_speaker(api_client, seeded_db): room = seeded_db.execute("SELECT id FROM rooms WHERE name='Kitchen'").fetchone() speaker = seeded_db.execute("SELECT id FROM speakers WHERE name='Jeremy'").fetchone() seeded_db.execute( "INSERT INTO utterances (room_id, transcript, match_status, start_time, end_time, embedding) VALUES (?,?,?,?,?,?)", (room["id"], "tag me", "unknown", "2026-01-01 10:00:00", "2026-01-01 10:00:03", b"\x00" * 1024), ) seeded_db.commit() utt = seeded_db.execute("SELECT id FROM utterances").fetchone() resp = api_client.post(f"/api/utterances/{utt['id']}/tag", json={"speaker_id": speaker["id"]}) assert resp.status_code == 200 updated = seeded_db.execute("SELECT * FROM utterances WHERE id=?", (utt["id"],)).fetchone() assert updated["speaker_id"] == speaker["id"] assert updated["match_status"] == "known" assert updated["match_confidence"] == 1.0 def test_tag_utterance_create_new_speaker(api_client, seeded_db): room = seeded_db.execute("SELECT id FROM rooms WHERE name='Kitchen'").fetchone() seeded_db.execute( "INSERT INTO utterances (room_id, transcript, match_status, start_time, end_time) VALUES (?,?,?,?,?)", (room["id"], "new speaker", "unknown", "2026-01-01 10:00:00", "2026-01-01 10:00:03"), ) seeded_db.commit() utt = seeded_db.execute("SELECT id FROM utterances WHERE transcript='new speaker'").fetchone() resp = api_client.post( f"/api/utterances/{utt['id']}/tag", json={"name": "Alice", "create_new": True}, ) assert resp.status_code == 200 alice = seeded_db.execute("SELECT id FROM speakers WHERE name='Alice'").fetchone() assert alice is not None def test_dismiss_utterance(api_client, seeded_db): room = seeded_db.execute("SELECT id FROM rooms WHERE name='Kitchen'").fetchone() seeded_db.execute( "INSERT INTO utterances (room_id, transcript, match_status, start_time, end_time) VALUES (?,?,?,?,?)", (room["id"], "dismiss me", "unreviewed", "2026-01-01 10:00:00", "2026-01-01 10:00:03"), ) seeded_db.commit() utt = seeded_db.execute("SELECT id FROM utterances WHERE transcript='dismiss me'").fetchone() resp = api_client.post(f"/api/utterances/{utt['id']}/dismiss") assert resp.status_code == 200 updated = seeded_db.execute("SELECT match_status FROM utterances WHERE id=?", (utt["id"],)).fetchone() assert updated["match_status"] == "unknown" ``` - [ ] **Run to confirm failure** ```bash pytest tests/routes/test_utterances.py -v ``` Expected: `ModuleNotFoundError: No module named 'app.api.main'` - [ ] **Write `app/api/routes/utterances.py`** ```python import sqlite3 from typing import Optional from fastapi import APIRouter, Depends, HTTPException from fastapi.responses import FileResponse from pydantic import BaseModel from app.shared.config import Settings from app.pipeline.recognition import pack_embedding, unpack_embedding router = APIRouter(prefix="/api/utterances") class TagRequest(BaseModel): speaker_id: Optional[int] = None name: Optional[str] = None create_new: bool = False def _row_to_dict(row: sqlite3.Row) -> dict: return dict(row) def get_utterances( db: sqlite3.Connection, date: Optional[str] = None, room_id: Optional[int] = None, speaker_id: Optional[int] = None, match_status: Optional[str] = None, limit: int = 200, ) -> list[dict]: query = """ SELECT u.*, r.name as room_name, s.name as speaker_name FROM utterances u JOIN rooms r ON u.room_id = r.id LEFT JOIN speakers s ON u.speaker_id = s.id WHERE 1=1 """ params: list = [] if date: query += " AND date(u.start_time) = ?" params.append(date) if room_id: query += " AND u.room_id = ?" params.append(room_id) if speaker_id: query += " AND u.speaker_id = ?" params.append(speaker_id) if match_status: query += " AND u.match_status = ?" params.append(match_status) query += " ORDER BY u.start_time DESC LIMIT ?" params.append(limit) rows = db.execute(query, params).fetchall() return [_row_to_dict(r) for r in rows] def make_router(settings: Settings, db_factory): @router.get("") def list_utterances( date: Optional[str] = None, room_id: Optional[int] = None, speaker_id: Optional[int] = None, match_status: Optional[str] = None, ): db = db_factory() return get_utterances(db, date, room_id, speaker_id, match_status) @router.get("/{utterance_id}/clip") def get_clip(utterance_id: int): db = db_factory() row = db.execute( "SELECT audio_clip_path FROM utterances WHERE id=?", (utterance_id,) ).fetchone() if not row or not row["audio_clip_path"]: raise HTTPException(404, "Clip not found or expired") return FileResponse(row["audio_clip_path"], media_type="audio/wav") @router.post("/{utterance_id}/tag") def tag_utterance(utterance_id: int, body: TagRequest): db = db_factory() row = db.execute("SELECT * FROM utterances WHERE id=?", (utterance_id,)).fetchone() if not row: raise HTTPException(404, "Utterance not found") speaker_id = body.speaker_id if body.create_new and body.name: cur = db.execute("INSERT INTO speakers (name) VALUES (?)", (body.name,)) db.commit() speaker_id = cur.lastrowid if speaker_id is None: raise HTTPException(400, "Provide speaker_id or name+create_new") db.execute( "UPDATE utterances SET speaker_id=?, match_status='known', match_confidence=1.0 WHERE id=?", (speaker_id, utterance_id), ) db.commit() # Store utterance embedding as a reference for this speaker if row["embedding"]: db.execute( "INSERT INTO voice_embeddings (speaker_id, embedding) VALUES (?,?)", (speaker_id, row["embedding"]), ) db.commit() return {"ok": True, "speaker_id": speaker_id} @router.post("/{utterance_id}/dismiss") def dismiss_utterance(utterance_id: int): db = db_factory() db.execute( "UPDATE utterances SET match_status='unknown' WHERE id=?", (utterance_id,) ) db.commit() return {"ok": True} return router ``` - [ ] **Write minimal `app/api/main.py` (enough for tests to import)** ```python import sqlite3 from fastapi import FastAPI from fastapi.staticfiles import StaticFiles from pathlib import Path from app.shared.config import Settings from app.shared.database import get_db, init_schema from app.api.routes import utterances, speakers, rooms, search, rag def create_app(settings: Settings) -> FastAPI: init_schema(settings) app = FastAPI(title="LinkedStorm") def db_factory() -> sqlite3.Connection: return get_db(settings) app.include_router(utterances.make_router(settings, db_factory)) app.include_router(speakers.make_router(settings, db_factory)) app.include_router(rooms.make_router(settings, db_factory)) app.include_router(search.make_router(settings, db_factory)) app.include_router(rag.make_router(settings, db_factory)) static_dir = Path(__file__).parent / "static" if static_dir.exists(): app.mount("/", StaticFiles(directory=str(static_dir), html=True), name="static") return app ``` Note: `speakers`, `rooms`, `search`, `rag` routers are stub files — write them now: - [ ] **Write stub `app/api/routes/speakers.py`** ```python from fastapi import APIRouter router = APIRouter(prefix="/api/speakers") def make_router(settings, db_factory): return router ``` - [ ] **Write stub `app/api/routes/rooms.py`** ```python from fastapi import APIRouter router = APIRouter(prefix="/api/rooms") def make_router(settings, db_factory): return router ``` - [ ] **Write stub `app/api/routes/search.py`** ```python from fastapi import APIRouter router = APIRouter(prefix="/api/search") def make_router(settings, db_factory): return router ``` - [ ] **Write stub `app/api/routes/rag.py`** ```python from fastapi import APIRouter router = APIRouter(prefix="/api") def make_router(settings, db_factory): return router ``` - [ ] **Run tests — expect PASS** ```bash pytest tests/routes/test_utterances.py -v ``` Expected: `6 passed` - [ ] **Commit** ```bash git add app/api/ tests/routes/ git commit -m "feat: utterances API routes — list, clip, tag, dismiss" git push origin master ``` --- ## Task 9: API routes — speakers, rooms, search, RAG (TDD) **Files:** `app/api/routes/speakers.py`, `rooms.py`, `search.py`, `rag.py` (replace stubs), plus tests - [ ] **Write `tests/routes/test_speakers.py`** ```python def test_list_speakers_empty(api_client): resp = api_client.get("/api/speakers") assert resp.status_code == 200 assert resp.json() == [] def test_create_speaker(api_client): resp = api_client.post("/api/speakers", json={"name": "Alice"}) assert resp.status_code == 200 data = resp.json() assert data["name"] == "Alice" assert "id" in data def test_list_speakers_includes_utterance_count(api_client, seeded_db): room = seeded_db.execute("SELECT id FROM rooms WHERE name='Kitchen'").fetchone() speaker = seeded_db.execute("SELECT id FROM speakers WHERE name='Jeremy'").fetchone() seeded_db.execute( "INSERT INTO utterances (room_id, speaker_id, transcript, match_status, start_time, end_time) VALUES (?,?,?,?,?,?)", (room["id"], speaker["id"], "hi", "known", "2026-01-01 10:00:00", "2026-01-01 10:00:01"), ) seeded_db.commit() resp = api_client.get("/api/speakers") data = resp.json() jeremy = next(s for s in data if s["name"] == "Jeremy") assert jeremy["utterance_count"] == 1 ``` - [ ] **Write `tests/routes/test_rooms.py`** ```python def test_list_rooms(api_client, seeded_db): resp = api_client.get("/api/rooms") assert resp.status_code == 200 data = resp.json() assert any(r["name"] == "Kitchen" for r in data) def test_mute_room_calls_pipeline(api_client, seeded_db, mocker): mock_post = mocker.patch("app.api.routes.rooms.httpx.post") mock_post.return_value.status_code = 200 room = seeded_db.execute("SELECT id FROM rooms WHERE name='Kitchen'").fetchone() resp = api_client.post(f"/api/rooms/{room['id']}/mute") assert resp.status_code == 200 mock_post.assert_called_once() ``` - [ ] **Write `tests/routes/test_search.py`** ```python def test_search_returns_matches(api_client, seeded_db): room = seeded_db.execute("SELECT id FROM rooms WHERE name='Kitchen'").fetchone() seeded_db.execute( "INSERT INTO utterances (room_id, transcript, match_status, start_time, end_time) VALUES (?,?,?,?,?)", (room["id"], "the grocery list is ready", "unknown", "2026-01-01 10:00:00", "2026-01-01 10:00:03"), ) seeded_db.commit() resp = api_client.get("/api/search?q=grocery") assert resp.status_code == 200 data = resp.json() assert len(data) == 1 assert "grocery" in data[0]["transcript"] def test_search_no_results(api_client): resp = api_client.get("/api/search?q=xyznotfound") assert resp.status_code == 200 assert resp.json() == [] ``` - [ ] **Write `tests/routes/test_rag.py`** ```python def test_ask_returns_response(api_client, seeded_db, mocker): mock_post = mocker.patch("app.api.routes.rag.httpx.post") mock_post.return_value.json.return_value = { "message": {"content": "You talked about groceries."} } mock_post.return_value.status_code = 200 room = seeded_db.execute("SELECT id FROM rooms WHERE name='Kitchen'").fetchone() seeded_db.execute( "INSERT INTO utterances (room_id, transcript, match_status, start_time, end_time) VALUES (?,?,?,?,?)", (room["id"], "we need milk from the store", "unknown", "2026-01-01 10:00:00", "2026-01-01 10:00:03"), ) seeded_db.commit() resp = api_client.post("/api/ask", json={"question": "What did we need from the store?"}) assert resp.status_code == 200 data = resp.json() assert "answer" in data def test_get_summary_not_found(api_client): resp = api_client.get("/api/summary/2026-01-01") assert resp.status_code == 404 ``` - [ ] **Run to confirm failures** ```bash pytest tests/routes/ -v ``` Expected: New test files fail. `test_utterances.py` still passes. - [ ] **Replace stub `app/api/routes/speakers.py`** ```python import sqlite3 from fastapi import APIRouter, HTTPException from pydantic import BaseModel router = APIRouter(prefix="/api/speakers") class CreateSpeakerRequest(BaseModel): name: str def make_router(settings, db_factory): @router.get("") def list_speakers(): db: sqlite3.Connection = db_factory() rows = db.execute(""" SELECT s.id, s.name, s.created_at, COUNT(u.id) as utterance_count FROM speakers s LEFT JOIN utterances u ON u.speaker_id = s.id GROUP BY s.id ORDER BY s.name """).fetchall() return [dict(r) for r in rows] @router.post("") def create_speaker(body: CreateSpeakerRequest): db: sqlite3.Connection = db_factory() cur = db.execute("INSERT INTO speakers (name) VALUES (?)", (body.name,)) db.commit() row = db.execute("SELECT * FROM speakers WHERE id=?", (cur.lastrowid,)).fetchone() return dict(row) return router ``` - [ ] **Replace stub `app/api/routes/rooms.py`** ```python import sqlite3 import httpx from fastapi import APIRouter, HTTPException from app.shared.config import Settings router = APIRouter(prefix="/api/rooms") def make_router(settings: Settings, db_factory): @router.get("") def list_rooms(): db: sqlite3.Connection = db_factory() rows = db.execute( "SELECT id, name, device_label, is_active, last_seen, created_at FROM rooms ORDER BY name" ).fetchall() return [dict(r) for r in rows] @router.post("/{room_id}/mute") def mute_room(room_id: int): db: sqlite3.Connection = db_factory() row = db.execute("SELECT name FROM rooms WHERE id=?", (room_id,)).fetchone() if not row: raise HTTPException(404, "Room not found") httpx.post(f"{settings.pipeline_internal_url}/internal/rooms/{row['name']}/mute") return {"ok": True, "room": row["name"]} @router.post("/{room_id}/unmute") def unmute_room(room_id: int): db: sqlite3.Connection = db_factory() row = db.execute("SELECT name FROM rooms WHERE id=?", (room_id,)).fetchone() if not row: raise HTTPException(404, "Room not found") httpx.post(f"{settings.pipeline_internal_url}/internal/rooms/{row['name']}/unmute") return {"ok": True, "room": row["name"]} return router ``` - [ ] **Replace stub `app/api/routes/search.py`** ```python import sqlite3 from typing import Optional from fastapi import APIRouter, HTTPException router = APIRouter(prefix="/api") def make_router(settings, db_factory): @router.get("/search") def search(q: str, date: Optional[str] = None): if not q.strip(): return [] db: sqlite3.Connection = db_factory() fts_query = q.strip() rows = db.execute(""" SELECT u.*, r.name as room_name, s.name as speaker_name FROM utterances_fts fts JOIN utterances u ON u.id = fts.rowid JOIN rooms r ON u.room_id = r.id LEFT JOIN speakers s ON u.speaker_id = s.id WHERE utterances_fts MATCH ? ORDER BY u.start_time DESC LIMIT 100 """, (fts_query,)).fetchall() return [dict(r) for r in rows] return router ``` - [ ] **Replace stub `app/api/routes/rag.py`** ```python import sqlite3 import httpx from fastapi import APIRouter, HTTPException from pydantic import BaseModel from app.shared.config import Settings router = APIRouter(prefix="/api") class AskRequest(BaseModel): question: str date: str | None = None def make_router(settings: Settings, db_factory): @router.post("/ask") def ask(body: AskRequest): db: sqlite3.Connection = db_factory() # Fetch recent transcripts as context (last 200 utterances for the date or today) date_filter = body.date or "date('now')" rows = db.execute(""" SELECT u.start_time, u.transcript, r.name as room_name, s.name as speaker_name FROM utterances u JOIN rooms r ON u.room_id = r.id LEFT JOIN speakers s ON u.speaker_id = s.id WHERE date(u.start_time) = date(?) ORDER BY u.start_time ASC LIMIT 200 """, (body.date or "now",)).fetchall() context_lines = [ f"[{r['room_name']} {r['start_time']}] {r['speaker_name'] or 'Unknown'}: {r['transcript']}" for r in rows ] context = "\n".join(context_lines) if context_lines else "(no transcripts for this date)" prompt = ( f"You are a helpful assistant with access to household conversation transcripts.\n\n" f"Transcripts:\n{context}\n\n" f"Question: {body.question}\n\n" f"Answer based only on the transcripts above. Cite the time and room when relevant." ) resp = httpx.post( f"{settings.ollama_url}/api/chat", json={ "model": settings.ollama_model, "messages": [{"role": "user", "content": prompt}], "stream": False, }, timeout=60.0, ) resp.raise_for_status() answer = resp.json()["message"]["content"] return {"answer": answer, "context_count": len(rows)} @router.get("/summary/{date}") def get_summary(date: str): db: sqlite3.Connection = db_factory() row = db.execute( "SELECT * FROM daily_summaries WHERE date=?", (date,) ).fetchone() if not row: raise HTTPException(404, f"No summary for {date}") return dict(row) return router ``` - [ ] **Run all route tests** ```bash pytest tests/routes/ -v ``` Expected: `~14 passed` - [ ] **Commit** ```bash git add app/api/routes/ tests/routes/ git commit -m "feat: API routes — speakers, rooms, search, RAG" git push origin master ``` --- ## Task 10: Cron jobs (TDD) **Files:** `app/cron/daily_summary.py`, `app/cron/cluster_unknowns.py`, `app/cron/purge_clips.py`, `app/cron/main.py`, `tests/test_cron.py` - [ ] **Write `tests/test_cron.py`** ```python import pytest import os from datetime import datetime, timezone from pathlib import Path from unittest.mock import patch, MagicMock def _seed_utterances(db, room_id, count=3): for i in range(count): db.execute( "INSERT INTO utterances (room_id, transcript, match_status, start_time, end_time) VALUES (?,?,?,?,?)", (room_id, f"utterance {i}", "unknown", "2026-05-27 10:00:00", "2026-05-27 10:00:03"), ) db.commit() def test_daily_summary_inserts_row(mocker, settings, db): db.execute("INSERT INTO rooms (name) VALUES (?)", ("Kitchen",)) db.commit() room = db.execute("SELECT id FROM rooms WHERE name='Kitchen'").fetchone() _seed_utterances(db, room["id"]) mock_post = mocker.patch("app.cron.daily_summary.httpx.post") mock_post.return_value.json.return_value = {"message": {"content": "Busy day."}} mock_post.return_value.raise_for_status = MagicMock() from app.cron.daily_summary import generate_daily_summary generate_daily_summary("2026-05-27", settings, db) row = db.execute("SELECT * FROM daily_summaries WHERE date='2026-05-27'").fetchone() assert row is not None assert row["summary_text"] == "Busy day." def test_daily_summary_skips_if_no_utterances(mocker, settings, db): mock_post = mocker.patch("app.cron.daily_summary.httpx.post") from app.cron.daily_summary import generate_daily_summary generate_daily_summary("2026-05-27", settings, db) mock_post.assert_not_called() def test_purge_clips_deletes_old_files(settings, db, tmp_path): from app.cron.purge_clips import purge_old_clips import shutil clips_dir = tmp_path / "clips" old_dir = clips_dir / "2026-04-01" old_dir.mkdir(parents=True) old_file = old_dir / "1.wav" old_file.write_bytes(b"\x00" * 100) new_dir = clips_dir / "2026-05-27" new_dir.mkdir(parents=True) new_file = new_dir / "2.wav" new_file.write_bytes(b"\x00" * 100) settings_override = type(settings)( db_path=settings.db_path, clips_dir=str(clips_dir), clip_retention_days=30, speaches_url=settings.speaches_url, ollama_url=settings.ollama_url, livekit_api_key=settings.livekit_api_key, livekit_api_secret=settings.livekit_api_secret, ) purge_old_clips("2026-05-28", settings_override, db) assert not old_file.exists() assert new_file.exists() ``` - [ ] **Run to confirm failures** ```bash pytest tests/test_cron.py -v ``` Expected: `ModuleNotFoundError: No module named 'app.cron.daily_summary'` - [ ] **Write `app/cron/daily_summary.py`** ```python import sqlite3 import httpx from app.shared.config import Settings def generate_daily_summary(date: str, settings: Settings, db: sqlite3.Connection) -> None: rows = db.execute(""" SELECT u.start_time, u.transcript, r.name as room_name, s.name as speaker_name FROM utterances u JOIN rooms r ON u.room_id = r.id LEFT JOIN speakers s ON u.speaker_id = s.id WHERE date(u.start_time) = ? ORDER BY u.start_time ASC """, (date,)).fetchall() if not rows: return lines = [ f"[{r['room_name']} {r['start_time']}] {r['speaker_name'] or 'Unknown'}: {r['transcript']}" for r in rows ] context = "\n".join(lines) prompt = ( f"Summarize the following household conversations from {date} in 2-4 sentences. " f"Focus on key decisions, activities, and topics discussed. Do not transcribe verbatim.\n\n" f"{context}" ) resp = httpx.post( f"{settings.ollama_url}/api/chat", json={ "model": settings.ollama_model, "messages": [{"role": "user", "content": prompt}], "stream": False, }, timeout=120.0, ) resp.raise_for_status() summary = resp.json()["message"]["content"] db.execute( "INSERT INTO daily_summaries (date, summary_text) VALUES (?,?) " "ON CONFLICT(date) DO UPDATE SET summary_text=?, generated_at=CURRENT_TIMESTAMP", (date, summary, summary), ) db.commit() ``` - [ ] **Write `app/cron/cluster_unknowns.py`** ```python import sqlite3 import numpy as np from app.shared.config import Settings from app.pipeline.recognition import unpack_embedding, cosine_similarity def cluster_unknown_speakers(settings: Settings, db: sqlite3.Connection) -> None: """ Groups unknown/ambiguous utterances from the last 7 days by voice similarity. Assigns speaker_id from an existing 'Unknown #N' speaker or creates a new one. Threshold: 0.70 cosine similarity. """ CLUSTER_THRESHOLD = 0.70 rows = db.execute(""" SELECT id, embedding FROM utterances WHERE match_status IN ('unknown', 'ambiguous') AND embedding IS NOT NULL AND date(start_time) >= date('now', '-7 days') ORDER BY start_time ASC """).fetchall() if not rows: return # Build clusters greedily clusters: list[list[int]] = [] # list of utterance id lists cluster_centroids: list[np.ndarray] = [] for row in rows: emb = unpack_embedding(row["embedding"]) best_cluster = None best_sim = 0.0 for i, centroid in enumerate(cluster_centroids): sim = cosine_similarity(emb, centroid) if sim > best_sim and sim >= CLUSTER_THRESHOLD: best_sim = sim best_cluster = i if best_cluster is not None: clusters[best_cluster].append(row["id"]) # Update centroid as mean n = len(clusters[best_cluster]) cluster_centroids[best_cluster] = ( (cluster_centroids[best_cluster] * (n - 1) + emb) / n ) else: clusters.append([row["id"]]) cluster_centroids.append(emb) # Assign Unknown #N speakers to clusters with > 1 utterance for i, cluster in enumerate(clusters): if len(cluster) < 2: continue name = f"Unknown #{i + 1}" existing = db.execute("SELECT id FROM speakers WHERE name=?", (name,)).fetchone() if existing: speaker_id = existing["id"] else: cur = db.execute("INSERT INTO speakers (name) VALUES (?)", (name,)) db.commit() speaker_id = cur.lastrowid for utt_id in cluster: db.execute( "UPDATE utterances SET speaker_id=?, match_status='ambiguous' WHERE id=?", (speaker_id, utt_id), ) db.commit() ``` - [ ] **Write `app/cron/purge_clips.py`** ```python import sqlite3 from datetime import datetime, timedelta from pathlib import Path from app.shared.config import Settings def purge_old_clips(today: str, settings: Settings, db: sqlite3.Connection) -> None: cutoff = datetime.strptime(today, "%Y-%m-%d") - timedelta(days=settings.clip_retention_days) clips_root = Path(settings.clips_dir) deleted = 0 for date_dir in clips_root.iterdir(): if not date_dir.is_dir(): continue try: dir_date = datetime.strptime(date_dir.name, "%Y-%m-%d") except ValueError: continue if dir_date < cutoff: for wav in date_dir.glob("*.wav"): wav.unlink() deleted += 1 try: date_dir.rmdir() except OSError: pass # Null out audio_clip_path for purged clips cutoff_str = cutoff.strftime("%Y-%m-%d") db.execute( "UPDATE utterances SET audio_clip_path=NULL WHERE date(start_time) < ?", (cutoff_str,), ) db.commit() ``` - [ ] **Write `app/cron/main.py`** ```python import time from datetime import datetime, timezone from app.shared.config import get_settings from app.shared.database import get_db, init_schema from app.cron.daily_summary import generate_daily_summary from app.cron.cluster_unknowns import cluster_unknown_speakers from app.cron.purge_clips import purge_old_clips def run_nightly(settings, db): today = datetime.now(timezone.utc).strftime("%Y-%m-%d") yesterday = (datetime.now(timezone.utc).date()).strftime("%Y-%m-%d") generate_daily_summary(yesterday, settings, db) cluster_unknown_speakers(settings, db) purge_old_clips(today, settings, db) def main(): settings = get_settings() init_schema(settings) db = get_db(settings) while True: now = datetime.now(timezone.utc) # Sleep until midnight UTC next_midnight = now.replace(hour=0, minute=0, second=0, microsecond=0) if next_midnight <= now: from datetime import timedelta next_midnight += timedelta(days=1) sleep_secs = (next_midnight - now).total_seconds() time.sleep(sleep_secs) run_nightly(settings, db) if __name__ == "__main__": main() ``` - [ ] **Run tests — expect PASS** ```bash pytest tests/test_cron.py -v ``` Expected: `3 passed` - [ ] **Commit** ```bash git add app/cron/ tests/test_cron.py git commit -m "feat: cron jobs — daily summary, speaker clustering, clip purge" git push origin master ``` --- ## Task 11: Capture page **Files:** `app/api/static/capture.html` *(No unit tests — browser JS; verified manually in Task 18)* - [ ] **Create `app/api/static/capture.html`** ```html LinkedStorm Capture

⚡ LinkedStorm

``` - [ ] **Commit** ```bash git add app/api/static/capture.html git commit -m "feat: browser capture page — MediaRecorder WebSocket streaming + mute" git push origin master ``` --- ## Task 12: Dashboard SPA **Files:** `app/api/static/index.html`, `app/api/static/app.js` - [ ] **Create `app/api/static/index.html`** ```html LinkedStorm
``` - [ ] **Create `app/api/static/app.js`** ```javascript // --- Utilities --- const COLORS = ['#60a5fa','#a78bfa','#f472b6','#34d399','#fb923c','#38bdf8','#e879f9','#a3e635']; const speakerColor = (id) => COLORS[(id || 0) % COLORS.length]; const speakerInitial = (name) => (name || '?')[0].toUpperCase(); const fmtTime = (iso) => new Date(iso).toLocaleTimeString([], {hour:'2-digit',minute:'2-digit',second:'2-digit'}); const fmtDate = (iso) => new Date(iso).toLocaleDateString([], {weekday:'long',month:'long',day:'numeric',year:'numeric'}); const today = () => new Date().toISOString().slice(0,10); let _currentDate = today(); function showPanel(name, el) { document.querySelectorAll('.panel').forEach(p => p.classList.remove('active')); document.querySelectorAll('.nav-item').forEach(n => n.classList.remove('active')); document.getElementById('panel-' + name).classList.add('active'); if (el) el.classList.add('active'); if (name === 'log') loadLog(); if (name === 'search') renderSearch(); if (name === 'speakers') loadSpeakers(); if (name === 'rag') renderRag(); if (name === 'rooms') loadRooms(); } // --- Daily Log --- async function loadLog(date) { if (date) _currentDate = date; const panel = document.getElementById('panel-log'); const utterances = await fetch(`/api/utterances?date=${_currentDate}`).then(r => r.json()); const nav = `
${fmtDate(_currentDate + 'T12:00:00')}
`; if (!utterances.length) { panel.innerHTML = nav + '
No utterances recorded for this day.
'; return; } // Group by hour const byHour = {}; for (const u of utterances.reverse()) { const h = new Date(u.start_time).getHours(); const label = new Date(u.start_time).toLocaleTimeString([], {hour:'2-digit',minute:'2-digit'}); const key = label.slice(0,5); if (!byHour[key]) byHour[key] = []; byHour[key].push(u); } let html = nav; for (const [timeKey, utts] of Object.entries(byHour)) { html += `
${timeKey}
`; for (const u of utts) { const color = speakerColor(u.speaker_id); const name = u.speaker_name || 'Unknown'; const isUnknown = !u.speaker_id; const badge = u.match_status === 'unknown' ? 'unknown' : u.match_status === 'ambiguous' ? 'ambiguous' : ''; const actions = isUnknown ? `
` : ''; html += `
${speakerInitial(name)}
${name}${badge} ${u.room_name} ${fmtTime(u.start_time)}
${escHtml(u.transcript)}
${actions}
`; } html += '
'; } panel.innerHTML = html; } function shiftDate(days) { const d = new Date(_currentDate + 'T12:00:00'); d.setDate(d.getDate() + days); return d.toISOString().slice(0,10); } async function playClip(id) { const a = document.createElement('audio'); a.src = `/api/utterances/${id}/clip`; a.controls = true; a.autoplay = true; document.body.appendChild(a); a.onended = () => a.remove(); } async function tagSpeaker(utteranceId) { const speakers = await fetch('/api/speakers').then(r => r.json()); const name = prompt('Speaker name (existing or new):' + speakers.map(s => '\n• ' + s.name).join('')); if (!name) return; const existing = speakers.find(s => s.name.toLowerCase() === name.toLowerCase()); const body = existing ? { speaker_id: existing.id } : { name: name.trim(), create_new: true }; await fetch(`/api/utterances/${utteranceId}/tag`, { method: 'POST', headers: {'Content-Type':'application/json'}, body: JSON.stringify(body) }); loadLog(); } async function dismiss(id) { await fetch(`/api/utterances/${id}/dismiss`, { method: 'POST' }); loadLog(); } // --- Search --- function renderSearch() { const panel = document.getElementById('panel-search'); panel.innerHTML = `

Search Transcripts

`; } async function doSearch() { const q = document.getElementById('searchQ').value.trim(); if (!q) return; const results = await fetch(`/api/search?q=${encodeURIComponent(q)}`).then(r => r.json()); const container = document.getElementById('searchResults'); if (!results.length) { container.innerHTML = '
No results.
'; return; } const highlight = (text) => text.replace(new RegExp(`(${q})`, 'gi'), '$1'); container.innerHTML = results.map(r => `
${r.speaker_name || 'Unknown'} · ${r.room_name} · ${fmtDate(r.start_time)} ${fmtTime(r.start_time)}
${highlight(escHtml(r.transcript))}
`).join(''); } // --- Speakers --- async function loadSpeakers() { const panel = document.getElementById('panel-speakers'); const [speakers, unknowns] = await Promise.all([ fetch('/api/speakers').then(r => r.json()), fetch('/api/utterances?match_status=unreviewed').then(r => r.json()), ]); let html = ''; if (unknowns.length) { html += `
⚠ ${unknowns.length} utterances need speaker tags
`; } html += `
` + speakers.map(s => { const color = speakerColor(s.id); return `
${speakerInitial(s.name)}
${escHtml(s.name)}
${s.utterance_count} utterances
`; }).join('') + `
+ Add speaker
`; panel.innerHTML = `

Speakers

` + html; } async function createSpeaker() { const name = prompt('Speaker name:'); if (!name) return; await fetch('/api/speakers', { method: 'POST', headers: {'Content-Type':'application/json'}, body: JSON.stringify({name}) }); loadSpeakers(); } // --- Ask AI --- const _ragHistory = []; function renderRag() { const panel = document.getElementById('panel-rag'); panel.innerHTML = `

Ask AI

`; renderRagMessages(); } function renderRagMessages() { const el = document.getElementById('ragMessages'); if (!el) return; if (!_ragHistory.length) { el.innerHTML = '
Ask a question about any conversation.
'; return; } el.innerHTML = _ragHistory.map(m => `
${m.role === 'user' ? 'You' : 'LinkedStorm AI'}
${escHtml(m.content)}${m.sources ? `
Sources: ${m.sources}
` : ''}
`).join(''); el.scrollTop = el.scrollHeight; } async function askRag() { const q = document.getElementById('ragQ').value.trim(); if (!q) return; document.getElementById('ragQ').value = ''; _ragHistory.push({ role: 'user', content: q }); renderRagMessages(); const resp = await fetch('/api/ask', { method: 'POST', headers: {'Content-Type':'application/json'}, body: JSON.stringify({ question: q, date: _currentDate }), }).then(r => r.json()); _ragHistory.push({ role: 'ai', content: resp.answer, sources: `${resp.context_count} transcript segments` }); renderRagMessages(); } // --- Rooms --- async function loadRooms() { const panel = document.getElementById('panel-rooms'); const rooms = await fetch('/api/rooms').then(r => r.json()); let html = `

Capture Rooms

`; for (const r of rooms) { const cls = r.is_active ? (r.muted ? 'muted' : 'active') : ''; const dotColor = r.is_active ? (r.muted ? '#ef4444' : '#4ade80') : '#475569'; const statusText = r.is_active ? (r.muted ? 'Muted' : 'Streaming') : 'Disconnected'; html += `
${escHtml(r.name)}
${statusText}
${r.last_seen ? `
Last: ${fmtTime(r.last_seen)}
` : ''} Open capture page ↗
`; } html += `
`; panel.innerHTML = html; } async function toggleMute(roomId) { const rooms = await fetch('/api/rooms').then(r => r.json()); const room = rooms.find(r => r.id === roomId); const action = room?.muted ? 'unmute' : 'mute'; await fetch(`/api/rooms/${roomId}/${action}`, { method: 'POST' }); loadRooms(); } function escHtml(s) { return String(s).replace(/&/g,'&').replace(//g,'>'); } // Initial load loadLog(); setInterval(loadLog, 30000); // Refresh every 30s ``` - [ ] **Commit** ```bash git add app/api/static/ git commit -m "feat: dashboard SPA — 5-tab log/search/speakers/RAG/rooms UI" git push origin master ``` --- ## Task 13: Dockerfiles **Files:** `pipeline/Dockerfile`, `api/Dockerfile`, `cron/Dockerfile` - [ ] **Write `pipeline/Dockerfile`** ```dockerfile FROM python:3.12-slim RUN apt-get update && apt-get install -y --no-install-recommends \ ffmpeg libsndfile1 && rm -rf /var/lib/apt/lists/* WORKDIR /app COPY pyproject.toml . RUN pip install --no-cache-dir torch --index-url https://download.pytorch.org/whl/cpu \ && pip install --no-cache-dir -e "." \ && pip uninstall -y torch && pip install --no-cache-dir torch --index-url https://download.pytorch.org/whl/cpu COPY app/ app/ # Pre-download Silero VAD and resemblyzer models into the image layer RUN python -c "import torch; torch.hub.load('snakers4/silero-vad','silero_vad',trust_repo=True)" \ && python -c "from resemblyzer import VoiceEncoder; VoiceEncoder()" ENV PYTHONUNBUFFERED=1 CMD ["python", "-m", "app.pipeline.main"] ``` - [ ] **Write `api/Dockerfile`** ```dockerfile FROM python:3.12-slim WORKDIR /app COPY pyproject.toml . RUN pip install --no-cache-dir fastapi uvicorn[standard] pydantic-settings httpx livekit-api COPY app/ app/ ENV PYTHONUNBUFFERED=1 CMD ["uvicorn", "app.api.main:app", "--host", "0.0.0.0", "--port", "8310", "--factory"] ``` Update `app/api/main.py` to add an `app` factory callable at module level: ```python # At the bottom of app/api/main.py, add: import os _app_instance = None def app_factory(): global _app_instance if _app_instance is None: from app.shared.config import get_settings _app_instance = create_app(get_settings()) return _app_instance # Uvicorn --factory needs this to be importable as app.api.main:app # Use: uvicorn app.api.main:app_factory --factory ``` Change CMD in api/Dockerfile: ```dockerfile CMD ["uvicorn", "app.api.main:app_factory", "--factory", "--host", "0.0.0.0", "--port", "8310"] ``` - [ ] **Write `cron/Dockerfile`** ```dockerfile FROM python:3.12-slim RUN apt-get update && apt-get install -y --no-install-recommends \ libsndfile1 && rm -rf /var/lib/apt/lists/* WORKDIR /app COPY pyproject.toml . RUN pip install --no-cache-dir torch --index-url https://download.pytorch.org/whl/cpu \ && pip install --no-cache-dir -e "." COPY app/ app/ # Pre-download resemblyzer model RUN python -c "from resemblyzer import VoiceEncoder; VoiceEncoder()" ENV PYTHONUNBUFFERED=1 CMD ["python", "-m", "app.cron.main"] ``` - [ ] **Commit** ```bash git add pipeline/ api/ cron/ git commit -m "feat: Dockerfiles for pipeline, api, and cron containers" git push origin master ``` --- ## Task 14: Docker Compose + Traefik + deploy **Files:** - `compose/hf-pdocker-01/linkedstorm/docker-compose.yml` (in docker repo) - `compose/hf-pdocker-01/linkedstorm/.env.example` (in docker repo) - `stacks/traefik/dynamic/linkedstorm.yml` (in docker repo) - [ ] **Create `compose/hf-pdocker-01/linkedstorm/docker-compose.yml`** (in the docker repo at `/mnt/c/Users/Jeremy/projects/docker/`) ```yaml services: linkedstorm-pipeline: build: context: /opt/linkedstorm dockerfile: pipeline/Dockerfile container_name: linkedstorm-pipeline restart: unless-stopped ports: - "8300:8300" volumes: - /mnt/docker-data/configs/linkedstorm:/data environment: - DB_PATH=/data/linkedstorm.db - CLIPS_DIR=/data/clips - SPEACHES_URL=${SPEACHES_URL} - OLLAMA_URL=${OLLAMA_URL} - LIVEKIT_API_KEY=${LIVEKIT_API_KEY} - LIVEKIT_API_SECRET=${LIVEKIT_API_SECRET} networks: - linkedstorm-net labels: - com.centurylinklabs.watchtower.enable=false linkedstorm-api: build: context: /opt/linkedstorm dockerfile: api/Dockerfile container_name: linkedstorm-api restart: unless-stopped ports: - "8310:8310" volumes: - /mnt/docker-data/configs/linkedstorm:/data environment: - DB_PATH=/data/linkedstorm.db - CLIPS_DIR=/data/clips - PIPELINE_INTERNAL_URL=http://linkedstorm-pipeline:8300 - OLLAMA_URL=${OLLAMA_URL} - OLLAMA_MODEL=${OLLAMA_MODEL:-qwen2.5:14b} - LIVEKIT_API_KEY=${LIVEKIT_API_KEY} - LIVEKIT_API_SECRET=${LIVEKIT_API_SECRET} networks: - linkedstorm-net depends_on: - linkedstorm-pipeline labels: - com.centurylinklabs.watchtower.enable=false linkedstorm-cron: build: context: /opt/linkedstorm dockerfile: cron/Dockerfile container_name: linkedstorm-cron restart: unless-stopped volumes: - /mnt/docker-data/configs/linkedstorm:/data environment: - DB_PATH=/data/linkedstorm.db - CLIPS_DIR=/data/clips - OLLAMA_URL=${OLLAMA_URL} - OLLAMA_MODEL=${OLLAMA_MODEL:-qwen2.5:14b} - CLIP_RETENTION_DAYS=${CLIP_RETENTION_DAYS:-30} networks: - linkedstorm-net labels: - com.centurylinklabs.watchtower.enable=false networks: linkedstorm-net: driver: bridge ``` - [ ] **Create `compose/hf-pdocker-01/linkedstorm/.env.example`** ```bash SPEACHES_URL=http://192.168.86.150:8000 OLLAMA_URL=http://192.168.86.150:11434 OLLAMA_MODEL=qwen2.5:14b LIVEKIT_API_KEY=your-livekit-api-key LIVEKIT_API_SECRET=your-livekit-api-secret CLIP_RETENTION_DAYS=30 ``` - [ ] **Create `stacks/traefik/dynamic/linkedstorm.yml`** ```yaml http: routers: linkedstorm: rule: "Host(`linkedstorm.hobbs.farm`)" entryPoints: - websecure service: linkedstorm tls: certResolver: letsencrypt services: linkedstorm: loadBalancer: servers: - url: "http://192.168.86.192:8310" ``` - [ ] **Commit to docker repo** ```bash cd /mnt/c/Users/Jeremy/projects/docker git add compose/hf-pdocker-01/linkedstorm/ stacks/traefik/dynamic/linkedstorm.yml git commit -m "feat: linkedstorm docker-compose and Traefik route" git push origin master ``` --- ## Task 15: Deploy and verify on hf-pdocker-01 - [ ] **SSH to hf-pdocker-01** ```bash ssh glyptic@192.168.86.192 ``` - [ ] **Clone linkedstorm repo and create data directory** ```bash sudo git clone http://192.168.86.100:3003/jeremy/linkedstorm /opt/linkedstorm sudo mkdir -p /mnt/docker-data/configs/linkedstorm/clips sudo chown -R 1000:1000 /opt/linkedstorm ``` - [ ] **Copy compose files from NAS** ```bash # Compose files are already synced to NAS via gitea-backup cd /mnt/docker-data/compose/hf-pdocker-01/linkedstorm # OR pull from docker repo: sudo git clone http://192.168.86.100:3003/jeremy/hobbs-iac /opt/hobbs-iac # then copy compose dir to /mnt/docker-data/compose/hf-pdocker-01/linkedstorm/ ``` - [ ] **Create `.env` from example and fill in real credentials** ```bash cd /mnt/docker-data/compose/hf-pdocker-01/linkedstorm cp .env.example .env # Edit .env: add real LIVEKIT_API_KEY and LIVEKIT_API_SECRET from Vaultwarden nano .env ``` - [ ] **Build and start all 3 containers** ```bash cd /mnt/docker-data/compose/hf-pdocker-01/linkedstorm docker compose up -d --build ``` Expected: 3 containers start: `linkedstorm-pipeline`, `linkedstorm-api`, `linkedstorm-cron` - [ ] **Verify 1 — all containers healthy** ```bash docker ps --format "table {{.Names}}\t{{.Status}}" | grep linkedstorm ``` Expected: ``` linkedstorm-pipeline Up X minutes linkedstorm-api Up X minutes linkedstorm-cron Up X minutes ``` - [ ] **Verify 2 — health endpoint responds** ```bash curl -s http://localhost:8310/health # Note: add /health to api/main.py curl -s http://localhost:8300/health ``` Expected: `{"status":"ok"}` from both Note: Add health endpoint to `app/api/main.py` before deploying: ```python @app.get("/health") async def health(): return {"status": "ok"} ``` - [ ] **Verify 3 — open capture page and speak** On any device on the LAN, open: `http://192.168.86.192:8310/capture?room=Kitchen` Allow microphone access, speak a sentence. Check dashboard at `http://192.168.86.192:8310`: ```bash # Or check directly: curl -s "http://localhost:8310/api/utterances" | python3 -m json.tool | head -40 ``` Expected: At least one utterance with a transcript appears within a few seconds. - [ ] **Verify 4 — speaker labeling** In the dashboard Speakers tab, click "Review now", tag the unknown utterance as your name. Speak again. Confirm the next utterance resolves to your name automatically. - [ ] **Verify 5 — search** Search for a word you spoke. Confirm it appears with keyword highlighting. - [ ] **Verify 6 — Ask AI** Ask "What did we talk about today?" in the Ask AI tab. Confirm Ollama returns a coherent response. - [ ] **Verify 7 — mute** Click mute on the capture page. Speak. Confirm no new utterances appear in the log. Unmute and speak again — utterance appears. - [ ] **Verify 8 — Traefik route** ```bash curl -s https://linkedstorm.hobbs.farm/health ``` Expected: `{"status":"ok"}` (may take a minute for Let's Encrypt cert) - [ ] **Verify 9 — manual cron test** ```bash docker exec linkedstorm-cron python -c " from app.shared.config import get_settings from app.shared.database import get_db from app.cron.daily_summary import generate_daily_summary import datetime s = get_settings() db = get_db(s) generate_daily_summary(datetime.date.today().isoformat(), s, db) print('Done') " ``` Expected: `Done` (if Ollama reachable, a row appears in daily_summaries) - [ ] **Verify 10 — add Prometheus probe** (in docker repo) ```bash cd /mnt/c/Users/Jeremy/projects/docker ``` Edit `compose/hf-pdocker-01/monitoring/prometheus.yml` — add to the `http-pdocker01` job's `static_configs` targets: ```yaml - http://192.168.86.192:8310/health ``` Then hot-reload Prometheus: ```bash ssh glyptic@192.168.86.192 "curl -s -X POST http://localhost:9095/-/reload" ``` - [ ] **Final commit to docker repo** ```bash cd /mnt/c/Users/Jeremy/projects/docker git add compose/hf-pdocker-01/monitoring/ git commit -m "monitoring: add Prometheus probe for linkedstorm-api" git push origin master ``` --- ## Self-Review Notes **Spec coverage check:** - ✅ 3 containers: pipeline (Task 7), api (Tasks 8-9, 12-13), cron (Task 10) - ✅ Browser capture page with WebSocket + mute (Task 11) - ✅ Silero VAD (Task 5) - ✅ Whisper STT (Task 6, processor.py) - ✅ resemblyzer voice embedding + cosine similarity (Task 4) - ✅ Three-tier matching (Task 4, recognition.py) - ✅ SQLite schema + FTS5 (Task 3) - ✅ Speaker labeling workflow — tag endpoint (Task 8, utterances.py) - ✅ Daily log timeline (Task 12, app.js) - ✅ Full-text search (Task 9, search.py + Task 12) - ✅ RAG via Ollama (Task 9, rag.py + Task 12) - ✅ Room status + mute relay (Task 9, rooms.py + Task 12) - ✅ Nightly daily summary (Task 10) - ✅ Unknown speaker clustering (Task 10) - ✅ 30-day clip purge (Task 10) - ✅ Docker Compose (Task 14) - ✅ Traefik route (Task 14) - ✅ Prometheus probe (Task 15) - ⚠️ LiveKit publishing from pipeline — simplified: capture page connects directly to LiveKit via JS SDK for live monitoring (future work, out of scope for v1). Replay uses stored WAV clips via /api/utterances/{id}/clip. **Type consistency:** `find_best_speaker_match` returns `tuple[str, int | None, float | None]` — used consistently in recognition.py (Task 4) and processor.py (Task 6). `pack_embedding`/`unpack_embedding` used in recognition.py, processor.py, and utterances.py — same import path `app.pipeline.recognition` throughout.