Files
hobbs-iac/docs/superpowers/plans/2026-05-28-linkedstorm-implementation.md
jeremy 27f6d95bf6 chore: sync accumulated infra changes and docs
Removals: openwebui and netclaw compose + Traefik routes (both decommissioned).
Updates: litellm, n8n, vaultwarden, ai-agents, media, traefik routes, CLAUDE.md
(nodes/ports/known-issues), agent definitions.
Additions: Rongorongo tree, LinkedStorm plan, rename-bart spec, vllm brief,
tls-options and security-headers Traefik config.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-11 11:19:21 -05:00

3188 lines
102 KiB
Markdown

# 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
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>LinkedStorm Capture</title>
<style>
body { font-family: system-ui, sans-serif; background: #0f172a; color: #e2e8f0;
display: flex; align-items: center; justify-content: center; min-height: 100vh; margin: 0; }
.card { background: #1e293b; border-radius: 16px; padding: 32px; text-align: center;
min-width: 280px; box-shadow: 0 4px 32px rgba(0,0,0,0.4); }
h1 { font-size: 1.4em; margin: 0 0 8px; color: #f1f5f9; }
.room-label { font-size: 1em; color: #94a3b8; margin-bottom: 24px; }
.status { display: flex; align-items: center; justify-content: center; gap: 10px;
font-size: 0.9em; margin-bottom: 24px; }
.dot { width: 12px; height: 12px; border-radius: 50%; }
.dot.green { background: #4ade80; box-shadow: 0 0 8px #4ade80; animation: pulse 2s infinite; }
.dot.red { background: #ef4444; }
.dot.gray { background: #475569; }
@keyframes pulse { 0%,100%{opacity:1} 50%{opacity:0.5} }
button { padding: 12px 28px; border-radius: 8px; border: none; font-size: 1em;
cursor: pointer; transition: all 0.15s; }
#muteBtn { background: #334155; color: #e2e8f0; }
#muteBtn:hover { background: #475569; }
#muteBtn.muted { background: #7f1d1d; color: #fca5a5; }
.room-input { background: #334155; border: 1px solid #475569; border-radius: 8px;
padding: 8px 12px; color: #e2e8f0; font-size: 0.9em; width: 100%;
box-sizing: border-box; margin-bottom: 16px; }
</style>
</head>
<body>
<div class="card">
<h1>⚡ LinkedStorm</h1>
<div id="setup" style="display:none">
<p style="color:#94a3b8;font-size:0.85em">Enter room name to start capturing</p>
<input id="roomInput" class="room-input" placeholder="e.g. Kitchen" />
<button onclick="startCapture()" style="background:#4ade80;color:#052e16">Start Capturing</button>
</div>
<div id="capture" style="display:none">
<div class="room-label">📍 <span id="roomDisplay"></span></div>
<div class="status">
<div class="dot" id="statusDot"></div>
<span id="statusText">Connecting…</span>
</div>
<button id="muteBtn" onclick="toggleMute()">🔇 Mute</button>
</div>
</div>
<script>
const params = new URLSearchParams(location.search);
let roomName = params.get('room') || localStorage.getItem('linkedstorm_room');
let ws = null, mediaRecorder = null, muted = false, reconnectTimer = null;
function init() {
if (roomName) {
document.getElementById('setup').style.display = 'none';
document.getElementById('capture').style.display = 'block';
document.getElementById('roomDisplay').textContent = roomName;
connect();
} else {
document.getElementById('setup').style.display = 'block';
}
}
function startCapture() {
roomName = document.getElementById('roomInput').value.trim();
if (!roomName) return;
localStorage.setItem('linkedstorm_room', roomName);
document.getElementById('setup').style.display = 'none';
document.getElementById('capture').style.display = 'block';
document.getElementById('roomDisplay').textContent = roomName;
connect();
}
function setStatus(state, text) {
const dot = document.getElementById('statusDot');
const label = document.getElementById('statusText');
dot.className = 'dot ' + state;
label.textContent = text;
}
async function connect() {
setStatus('gray', 'Connecting…');
try {
const stream = await navigator.mediaDevices.getUserMedia({
audio: { channelCount: 1, sampleRate: 16000 }
});
const wsProto = location.protocol === 'https:' ? 'wss:' : 'ws:';
ws = new WebSocket(`${wsProto}//${location.host.replace(':8310','').replace(':80','')}:8300/ws/${encodeURIComponent(roomName)}`);
ws.binaryType = 'arraybuffer';
ws.onopen = () => {
setStatus('green', muted ? 'Muted' : 'Streaming');
startRecorder(stream);
};
ws.onclose = () => {
setStatus('gray', 'Reconnecting…');
reconnectTimer = setTimeout(connect, 3000);
};
ws.onerror = () => ws.close();
} catch (e) {
setStatus('red', 'Mic denied — allow microphone access');
}
}
function startRecorder(stream) {
mediaRecorder = new MediaRecorder(stream, { mimeType: 'audio/webm;codecs=pcm' });
mediaRecorder.ondataavailable = e => {
if (ws && ws.readyState === WebSocket.OPEN && !muted && e.data.size > 0) {
ws.send(e.data);
}
};
mediaRecorder.start(250); // 250ms chunks
}
function toggleMute() {
muted = !muted;
const btn = document.getElementById('muteBtn');
if (muted) {
btn.textContent = '🎤 Unmute';
btn.classList.add('muted');
setStatus('red', 'Muted');
if (ws && ws.readyState === WebSocket.OPEN)
ws.send(JSON.stringify({ type: 'mute' }));
} else {
btn.textContent = '🔇 Mute';
btn.classList.remove('muted');
setStatus('green', 'Streaming');
if (ws && ws.readyState === WebSocket.OPEN)
ws.send(JSON.stringify({ type: 'unmute' }));
}
}
init();
</script>
</body>
</html>
```
- [ ] **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
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>LinkedStorm</title>
<style>
*{box-sizing:border-box;margin:0;padding:0}
body{font-family:system-ui,sans-serif;background:#0f172a;color:#e2e8f0;display:flex;height:100vh;overflow:hidden}
.sidebar{width:180px;flex-shrink:0;background:#1e293b;display:flex;flex-direction:column;border-right:1px solid #334155}
.brand{padding:16px;font-weight:700;font-size:1em;color:#4ade80;border-bottom:1px solid #334155}
.nav-item{padding:10px 16px;font-size:0.85em;color:#94a3b8;cursor:pointer;display:flex;align-items:center;gap:8px;transition:all 0.1s}
.nav-item:hover{background:#334155;color:#e2e8f0}
.nav-item.active{background:rgba(74,222,128,0.1);color:#4ade80;border-left:3px solid #4ade80}
.main{flex:1;overflow:auto;background:#0f172a}
.panel{display:none;padding:20px;height:100%}
.panel.active{display:block}
.toolbar{display:flex;gap:8px;margin-bottom:16px;align-items:center;flex-wrap:wrap}
input,select{background:#1e293b;border:1px solid #334155;border-radius:6px;padding:7px 10px;color:#e2e8f0;font-size:0.85em}
input:focus,select:focus{outline:none;border-color:#4ade80}
button{padding:7px 14px;border-radius:6px;border:none;cursor:pointer;font-size:0.85em;transition:all 0.15s}
.btn-primary{background:#4ade80;color:#052e16;font-weight:600}
.btn-secondary{background:#334155;color:#e2e8f0}
.btn-small{padding:3px 8px;font-size:0.72em}
.time-block{margin-bottom:16px}
.time-label{font-size:0.7em;text-transform:uppercase;letter-spacing:0.06em;color:#475569;margin-bottom:6px;display:flex;align-items:center;gap:8px}
.time-label::after{content:'';flex:1;height:1px;background:#1e293b}
.utterance{display:flex;gap:10px;margin-bottom:6px;align-items:flex-start}
.avatar{width:28px;height:28px;border-radius:50%;display:flex;align-items:center;justify-content:center;font-size:0.75em;font-weight:700;flex-shrink:0;margin-top:2px}
.utt-body{flex:1;background:#1e293b;border-radius:8px;padding:8px 12px}
.utt-meta{display:flex;gap:8px;align-items:center;margin-bottom:3px;font-size:0.72em}
.utt-name{font-weight:600}
.utt-room{color:#64748b}
.utt-time{color:#475569;margin-left:auto}
.utt-text{font-size:0.82em;color:#cbd5e1;line-height:1.45}
.utt-actions{display:flex;gap:4px;margin-top:5px}
.badge{font-size:0.62em;padding:1px 5px;border-radius:4px;margin-left:4px}
.badge-unknown{background:rgba(239,68,68,0.2);color:#f87171}
.badge-ambiguous{background:rgba(245,158,11,0.2);color:#fbbf24}
.search-result{background:#1e293b;border-radius:8px;padding:10px 12px;margin-bottom:8px;border-left:3px solid #334155}
.search-result:hover{border-left-color:#4ade80}
.sr-meta{font-size:0.7em;color:#64748b;margin-bottom:3px}
.sr-text{font-size:0.82em;color:#cbd5e1}
mark{background:rgba(74,222,128,0.2);color:#4ade80;border-radius:2px;padding:0 2px}
.speaker-grid{display:grid;grid-template-columns:repeat(auto-fill,minmax(150px,1fr));gap:10px}
.speaker-card{background:#1e293b;border:1px solid #334155;border-radius:10px;padding:14px;text-align:center;cursor:pointer}
.speaker-card:hover{border-color:#4ade80}
.sp-avatar{width:44px;height:44px;border-radius:50%;margin:0 auto 8px;display:flex;align-items:center;justify-content:center;font-weight:700;font-size:1.1em}
.sp-name{font-size:0.85em;font-weight:600}
.sp-count{font-size:0.72em;color:#64748b;margin-top:2px}
.room-grid{display:grid;grid-template-columns:repeat(auto-fill,minmax(170px,1fr));gap:10px}
.room-card{background:#1e293b;border:1px solid #334155;border-radius:10px;padding:14px}
.room-card.active{border-color:#4ade80}
.room-card.muted{border-color:#ef4444}
.room-name{font-size:0.9em;font-weight:600;margin-bottom:6px}
.status-row{display:flex;align-items:center;gap:6px;font-size:0.75em}
.sdot{width:8px;height:8px;border-radius:50%}
.rag-msg{margin-bottom:12px}
.rag-bubble{border-radius:10px;padding:10px 14px;font-size:0.85em;line-height:1.5;color:#cbd5e1}
.rag-user .rag-bubble{background:rgba(74,222,128,0.08);border:1px solid rgba(74,222,128,0.2);margin-left:15%}
.rag-ai .rag-bubble{background:#1e293b;border:1px solid #334155;margin-right:15%}
.rag-label{font-size:0.68em;color:#64748b;margin-bottom:3px;padding:0 4px}
.rag-source{font-size:0.7em;color:#475569;margin-top:6px;padding-top:6px;border-top:1px solid #334155;font-style:italic}
.review-banner{background:rgba(245,158,11,0.08);border:1px solid rgba(245,158,11,0.25);border-radius:8px;padding:10px 14px;margin-bottom:14px;display:flex;justify-content:space-between;align-items:center;font-size:0.82em;color:#fbbf24}
h2{font-size:1em;font-weight:600;margin-bottom:16px;color:#f1f5f9}
.empty{color:#475569;font-size:0.85em;padding:20px 0}
</style>
</head>
<body>
<div class="sidebar">
<div class="brand">⚡ LinkedStorm</div>
<div class="nav-item active" onclick="showPanel('log',this)">📋 Daily Log</div>
<div class="nav-item" onclick="showPanel('search',this)">🔍 Search</div>
<div class="nav-item" onclick="showPanel('speakers',this)">🧑 Speakers</div>
<div class="nav-item" onclick="showPanel('rag',this)">🤖 Ask AI</div>
<div class="nav-item" onclick="showPanel('rooms',this)">📍 Rooms</div>
</div>
<div class="main">
<div id="panel-log" class="panel active"></div>
<div id="panel-search" class="panel"></div>
<div id="panel-speakers" class="panel"></div>
<div id="panel-rag" class="panel"></div>
<div id="panel-rooms" class="panel"></div>
</div>
<script src="/app.js"></script>
</body>
</html>
```
- [ ] **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 = `<div class="toolbar">
<button class="btn-secondary btn-small" onclick="loadLog(shiftDate(-1))">← Prev</button>
<span style="font-size:0.85em;font-weight:600">${fmtDate(_currentDate + 'T12:00:00')}</span>
<button class="btn-secondary btn-small" onclick="loadLog(shiftDate(1))">Next →</button>
<button class="btn-secondary btn-small" onclick="loadLog('${today()}')">Today</button>
</div>`;
if (!utterances.length) {
panel.innerHTML = nav + '<div class="empty">No utterances recorded for this day.</div>';
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 += `<div class="time-block"><div class="time-label">${timeKey}</div>`;
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' ? '<span class="badge badge-unknown">unknown</span>'
: u.match_status === 'ambiguous' ? '<span class="badge badge-ambiguous">ambiguous</span>' : '';
const actions = isUnknown ? `<div class="utt-actions">
<button class="btn-small btn-secondary" onclick="playClip(${u.id})">▶ Play</button>
<button class="btn-small btn-secondary" onclick="tagSpeaker(${u.id})">🏷 Tag</button>
<button class="btn-small btn-secondary" onclick="dismiss(${u.id})">✕</button>
</div>` : '';
html += `<div class="utterance">
<div class="avatar" style="background:${color}22;color:${color}">${speakerInitial(name)}</div>
<div class="utt-body">
<div class="utt-meta">
<span class="utt-name" style="color:${color}">${name}${badge}</span>
<span class="utt-room">${u.room_name}</span>
<span class="utt-time">${fmtTime(u.start_time)}</span>
</div>
<div class="utt-text">${escHtml(u.transcript)}</div>
${actions}
</div>
</div>`;
}
html += '</div>';
}
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 = `<h2>Search Transcripts</h2>
<div class="toolbar">
<input id="searchQ" placeholder="Search…" style="flex:1" onkeydown="if(event.key==='Enter')doSearch()">
<button class="btn-primary" onclick="doSearch()">Search</button>
</div>
<div id="searchResults"></div>`;
}
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 = '<div class="empty">No results.</div>'; return; }
const highlight = (text) => text.replace(new RegExp(`(${q})`, 'gi'), '<mark>$1</mark>');
container.innerHTML = results.map(r => `<div class="search-result">
<div class="sr-meta">${r.speaker_name || 'Unknown'} · ${r.room_name} · ${fmtDate(r.start_time)} ${fmtTime(r.start_time)}</div>
<div class="sr-text">${highlight(escHtml(r.transcript))}</div>
</div>`).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 += `<div class="review-banner">⚠ ${unknowns.length} utterances need speaker tags
<button class="btn-small btn-secondary" onclick="showPanel('log',null);loadLog('${today()}')">Review now</button>
</div>`;
}
html += `<div class="speaker-grid">` + speakers.map(s => {
const color = speakerColor(s.id);
return `<div class="speaker-card">
<div class="sp-avatar" style="background:${color}22;color:${color}">${speakerInitial(s.name)}</div>
<div class="sp-name">${escHtml(s.name)}</div>
<div class="sp-count">${s.utterance_count} utterances</div>
</div>`;
}).join('') + `<div class="speaker-card" onclick="createSpeaker()" style="border-style:dashed;align-items:center;justify-content:center;display:flex;min-height:100px">
<div style="color:#475569;font-size:0.8em">+ Add speaker</div>
</div></div>`;
panel.innerHTML = `<h2>Speakers</h2>` + 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 = `<h2>Ask AI</h2>
<div id="ragMessages" style="margin-bottom:16px;min-height:200px"></div>
<div class="toolbar">
<input id="ragQ" placeholder="Ask about any conversation…" style="flex:1" onkeydown="if(event.key==='Enter')askRag()">
<button class="btn-primary" onclick="askRag()">Ask</button>
</div>`;
renderRagMessages();
}
function renderRagMessages() {
const el = document.getElementById('ragMessages');
if (!el) return;
if (!_ragHistory.length) { el.innerHTML = '<div class="empty">Ask a question about any conversation.</div>'; return; }
el.innerHTML = _ragHistory.map(m => `<div class="rag-msg rag-${m.role}">
<div class="rag-label">${m.role === 'user' ? 'You' : 'LinkedStorm AI'}</div>
<div class="rag-bubble">${escHtml(m.content)}${m.sources ? `<div class="rag-source">Sources: ${m.sources}</div>` : ''}</div>
</div>`).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 = `<h2>Capture Rooms</h2><div class="room-grid">`;
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 += `<div class="room-card ${cls}">
<div class="room-name">${escHtml(r.name)}</div>
<div class="status-row">
<div class="sdot" style="background:${dotColor}${r.is_active&&!r.muted?';box-shadow:0 0 6px '+dotColor:''}"></div>
<span style="color:${dotColor}">${statusText}</span>
</div>
${r.last_seen ? `<div style="font-size:0.7em;color:#475569;margin-top:6px">Last: ${fmtTime(r.last_seen)}</div>` : ''}
<button class="btn-small btn-secondary" style="margin-top:8px;width:100%"
onclick="toggleMute(${r.id}, '${r.name}')">
${r.muted ? '🎤 Unmute' : '🔇 Mute'}
</button>
<a href="/capture?room=${encodeURIComponent(r.name)}" target="_blank"
style="display:block;margin-top:6px;font-size:0.7em;color:#64748b;text-align:center">
Open capture page ↗
</a>
</div>`;
}
html += `</div>`;
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,'&amp;').replace(/</g,'&lt;').replace(/>/g,'&gt;');
}
// 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.