feat: reduce Hatchet payload size by removing words from topic chunk workflows

Remove ~6.5MB of redundant Word data from Hatchet task boundaries:
- Remove words from TopicChunkInput/TopicChunkResult (child workflow I/O)
- detect_topics maps words from local chunks by chunk_index instead
- TopicsResult carries empty transcript words (persisted to DB already)
- extract_subjects refetches topics from DB instead of task output
- Clear topics at detect_topics start for retry idempotency

docker-compose.standalone.yml

@@ -63,14 +63,16 @@ services:
   server:
     environment:
       TRANSCRIPT_BACKEND: modal
-      TRANSCRIPT_URL: http://cpu:8000
+      TRANSCRIPT_URL: http://localhost:8100
+      TRANSCRIPT_MODAL_API_KEY: local
       DIARIZATION_BACKEND: modal
-      DIARIZATION_URL: http://cpu:8000
+      DIARIZATION_URL: http://localhost:8100
 
   worker:
     environment:
       TRANSCRIPT_BACKEND: modal
       TRANSCRIPT_URL: http://cpu:8000
+      TRANSCRIPT_MODAL_API_KEY: local
       DIARIZATION_BACKEND: modal
       DIARIZATION_URL: http://cpu:8000
 
@@ -78,6 +80,8 @@ services:
     build:
       context: ./gpu/self_hosted
       dockerfile: Dockerfile.cpu
+    ports:
+      - "8100:8000"
     volumes:
       - gpu_cache:/root/.cache
     restart: unless-stopped

docker-compose.yml

@@ -2,8 +2,7 @@ services:
   server:
     build:
       context: server
-    ports:
-      - 1250:1250
+    network_mode: host
     volumes:
       - ./server/:/app/
       - /app/.venv
@@ -11,8 +10,12 @@ services:
       - ./server/.env
     environment:
       ENTRYPOINT: server
-    extra_hosts:
-      - "host.docker.internal:host-gateway"
+      DATABASE_URL: postgresql+asyncpg://reflector:reflector@localhost:5432/reflector
+      REDIS_HOST: localhost
+      CELERY_BROKER_URL: redis://localhost:6379/1
+      CELERY_RESULT_BACKEND: redis://localhost:6379/1
+      HATCHET_CLIENT_SERVER_URL: http://localhost:8889
+      HATCHET_CLIENT_HOST_PORT: localhost:7078
 
   worker:
     build:
@@ -24,6 +27,11 @@ services:
       - ./server/.env
     environment:
       ENTRYPOINT: worker
+      HATCHET_CLIENT_SERVER_URL: http://hatchet:8888
+      HATCHET_CLIENT_HOST_PORT: hatchet:7077
+    depends_on:
+      redis:
+        condition: service_started
 
   beat:
     build:
@@ -35,6 +43,9 @@ services:
       - ./server/.env
     environment:
       ENTRYPOINT: beat
+    depends_on:
+      redis:
+        condition: service_started
 
   hatchet-worker-cpu:
     build:
@@ -46,6 +57,8 @@ services:
       - ./server/.env
     environment:
       ENTRYPOINT: hatchet-worker-cpu
+      HATCHET_CLIENT_SERVER_URL: http://hatchet:8888
+      HATCHET_CLIENT_HOST_PORT: hatchet:7077
     depends_on:
       hatchet:
         condition: service_healthy
@@ -59,8 +72,8 @@ services:
       - ./server/.env
     environment:
       ENTRYPOINT: hatchet-worker-llm
-    extra_hosts:
-      - "host.docker.internal:host-gateway"
+      HATCHET_CLIENT_SERVER_URL: http://hatchet:8888
+      HATCHET_CLIENT_HOST_PORT: hatchet:7077
     depends_on:
       hatchet:
         condition: service_healthy
@@ -84,6 +97,11 @@ services:
       - ./www/.env.local
     environment:
       - NODE_ENV=development
+      - SERVER_API_URL=http://host.docker.internal:1250
+    extra_hosts:
+      - "host.docker.internal:host-gateway"
+    depends_on:
+      - server
 
   postgres:
     image: postgres:17
@@ -99,13 +117,14 @@ services:
       - ./server/docker/init-hatchet-db.sql:/docker-entrypoint-initdb.d/init-hatchet-db.sql:ro
     healthcheck:
       test: ["CMD-SHELL", "pg_isready -d reflector -U reflector"]
-      interval: 10s
-      timeout: 10s
-      retries: 5
-      start_period: 10s
+      interval: 5s
+      timeout: 5s
+      retries: 10
+      start_period: 15s
 
   hatchet:
     image: ghcr.io/hatchet-dev/hatchet/hatchet-lite:latest
+    restart: on-failure
     ports:
       - "8889:8888"
       - "7078:7077"
@@ -113,7 +132,7 @@ services:
       postgres:
         condition: service_healthy
     environment:
-      DATABASE_URL: "postgresql://reflector:reflector@postgres:5432/hatchet?sslmode=disable"
+      DATABASE_URL: "postgresql://reflector:reflector@postgres:5432/hatchet?sslmode=disable&connect_timeout=30"
       SERVER_AUTH_COOKIE_DOMAIN: localhost
       SERVER_AUTH_COOKIE_INSECURE: "t"
       SERVER_GRPC_BIND_ADDRESS: "0.0.0.0"
@@ -135,7 +154,3 @@ services:
 
 volumes:
   next_cache:
-
-networks:
-  default:
-    attachable: true

docs/docs/installation/setup-standalone.md

@@ -168,6 +168,8 @@ If the frontend or backend behaves unexpectedly (e.g., env vars seem ignored, ch
 lsof -i :3000   # frontend
 lsof -i :1250   # backend
 lsof -i :5432   # postgres
+lsof -i :3900   # Garage S3 API
+lsof -i :6379   # Redis
 
 # Kill stale processes on a port
 lsof -ti :3000 | xargs kill
@@ -175,9 +177,13 @@ lsof -ti :3000 | xargs kill
 
 Common causes:
 - A stale `next dev` or `pnpm dev` process from another terminal/worktree
-- Another Docker Compose project (different worktree) with containers on the same ports
+- Another Docker Compose project (different worktree) with containers on the same ports — the setup script only manages its own project; containers from other projects must be stopped manually (`docker ps` to find them, `docker stop` to kill them)
 
-The setup script checks for port conflicts before starting services.
+The setup script checks ports 3000, 1250, 5432, 6379, 3900, 3903 for conflicts before starting services. It ignores OrbStack/Docker Desktop port forwarding processes (which always bind these ports but are not real conflicts).
+
+### OrbStack false port-conflict warnings (Mac)
+
+If you use OrbStack as your Docker runtime, `lsof` will show OrbStack binding ports like 3000, 1250, etc. even when no containers are running. This is OrbStack's port forwarding mechanism — not a real conflict. The setup script filters these out automatically.
 
 ### Re-enabling authentication
 

scripts/setup-standalone.sh

@@ -113,7 +113,7 @@ step_llm() {
             echo ""
 
             # Pull model if not already present
-            if ollama list 2>/dev/null | grep -q "$MODEL"; then
+            if ollama list 2>/dev/null | awk '{print $1}' | grep -qx "$MODEL"; then
                 ok "Model $MODEL already pulled"
             else
                 info "Pulling model $MODEL (this may take a while)..."
@@ -143,7 +143,7 @@ step_llm() {
             echo ""
 
             # Pull model inside container
-            if compose_cmd exec "$OLLAMA_SVC" ollama list 2>/dev/null | grep -q "$MODEL"; then
+            if compose_cmd exec "$OLLAMA_SVC" ollama list 2>/dev/null | awk '{print $1}' | grep -qx "$MODEL"; then
                 ok "Model $MODEL already pulled"
             else
                 info "Pulling model $MODEL inside container (this may take a while)..."
@@ -290,16 +290,23 @@ ENVEOF
 step_services() {
     info "Step 5: Starting Docker services"
 
-    # Check for port conflicts — stale processes silently shadow Docker port mappings
+    # Check for port conflicts — stale processes silently shadow Docker port mappings.
+    # OrbStack/Docker Desktop bind ports for forwarding; ignore those PIDs.
     local ports_ok=true
-    for port in 3000 1250; do
-        local pid
-        pid=$(lsof -ti :"$port" 2>/dev/null || true)
-        if [[ -n "$pid" ]]; then
-            warn "Port $port already in use by PID $pid"
+    for port in 3000 1250 5432 6379 3900 3903; do
+        local pids
+        pids=$(lsof -ti :"$port" 2>/dev/null || true)
+        for pid in $pids; do
+            local pname
+            pname=$(ps -p "$pid" -o comm= 2>/dev/null || true)
+            # OrbStack and Docker Desktop own port forwarding — not real conflicts
+            if [[ "$pname" == *"OrbStack"* ]] || [[ "$pname" == *"com.docker"* ]] || [[ "$pname" == *"vpnkit"* ]]; then
+                continue
+            fi
+            warn "Port $port already in use by PID $pid ($pname)"
             warn "Kill it with: lsof -ti :$port | xargs kill"
             ports_ok=false
-        fi
+        done
     done
     if [[ "$ports_ok" == "false" ]]; then
         warn "Port conflicts detected — Docker containers may not be reachable"

server/reflector/auth/auth_none.py

@@ -1,11 +1,5 @@
-from typing import Annotated
-
-from fastapi import Depends
-from fastapi.security import OAuth2PasswordBearer
 from pydantic import BaseModel
 
-oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token", auto_error=False)
-
 
 class UserInfo(BaseModel):
     sub: str
@@ -15,13 +9,13 @@ class AccessTokenInfo(BaseModel):
     pass
 
 
-def authenticated(token: Annotated[str, Depends(oauth2_scheme)]):
+def authenticated():
     return None
 
 
-def current_user(token: Annotated[str, Depends(oauth2_scheme)]):
+def current_user():
     return None
 
 
-def current_user_optional(token: Annotated[str, Depends(oauth2_scheme)]):
+def current_user_optional():
     return None

server/reflector/hatchet/workflows/daily_multitrack_pipeline.py

@@ -720,7 +720,6 @@ async def detect_topics(input: PipelineInput, ctx: Context) -> TopicsResult:
                 chunk_text=chunk["text"],
                 timestamp=chunk["timestamp"],
                 duration=chunk["duration"],
-                words=chunk["words"],
             )
         )
         for chunk in chunks
@@ -732,31 +731,41 @@ async def detect_topics(input: PipelineInput, ctx: Context) -> TopicsResult:
         TopicChunkResult(**result[TaskName.DETECT_CHUNK_TOPIC]) for result in results
     ]
 
+    # Build index-to-words map from local chunks (words not in child workflow results)
+    chunks_by_index = {chunk["index"]: chunk["words"] for chunk in chunks}
+
     async with fresh_db_connection():
         transcript = await transcripts_controller.get_by_id(input.transcript_id)
         if not transcript:
             raise ValueError(f"Transcript {input.transcript_id} not found")
 
+        # Clear topics for idempotency on retry (each topic gets a fresh UUID,
+        # so upsert_topic would append duplicates without this)
+        await transcripts_controller.update(transcript, {"topics": []})
+
         for chunk in topic_chunks:
+            chunk_words = chunks_by_index[chunk.chunk_index]
             topic = TranscriptTopic(
                 title=chunk.title,
                 summary=chunk.summary,
                 timestamp=chunk.timestamp,
-                transcript=" ".join(w.text for w in chunk.words),
-                words=chunk.words,
+                transcript=" ".join(w.text for w in chunk_words),
+                words=chunk_words,
             )
             await transcripts_controller.upsert_topic(transcript, topic)
             await append_event_and_broadcast(
                 input.transcript_id, transcript, "TOPIC", topic, logger=logger
             )
 
+    # Words omitted from TopicsResult — already persisted to DB above.
+    # Downstream tasks that need words refetch from DB.
     topics_list = [
         TitleSummary(
             title=chunk.title,
             summary=chunk.summary,
             timestamp=chunk.timestamp,
             duration=chunk.duration,
-            transcript=TranscriptType(words=chunk.words),
+            transcript=TranscriptType(words=[]),
         )
         for chunk in topic_chunks
     ]
@@ -842,9 +851,8 @@ async def extract_subjects(input: PipelineInput, ctx: Context) -> SubjectsResult
     ctx.log(f"extract_subjects: starting for transcript_id={input.transcript_id}")
 
     topics_result = ctx.task_output(detect_topics)
-    topics = topics_result.topics
 
-    if not topics:
+    if not topics_result.topics:
         ctx.log("extract_subjects: no topics, returning empty subjects")
         return SubjectsResult(
             subjects=[],
@@ -857,11 +865,13 @@ async def extract_subjects(input: PipelineInput, ctx: Context) -> SubjectsResult
     # sharing DB connections and LLM HTTP pools across forks
     from reflector.db.transcripts import transcripts_controller  # noqa: PLC0415
     from reflector.llm import LLM  # noqa: PLC0415
+    from reflector.processors.types import words_to_segments  # noqa: PLC0415
 
     async with fresh_db_connection():
         transcript = await transcripts_controller.get_by_id(input.transcript_id)
 
-        # Build transcript text from topics (same logic as TranscriptFinalSummaryProcessor)
+        # Build transcript text from DB topics (words omitted from task output
+        # to reduce Hatchet payload size — refetch from DB where they were persisted)
         speakermap = {}
         if transcript and transcript.participants:
             speakermap = {
@@ -871,8 +881,8 @@ async def extract_subjects(input: PipelineInput, ctx: Context) -> SubjectsResult
             }
 
         text_lines = []
-        for topic in topics:
-            for segment in topic.transcript.as_segments():
+        for db_topic in transcript.topics:
+            for segment in words_to_segments(db_topic.words):
                 name = speakermap.get(segment.speaker, f"Speaker {segment.speaker}")
                 text_lines.append(f"{name}: {segment.text}")
 

server/reflector/hatchet/workflows/models.py

@@ -95,7 +95,6 @@ class TopicChunkResult(BaseModel):
     summary: str
     timestamp: float
     duration: float
-    words: list[Word]
 
 
 class TopicsResult(BaseModel):

server/reflector/hatchet/workflows/topic_chunk_processing.py

@@ -20,7 +20,6 @@ from reflector.hatchet.constants import LLM_RATE_LIMIT_KEY, TIMEOUT_MEDIUM
 from reflector.hatchet.workflows.models import TopicChunkResult
 from reflector.logger import logger
 from reflector.processors.prompts import TOPIC_PROMPT
-from reflector.processors.types import Word
 
 
 class TopicChunkInput(BaseModel):
@@ -30,7 +29,6 @@ class TopicChunkInput(BaseModel):
     chunk_text: str
     timestamp: float
     duration: float
-    words: list[Word]
 
 
 hatchet = HatchetClientManager.get_client()
@@ -99,5 +97,4 @@ async def detect_chunk_topic(input: TopicChunkInput, ctx: Context) -> TopicChunk
         summary=response.summary,
         timestamp=input.timestamp,
         duration=input.duration,
-        words=input.words,
     )

server/reflector/ws_manager.py

@@ -48,7 +48,15 @@ class RedisPubSubManager:
         if not self.redis_connection:
             await self.connect()
         message = json.dumps(message)
-        await self.redis_connection.publish(room_id, message)
+        try:
+            await self.redis_connection.publish(room_id, message)
+        except RuntimeError:
+            # Celery workers run each task in a new event loop (asyncio.run),
+            # which closes the previous loop. Cached Redis connection is dead.
+            # Reconnect on the current loop and retry.
+            self.redis_connection = None
+            await self.connect()
+            await self.redis_connection.publish(room_id, message)
 
     async def subscribe(self, room_id: str) -> redis.Redis:
         await self.pubsub.subscribe(room_id)

server/tests/test_hatchet_dispatch.py

@@ -291,7 +291,12 @@ async def test_validation_idle_transcript_with_recording_allowed():
         recording_id="test-recording-id",
     )
 
-    result = await validate_transcript_for_processing(mock_transcript)
+    with patch(
+        "reflector.services.transcript_process.task_is_scheduled_or_active"
+    ) as mock_celery_check:
+        mock_celery_check.return_value = False
+
+        result = await validate_transcript_for_processing(mock_transcript)
 
     assert isinstance(result, ValidationOk)
     assert result.recording_id == "test-recording-id"

server/tests/test_hatchet_payload_thinning.py

@@ -0,0 +1,185 @@
+"""
+Tests for Hatchet payload thinning optimizations.
+
+Verifies that:
+1. TopicChunkInput no longer carries words
+2. TopicChunkResult no longer carries words
+3. words_to_segments() matches Transcript.as_segments(is_multitrack=False) — behavioral equivalence
+   for the extract_subjects refactoring
+4. TopicsResult can be constructed with empty transcript words
+"""
+
+from reflector.hatchet.workflows.models import TopicChunkResult
+from reflector.hatchet.workflows.topic_chunk_processing import TopicChunkInput
+from reflector.processors.types import Word
+
+
+def _make_words(speaker: int = 0, start: float = 0.0) -> list[Word]:
+    return [
+        Word(text="Hello", start=start, end=start + 0.5, speaker=speaker),
+        Word(text=" world.", start=start + 0.5, end=start + 1.0, speaker=speaker),
+    ]
+
+
+class TestTopicChunkInputNoWords:
+    """TopicChunkInput must not have a words field."""
+
+    def test_no_words_field(self):
+        assert "words" not in TopicChunkInput.model_fields
+
+    def test_construction_without_words(self):
+        inp = TopicChunkInput(
+            chunk_index=0, chunk_text="Hello world.", timestamp=0.0, duration=1.0
+        )
+        assert inp.chunk_index == 0
+        assert inp.chunk_text == "Hello world."
+
+    def test_rejects_words_kwarg(self):
+        """Passing words= should raise a validation error (field doesn't exist)."""
+        import pydantic
+
+        try:
+            TopicChunkInput(
+                chunk_index=0,
+                chunk_text="text",
+                timestamp=0.0,
+                duration=1.0,
+                words=_make_words(),
+            )
+            # If pydantic is configured to ignore extra, this won't raise.
+            # Verify the field is still absent from the model.
+            assert "words" not in TopicChunkInput.model_fields
+        except pydantic.ValidationError:
+            pass  # Expected
+
+
+class TestTopicChunkResultNoWords:
+    """TopicChunkResult must not have a words field."""
+
+    def test_no_words_field(self):
+        assert "words" not in TopicChunkResult.model_fields
+
+    def test_construction_without_words(self):
+        result = TopicChunkResult(
+            chunk_index=0,
+            title="Test",
+            summary="Summary",
+            timestamp=0.0,
+            duration=1.0,
+        )
+        assert result.title == "Test"
+        assert result.chunk_index == 0
+
+    def test_serialization_roundtrip(self):
+        """Serialized TopicChunkResult has no words key."""
+        result = TopicChunkResult(
+            chunk_index=0,
+            title="Test",
+            summary="Summary",
+            timestamp=0.0,
+            duration=1.0,
+        )
+        data = result.model_dump()
+        assert "words" not in data
+        reconstructed = TopicChunkResult(**data)
+        assert reconstructed == result
+
+
+class TestWordsToSegmentsBehavioralEquivalence:
+    """words_to_segments() must produce same output as Transcript.as_segments(is_multitrack=False).
+
+    This ensures the extract_subjects refactoring (from task output topic.transcript.as_segments()
+    to words_to_segments(db_topic.words)) preserves identical behavior.
+    """
+
+    def test_single_speaker(self):
+        from reflector.processors.types import Transcript as TranscriptType
+        from reflector.processors.types import words_to_segments
+
+        words = _make_words(speaker=0)
+        direct = words_to_segments(words)
+        via_transcript = TranscriptType(words=words).as_segments(is_multitrack=False)
+
+        assert len(direct) == len(via_transcript)
+        for d, v in zip(direct, via_transcript):
+            assert d.text == v.text
+            assert d.speaker == v.speaker
+            assert d.start == v.start
+            assert d.end == v.end
+
+    def test_multiple_speakers(self):
+        from reflector.processors.types import Transcript as TranscriptType
+        from reflector.processors.types import words_to_segments
+
+        words = [
+            Word(text="Hello", start=0.0, end=0.5, speaker=0),
+            Word(text=" world.", start=0.5, end=1.0, speaker=0),
+            Word(text=" How", start=1.0, end=1.5, speaker=1),
+            Word(text=" are", start=1.5, end=2.0, speaker=1),
+            Word(text=" you?", start=2.0, end=2.5, speaker=1),
+        ]
+
+        direct = words_to_segments(words)
+        via_transcript = TranscriptType(words=words).as_segments(is_multitrack=False)
+
+        assert len(direct) == len(via_transcript)
+        for d, v in zip(direct, via_transcript):
+            assert d.text == v.text
+            assert d.speaker == v.speaker
+
+    def test_empty_words(self):
+        from reflector.processors.types import Transcript as TranscriptType
+        from reflector.processors.types import words_to_segments
+
+        assert words_to_segments([]) == []
+        assert TranscriptType(words=[]).as_segments(is_multitrack=False) == []
+
+
+class TestTopicsResultEmptyWords:
+    """TopicsResult can carry topics with empty transcript words."""
+
+    def test_construction_with_empty_words(self):
+        from reflector.hatchet.workflows.models import TopicsResult
+        from reflector.processors.types import TitleSummary
+        from reflector.processors.types import Transcript as TranscriptType
+
+        topics = [
+            TitleSummary(
+                title="Topic A",
+                summary="Summary A",
+                timestamp=0.0,
+                duration=5.0,
+                transcript=TranscriptType(words=[]),
+            ),
+            TitleSummary(
+                title="Topic B",
+                summary="Summary B",
+                timestamp=5.0,
+                duration=5.0,
+                transcript=TranscriptType(words=[]),
+            ),
+        ]
+        result = TopicsResult(topics=topics)
+        assert len(result.topics) == 2
+        for t in result.topics:
+            assert t.transcript.words == []
+
+    def test_serialization_roundtrip(self):
+        from reflector.hatchet.workflows.models import TopicsResult
+        from reflector.processors.types import TitleSummary
+        from reflector.processors.types import Transcript as TranscriptType
+
+        topics = [
+            TitleSummary(
+                title="Topic",
+                summary="Summary",
+                timestamp=0.0,
+                duration=1.0,
+                transcript=TranscriptType(words=[]),
+            )
+        ]
+        result = TopicsResult(topics=topics)
+        data = result.model_dump()
+        reconstructed = TopicsResult(**data)
+        assert len(reconstructed.topics) == 1
+        assert reconstructed.topics[0].transcript.words == []

www/app/(app)/transcripts/[transcriptId]/processing/page.tsx

@@ -11,6 +11,7 @@ import {
 import { useRouter } from "next/navigation";
 import { useTranscriptGet } from "../../../../lib/apiHooks";
 import { parseNonEmptyString } from "../../../../lib/utils";
+import { useWebSockets } from "../../useWebSockets";
 
 type TranscriptProcessing = {
   params: Promise<{
@@ -24,6 +25,7 @@ export default function TranscriptProcessing(details: TranscriptProcessing) {
   const router = useRouter();
 
   const transcript = useTranscriptGet(transcriptId);
+  useWebSockets(transcriptId);
 
   useEffect(() => {
     const status = transcript.data?.status;

www/app/(app)/transcripts/useWebRTC.ts

@@ -23,7 +23,16 @@ const useWebRTC = (
     let p: Peer;
 
     try {
-      p = new Peer({ initiator: true, stream: stream });
+      p = new Peer({
+        initiator: true,
+        stream: stream,
+        // Disable trickle ICE: single SDP exchange (offer + answer) with all candidates.
+        // Required for HTTP-based signaling; trickle needs WebSocket for candidate exchange.
+        trickle: false,
+        config: {
+          iceServers: [{ urls: "stun:stun.l.google.com:19302" }],
+        },
+      });
     } catch (error) {
       setError(error as Error, "Error creating WebRTC");
       return;

www/package.json

@@ -3,7 +3,7 @@
   "version": "0.1.0",
   "private": true,
   "scripts": {
-    "dev": "next dev --turbopack",
+    "dev": "next dev",
     "build": "next build",
     "build-production": "next build --experimental-build-mode compile",
     "start": "next start",