refactor: break up _record_utterance() into focused helpers

Extract _capture_audio_blocks() and _finalize_utterance() from the 73-line
_record_utterance() method, and promote _rms() to a module-level function.

Fixes #570

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

src/infrastructure/hands/shell.py

@@ -144,6 +144,65 @@ class ShellHand:
 
         return None
 
+    @staticmethod
+    def _build_run_env(env: dict | None) -> dict:
+        """Merge *env* overrides into a copy of the current environment."""
+        import os
+
+        run_env = os.environ.copy()
+        if env:
+            run_env.update(env)
+        return run_env
+
+    async def _execute_subprocess(
+        self,
+        command: str,
+        effective_timeout: int,
+        cwd: str | None,
+        run_env: dict,
+        start: float,
+    ) -> ShellResult:
+        """Run *command* as a subprocess with timeout enforcement."""
+        proc = await asyncio.create_subprocess_shell(
+            command,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+            cwd=cwd,
+            env=run_env,
+        )
+
+        try:
+            stdout_bytes, stderr_bytes = await asyncio.wait_for(
+                proc.communicate(), timeout=effective_timeout
+            )
+        except TimeoutError:
+            proc.kill()
+            await proc.wait()
+            latency = (time.time() - start) * 1000
+            logger.warning("Shell command timed out after %ds: %s", effective_timeout, command)
+            return ShellResult(
+                command=command,
+                success=False,
+                exit_code=-1,
+                error=f"Command timed out after {effective_timeout}s",
+                latency_ms=latency,
+                timed_out=True,
+            )
+
+        latency = (time.time() - start) * 1000
+        exit_code = proc.returncode if proc.returncode is not None else -1
+        stdout = stdout_bytes.decode("utf-8", errors="replace").strip()
+        stderr = stderr_bytes.decode("utf-8", errors="replace").strip()
+
+        return ShellResult(
+            command=command,
+            success=exit_code == 0,
+            exit_code=exit_code,
+            stdout=stdout,
+            stderr=stderr,
+            latency_ms=latency,
+        )
+
     async def run(
         self,
         command: str,
@@ -164,7 +223,6 @@ class ShellHand:
         """
         start = time.time()
 
-        # Validate
         validation_error = self._validate_command(command)
         if validation_error:
             return ShellResult(
@@ -178,52 +236,8 @@ class ShellHand:
         cwd = working_dir or self._working_dir
 
         try:
-            import os
+            run_env = self._build_run_env(env)
-
+            return await self._execute_subprocess(command, effective_timeout, cwd, run_env, start)
-            run_env = os.environ.copy()
-            if env:
-                run_env.update(env)
-
-            proc = await asyncio.create_subprocess_shell(
-                command,
-                stdout=asyncio.subprocess.PIPE,
-                stderr=asyncio.subprocess.PIPE,
-                cwd=cwd,
-                env=run_env,
-            )
-
-            try:
-                stdout_bytes, stderr_bytes = await asyncio.wait_for(
-                    proc.communicate(), timeout=effective_timeout
-                )
-            except TimeoutError:
-                proc.kill()
-                await proc.wait()
-                latency = (time.time() - start) * 1000
-                logger.warning("Shell command timed out after %ds: %s", effective_timeout, command)
-                return ShellResult(
-                    command=command,
-                    success=False,
-                    exit_code=-1,
-                    error=f"Command timed out after {effective_timeout}s",
-                    latency_ms=latency,
-                    timed_out=True,
-                )
-
-            latency = (time.time() - start) * 1000
-            exit_code = proc.returncode if proc.returncode is not None else -1
-            stdout = stdout_bytes.decode("utf-8", errors="replace").strip()
-            stderr = stderr_bytes.decode("utf-8", errors="replace").strip()
-
-            return ShellResult(
-                command=command,
-                success=exit_code == 0,
-                exit_code=exit_code,
-                stdout=stdout,
-                stderr=stderr,
-                latency_ms=latency,
-            )
-
         except Exception as exc:
             latency = (time.time() - start) * 1000
             logger.warning("Shell command failed: %s — %s", command, exc)

src/timmy/memory_system.py

@@ -98,6 +98,73 @@ def _get_table_columns(conn: sqlite3.Connection, table_name: str) -> set[str]:
     return {row[1] for row in cursor.fetchall()}
 
 
+def _migrate_episodes(conn: sqlite3.Connection) -> None:
+    """Migrate episodes table rows into the unified memories table."""
+    logger.info("Migration: Converting episodes table to memories")
+    try:
+        cols = _get_table_columns(conn, "episodes")
+        context_type_col = "context_type" if "context_type" in cols else "'conversation'"
+
+        conn.execute(f"""
+            INSERT INTO memories (
+                id, content, memory_type, source, embedding,
+                metadata, agent_id, task_id, session_id,
+                created_at, access_count, last_accessed
+            )
+            SELECT
+                id, content,
+                COALESCE({context_type_col}, 'conversation'),
+                COALESCE(source, 'agent'),
+                embedding,
+                metadata, agent_id, task_id, session_id,
+                COALESCE(timestamp, datetime('now')), 0, NULL
+            FROM episodes
+        """)
+        conn.execute("DROP TABLE episodes")
+        logger.info("Migration: Migrated episodes to memories")
+    except sqlite3.Error as exc:
+        logger.warning("Migration: Failed to migrate episodes: %s", exc)
+
+
+def _migrate_chunks(conn: sqlite3.Connection) -> None:
+    """Migrate chunks table rows into the unified memories table."""
+    logger.info("Migration: Converting chunks table to memories")
+    try:
+        cols = _get_table_columns(conn, "chunks")
+
+        id_col = "id" if "id" in cols else "CAST(rowid AS TEXT)"
+        content_col = "content" if "content" in cols else "text"
+        source_col = (
+            "filepath" if "filepath" in cols else ("source" if "source" in cols else "'vault'")
+        )
+        embedding_col = "embedding" if "embedding" in cols else "NULL"
+        created_col = "created_at" if "created_at" in cols else "datetime('now')"
+
+        conn.execute(f"""
+            INSERT INTO memories (
+                id, content, memory_type, source, embedding,
+                created_at, access_count
+            )
+            SELECT
+                {id_col}, {content_col}, 'vault_chunk', {source_col},
+                {embedding_col}, {created_col}, 0
+            FROM chunks
+        """)
+        conn.execute("DROP TABLE chunks")
+        logger.info("Migration: Migrated chunks to memories")
+    except sqlite3.Error as exc:
+        logger.warning("Migration: Failed to migrate chunks: %s", exc)
+
+
+def _drop_legacy_table(conn: sqlite3.Connection, table: str) -> None:
+    """Drop a legacy table if it exists."""
+    try:
+        conn.execute(f"DROP TABLE {table}")  # noqa: S608
+        logger.info("Migration: Dropped old %s table", table)
+    except sqlite3.Error as exc:
+        logger.warning("Migration: Failed to drop %s: %s", table, exc)
+
+
 def _migrate_schema(conn: sqlite3.Connection) -> None:
     """Migrate from old three-table schema to unified memories table.
 
@@ -110,78 +177,16 @@ def _migrate_schema(conn: sqlite3.Connection) -> None:
     tables = {row[0] for row in cursor.fetchall()}
 
     has_memories = "memories" in tables
-    has_episodes = "episodes" in tables
-    has_chunks = "chunks" in tables
-    has_facts = "facts" in tables
 
-    # Check if we need to migrate (old schema exists)
+    if not has_memories and (tables & {"episodes", "chunks", "facts"}):
-    if not has_memories and (has_episodes or has_chunks or has_facts):
         logger.info("Migration: Creating unified memories table")
-        # Schema will be created by _ensure_schema above
 
-    # Migrate episodes -> memories
+    if "episodes" in tables and has_memories:
-    if has_episodes and has_memories:
+        _migrate_episodes(conn)
-        logger.info("Migration: Converting episodes table to memories")
+    if "chunks" in tables and has_memories:
-        try:
+        _migrate_chunks(conn)
-            cols = _get_table_columns(conn, "episodes")
+    if "facts" in tables:
-            context_type_col = "context_type" if "context_type" in cols else "'conversation'"
+        _drop_legacy_table(conn, "facts")
-
-            conn.execute(f"""
-                INSERT INTO memories (
-                    id, content, memory_type, source, embedding,
-                    metadata, agent_id, task_id, session_id,
-                    created_at, access_count, last_accessed
-                )
-                SELECT 
-                    id, content, 
-                    COALESCE({context_type_col}, 'conversation'),
-                    COALESCE(source, 'agent'),
-                    embedding,
-                    metadata, agent_id, task_id, session_id,
-                    COALESCE(timestamp, datetime('now')), 0, NULL
-                FROM episodes
-            """)
-            conn.execute("DROP TABLE episodes")
-            logger.info("Migration: Migrated episodes to memories")
-        except sqlite3.Error as exc:
-            logger.warning("Migration: Failed to migrate episodes: %s", exc)
-
-    # Migrate chunks -> memories as vault_chunk
-    if has_chunks and has_memories:
-        logger.info("Migration: Converting chunks table to memories")
-        try:
-            cols = _get_table_columns(conn, "chunks")
-
-            id_col = "id" if "id" in cols else "CAST(rowid AS TEXT)"
-            content_col = "content" if "content" in cols else "text"
-            source_col = (
-                "filepath" if "filepath" in cols else ("source" if "source" in cols else "'vault'")
-            )
-            embedding_col = "embedding" if "embedding" in cols else "NULL"
-            created_col = "created_at" if "created_at" in cols else "datetime('now')"
-
-            conn.execute(f"""
-                INSERT INTO memories (
-                    id, content, memory_type, source, embedding,
-                    created_at, access_count
-                )
-                SELECT 
-                    {id_col}, {content_col}, 'vault_chunk', {source_col},
-                    {embedding_col}, {created_col}, 0
-                FROM chunks
-            """)
-            conn.execute("DROP TABLE chunks")
-            logger.info("Migration: Migrated chunks to memories")
-        except sqlite3.Error as exc:
-            logger.warning("Migration: Failed to migrate chunks: %s", exc)
-
-    # Drop old tables
-    if has_facts:
-        try:
-            conn.execute("DROP TABLE facts")
-            logger.info("Migration: Dropped old facts table")
-        except sqlite3.Error as exc:
-            logger.warning("Migration: Failed to drop facts: %s", exc)
 
     conn.commit()
 
@@ -298,6 +303,85 @@ def store_memory(
     return entry
 
 
+def _build_search_filters(
+    context_type: str | None,
+    agent_id: str | None,
+    session_id: str | None,
+) -> tuple[str, list]:
+    """Build SQL WHERE clause and params from search filters."""
+    conditions: list[str] = []
+    params: list = []
+
+    if context_type:
+        conditions.append("memory_type = ?")
+        params.append(context_type)
+    if agent_id:
+        conditions.append("agent_id = ?")
+        params.append(agent_id)
+    if session_id:
+        conditions.append("session_id = ?")
+        params.append(session_id)
+
+    where_clause = "WHERE " + " AND ".join(conditions) if conditions else ""
+    return where_clause, params
+
+
+def _fetch_memory_candidates(
+    where_clause: str, params: list, candidate_limit: int
+) -> list[sqlite3.Row]:
+    """Fetch candidate memory rows from the database."""
+    query_sql = f"""
+        SELECT * FROM memories
+        {where_clause}
+        ORDER BY created_at DESC
+        LIMIT ?
+    """
+    params.append(candidate_limit)
+
+    with get_connection() as conn:
+        return conn.execute(query_sql, params).fetchall()
+
+
+def _row_to_entry(row: sqlite3.Row) -> MemoryEntry:
+    """Convert a database row to a MemoryEntry."""
+    return MemoryEntry(
+        id=row["id"],
+        content=row["content"],
+        source=row["source"],
+        context_type=row["memory_type"],  # DB column -> API field
+        agent_id=row["agent_id"],
+        task_id=row["task_id"],
+        session_id=row["session_id"],
+        metadata=json.loads(row["metadata"]) if row["metadata"] else None,
+        embedding=json.loads(row["embedding"]) if row["embedding"] else None,
+        timestamp=row["created_at"],
+    )
+
+
+def _score_and_filter(
+    rows: list[sqlite3.Row],
+    query: str,
+    query_embedding: list[float],
+    min_relevance: float,
+) -> list[MemoryEntry]:
+    """Score candidate rows by similarity and filter by min_relevance."""
+    results = []
+    for row in rows:
+        entry = _row_to_entry(row)
+
+        if entry.embedding:
+            score = cosine_similarity(query_embedding, entry.embedding)
+        else:
+            score = _keyword_overlap(query, entry.content)
+
+        entry.relevance_score = score
+        if score >= min_relevance:
+            results.append(entry)
+
+    results.sort(key=lambda x: x.relevance_score or 0, reverse=True)
+    return results
+
+
 def search_memories(
     query: str,
     limit: int = 10,
@@ -320,65 +404,9 @@ def search_memories(
         List of MemoryEntry objects sorted by relevance
     """
     query_embedding = embed_text(query)
-
+    where_clause, params = _build_search_filters(context_type, agent_id, session_id)
-    # Build query with filters
+    rows = _fetch_memory_candidates(where_clause, params, limit * 3)
-    conditions = []
+    results = _score_and_filter(rows, query, query_embedding, min_relevance)
-    params = []
-
-    if context_type:
-        conditions.append("memory_type = ?")
-        params.append(context_type)
-    if agent_id:
-        conditions.append("agent_id = ?")
-        params.append(agent_id)
-    if session_id:
-        conditions.append("session_id = ?")
-        params.append(session_id)
-
-    where_clause = "WHERE " + " AND ".join(conditions) if conditions else ""
-
-    # Fetch candidates (we'll do in-memory similarity for now)
-    query_sql = f"""
-        SELECT * FROM memories
-        {where_clause}
-        ORDER BY created_at DESC
-        LIMIT ?
-    """
-    params.append(limit * 3)  # Get more candidates for ranking
-
-    with get_connection() as conn:
-        rows = conn.execute(query_sql, params).fetchall()
-
-    # Compute similarity scores
-    results = []
-    for row in rows:
-        entry = MemoryEntry(
-            id=row["id"],
-            content=row["content"],
-            source=row["source"],
-            context_type=row["memory_type"],  # DB column -> API field
-            agent_id=row["agent_id"],
-            task_id=row["task_id"],
-            session_id=row["session_id"],
-            metadata=json.loads(row["metadata"]) if row["metadata"] else None,
-            embedding=json.loads(row["embedding"]) if row["embedding"] else None,
-            timestamp=row["created_at"],
-        )
-
-        if entry.embedding:
-            score = cosine_similarity(query_embedding, entry.embedding)
-            entry.relevance_score = score
-            if score >= min_relevance:
-                results.append(entry)
-        else:
-            # Fallback: check for keyword overlap
-            score = _keyword_overlap(query, entry.content)
-            entry.relevance_score = score
-            if score >= min_relevance:
-                results.append(entry)
-
-    # Sort by relevance and return top results
-    results.sort(key=lambda x: x.relevance_score or 0, reverse=True)
     return results[:limit]
 
 

src/timmy/thinking.py

@@ -772,23 +772,10 @@ class ThinkingEngine:
         except Exception as exc:
             logger.debug("Thought issue filing skipped: %s", exc)
 
-    def _gather_system_snapshot(self) -> str:
+    # ── System snapshot helpers ────────────────────────────────────────────
-        """Gather lightweight real system state for grounding thoughts in reality.
 
-        Returns a short multi-line string with current time, thought count,
+    def _snap_thought_count(self, now: datetime) -> str | None:
-        recent chat activity, and task queue status. Never crashes — every
+        """Return today's thought count, or *None* on failure."""
-        section is independently try/excepted.
-        """
-        parts: list[str] = []
-
-        # Current local time
-        now = datetime.now().astimezone()
-        tz = now.strftime("%Z") or "UTC"
-        parts.append(
-            f"Local time: {now.strftime('%I:%M %p').lstrip('0')} {tz}, {now.strftime('%A %B %d')}"
-        )
-
-        # Thought count today (cheap DB query)
         try:
             today_start = now.replace(hour=0, minute=0, second=0, microsecond=0)
             with _get_conn(self._db_path) as conn:
@@ -796,66 +783,94 @@ class ThinkingEngine:
                     "SELECT COUNT(*) as c FROM thoughts WHERE created_at >= ?",
                     (today_start.isoformat(),),
                 ).fetchone()["c"]
-            parts.append(f"Thoughts today: {count}")
+            return f"Thoughts today: {count}"
         except Exception as exc:
             logger.debug("Thought count query failed: %s", exc)
-            pass
+            return None
 
-        # Recent chat activity (in-memory, no I/O)
+    def _snap_chat_activity(self) -> list[str]:
+        """Return chat-activity lines (in-memory, no I/O)."""
         try:
             from infrastructure.chat_store import message_log
 
             messages = message_log.all()
             if messages:
-                parts.append(f"Chat messages this session: {len(messages)}")
                 last = messages[-1]
-                parts.append(f'Last chat ({last.role}): "{last.content[:80]}"')
+                return [
-            else:
+                    f"Chat messages this session: {len(messages)}",
-                parts.append("No chat messages this session")
+                    f'Last chat ({last.role}): "{last.content[:80]}"',
+                ]
+            return ["No chat messages this session"]
         except Exception as exc:
             logger.debug("Chat activity query failed: %s", exc)
-            pass
+            return []
 
-        # Task queue (lightweight DB query)
+    def _snap_task_queue(self) -> str | None:
+        """Return a one-line task queue summary, or *None*."""
         try:
             from swarm.task_queue.models import get_task_summary_for_briefing
 
-            summary = get_task_summary_for_briefing()
+            s = get_task_summary_for_briefing()
-            running = summary.get("running", 0)
+            running, pending = s.get("running", 0), s.get("pending_approval", 0)
-            pending = summary.get("pending_approval", 0)
+            done, failed = s.get("completed", 0), s.get("failed", 0)
-            done = summary.get("completed", 0)
-            failed = summary.get("failed", 0)
             if running or pending or done or failed:
-                parts.append(
+                return (
                     f"Tasks: {running} running, {pending} pending, "
                     f"{done} completed, {failed} failed"
                 )
         except Exception as exc:
             logger.debug("Task queue query failed: %s", exc)
-            pass
+        return None
 
-        # Workspace updates (file-based communication with Hermes)
+    def _snap_workspace(self) -> list[str]:
+        """Return workspace-update lines (file-based Hermes comms)."""
         try:
             from timmy.workspace import workspace_monitor
 
             updates = workspace_monitor.get_pending_updates()
+            lines: list[str] = []
             new_corr = updates.get("new_correspondence")
-            new_inbox = updates.get("new_inbox_files", [])
-
             if new_corr:
-                # Count entries (assuming each entry starts with a timestamp or header)
+                line_count = len([ln for ln in new_corr.splitlines() if ln.strip()])
-                line_count = len([line for line in new_corr.splitlines() if line.strip()])
+                lines.append(
-                parts.append(
                     f"Workspace: {line_count} new correspondence entries (latest from: Hermes)"
                 )
+            new_inbox = updates.get("new_inbox_files", [])
             if new_inbox:
                 files_str = ", ".join(new_inbox[:5])
                 if len(new_inbox) > 5:
                     files_str += f", ... (+{len(new_inbox) - 5} more)"
-                parts.append(f"Workspace: {len(new_inbox)} new inbox files: {files_str}")
+                lines.append(f"Workspace: {len(new_inbox)} new inbox files: {files_str}")
+            return lines
         except Exception as exc:
             logger.debug("Workspace check failed: %s", exc)
-            pass
+            return []
+
+    def _gather_system_snapshot(self) -> str:
+        """Gather lightweight real system state for grounding thoughts in reality.
+
+        Returns a short multi-line string with current time, thought count,
+        recent chat activity, and task queue status. Never crashes — every
+        section is independently try/excepted.
+        """
+        now = datetime.now().astimezone()
+        tz = now.strftime("%Z") or "UTC"
+
+        parts: list[str] = [
+            f"Local time: {now.strftime('%I:%M %p').lstrip('0')} {tz}, {now.strftime('%A %B %d')}"
+        ]
+
+        thought_line = self._snap_thought_count(now)
+        if thought_line:
+            parts.append(thought_line)
+
+        parts.extend(self._snap_chat_activity())
+
+        task_line = self._snap_task_queue()
+        if task_line:
+            parts.append(task_line)
+
+        parts.extend(self._snap_workspace())
 
         return "\n".join(parts) if parts else ""
 

src/timmy/tools.py

@@ -909,82 +909,35 @@ def _experiment_tool_catalog() -> dict:
     }
 
 
+_CREATIVE_CATALOG_SOURCES: list[tuple[str, str, list[str]]] = [
+    ("creative.tools.git_tools", "GIT_TOOL_CATALOG", ["forge", "helm", "orchestrator"]),
+    ("creative.tools.image_tools", "IMAGE_TOOL_CATALOG", ["pixel", "orchestrator"]),
+    ("creative.tools.music_tools", "MUSIC_TOOL_CATALOG", ["lyra", "orchestrator"]),
+    ("creative.tools.video_tools", "VIDEO_TOOL_CATALOG", ["reel", "orchestrator"]),
+    ("creative.director", "DIRECTOR_TOOL_CATALOG", ["orchestrator"]),
+    ("creative.assembler", "ASSEMBLER_TOOL_CATALOG", ["reel", "orchestrator"]),
+]
+
+
 def _import_creative_catalogs(catalog: dict) -> None:
     """Import and merge creative tool catalogs from creative module."""
-    # ── Git tools ─────────────────────────────────────────────────────────────
+    for module_path, attr_name, available_in in _CREATIVE_CATALOG_SOURCES:
-    try:
+        _merge_catalog(catalog, module_path, attr_name, available_in)
-        from creative.tools.git_tools import GIT_TOOL_CATALOG
 
-        for tool_id, info in GIT_TOOL_CATALOG.items():
+
+def _merge_catalog(
+    catalog: dict, module_path: str, attr_name: str, available_in: list[str]
+) -> None:
+    """Import a single creative catalog and merge its entries."""
+    try:
+        from importlib import import_module
+
+        source_catalog = getattr(import_module(module_path), attr_name)
+        for tool_id, info in source_catalog.items():
             catalog[tool_id] = {
                 "name": info["name"],
                 "description": info["description"],
-                "available_in": ["forge", "helm", "orchestrator"],
+                "available_in": available_in,
-            }
-    except ImportError:
-        pass
-
-    # ── Image tools ────────────────────────────────────────────────────────────
-    try:
-        from creative.tools.image_tools import IMAGE_TOOL_CATALOG
-
-        for tool_id, info in IMAGE_TOOL_CATALOG.items():
-            catalog[tool_id] = {
-                "name": info["name"],
-                "description": info["description"],
-                "available_in": ["pixel", "orchestrator"],
-            }
-    except ImportError:
-        pass
-
-    # ── Music tools ────────────────────────────────────────────────────────────
-    try:
-        from creative.tools.music_tools import MUSIC_TOOL_CATALOG
-
-        for tool_id, info in MUSIC_TOOL_CATALOG.items():
-            catalog[tool_id] = {
-                "name": info["name"],
-                "description": info["description"],
-                "available_in": ["lyra", "orchestrator"],
-            }
-    except ImportError:
-        pass
-
-    # ── Video tools ────────────────────────────────────────────────────────────
-    try:
-        from creative.tools.video_tools import VIDEO_TOOL_CATALOG
-
-        for tool_id, info in VIDEO_TOOL_CATALOG.items():
-            catalog[tool_id] = {
-                "name": info["name"],
-                "description": info["description"],
-                "available_in": ["reel", "orchestrator"],
-            }
-    except ImportError:
-        pass
-
-    # ── Creative pipeline ──────────────────────────────────────────────────────
-    try:
-        from creative.director import DIRECTOR_TOOL_CATALOG
-
-        for tool_id, info in DIRECTOR_TOOL_CATALOG.items():
-            catalog[tool_id] = {
-                "name": info["name"],
-                "description": info["description"],
-                "available_in": ["orchestrator"],
-            }
-    except ImportError:
-        pass
-
-    # ── Assembler tools ───────────────────────────────────────────────────────
-    try:
-        from creative.assembler import ASSEMBLER_TOOL_CATALOG
-
-        for tool_id, info in ASSEMBLER_TOOL_CATALOG.items():
-            catalog[tool_id] = {
-                "name": info["name"],
-                "description": info["description"],
-                "available_in": ["reel", "orchestrator"],
             }
     except ImportError:
         pass

src/timmy/voice_loop.py

@@ -78,6 +78,11 @@ DEFAULT_MAX_UTTERANCE = 30.0  # safety cap — don't record forever
 DEFAULT_SESSION_ID = "voice"
 
 
+def _rms(block: np.ndarray) -> float:
+    """Compute root-mean-square energy of an audio block."""
+    return float(np.sqrt(np.mean(block.astype(np.float32) ** 2)))
+
+
 @dataclass
 class VoiceConfig:
     """Configuration for the voice loop."""
@@ -161,13 +166,6 @@ class VoiceLoop:
         min_blocks = int(self.config.min_utterance / 0.1)
         max_blocks = int(self.config.max_utterance / 0.1)
 
-        audio_chunks: list[np.ndarray] = []
-        silent_count = 0
-        recording = False
-
-        def _rms(block: np.ndarray) -> float:
-            return float(np.sqrt(np.mean(block.astype(np.float32) ** 2)))
-
         sys.stdout.write("\n  🎤 Listening... (speak now)\n")
         sys.stdout.flush()
 
@@ -177,42 +175,70 @@ class VoiceLoop:
             dtype="float32",
             blocksize=block_size,
         ) as stream:
-            while self._running:
+            chunks = self._capture_audio_blocks(stream, block_size, silence_blocks, max_blocks)
-                block, overflowed = stream.read(block_size)
-                if overflowed:
-                    logger.debug("Audio buffer overflowed")
 
-                rms = _rms(block)
+        return self._finalize_utterance(chunks, min_blocks, sr)
 
-                if not recording:
+    def _capture_audio_blocks(
-                    if rms > self.config.silence_threshold:
+        self,
-                        recording = True
+        stream,
-                        silent_count = 0
+        block_size: int,
-                        audio_chunks.append(block.copy())
+        silence_blocks: int,
-                        sys.stdout.write("  📢 Recording...\r")
+        max_blocks: int,
-                        sys.stdout.flush()
+    ) -> list[np.ndarray]:
-                else:
+        """Read audio blocks from *stream* until silence or safety cap.
+
+        Returns the list of captured audio blocks (may be empty if no
+        speech was detected).
+        """
+        audio_chunks: list[np.ndarray] = []
+        silent_count = 0
+        recording = False
+
+        while self._running:
+            block, overflowed = stream.read(block_size)
+            if overflowed:
+                logger.debug("Audio buffer overflowed")
+
+            rms = _rms(block)
+
+            if not recording:
+                if rms > self.config.silence_threshold:
+                    recording = True
+                    silent_count = 0
                     audio_chunks.append(block.copy())
+                    sys.stdout.write("  📢 Recording...\r")
+                    sys.stdout.flush()
+            else:
+                audio_chunks.append(block.copy())
 
-                    if rms < self.config.silence_threshold:
+                if rms < self.config.silence_threshold:
-                        silent_count += 1
+                    silent_count += 1
-                    else:
+                else:
-                        silent_count = 0
+                    silent_count = 0
 
-                    # End of utterance
+                if silent_count >= silence_blocks:
-                    if silent_count >= silence_blocks:
+                    break
-                        break
 
-                    # Safety cap
+                if len(audio_chunks) >= max_blocks:
-                    if len(audio_chunks) >= max_blocks:
+                    logger.info("Max utterance length reached, stopping.")
-                        logger.info("Max utterance length reached, stopping.")
+                    break
-                        break
 
-        if not audio_chunks or len(audio_chunks) < min_blocks:
+        return audio_chunks
+
+    @staticmethod
+    def _finalize_utterance(
+        chunks: list[np.ndarray], min_blocks: int, sample_rate: int
+    ) -> np.ndarray | None:
+        """Concatenate captured chunks and report duration.
+
+        Returns None if the utterance is too short (below *min_blocks*).
+        """
+        if not chunks or len(chunks) < min_blocks:
             return None
 
-        audio = np.concatenate(audio_chunks, axis=0).flatten()
+        audio = np.concatenate(chunks, axis=0).flatten()
-        duration = len(audio) / sr
+        duration = len(audio) / sample_rate
         sys.stdout.write(f"  ✂️  Captured {duration:.1f}s of audio\n")
         sys.stdout.flush()
         return audio
@@ -369,15 +395,33 @@ class VoiceLoop:
 
     # ── Main Loop ───────────────────────────────────────────────────────
 
-    def run(self) -> None:
+    # Whisper hallucinates these on silence/noise — skip them.
-        """Run the voice loop. Blocks until Ctrl-C."""
+    _WHISPER_HALLUCINATIONS = frozenset(
-        self._ensure_piper()
+        {
+            "you",
+            "thanks.",
+            "thank you.",
+            "bye.",
+            "",
+            "thanks for watching!",
+            "thank you for watching!",
+        }
+    )
 
-        # Suppress MCP / Agno stderr noise during voice mode.
+    # Spoken phrases that end the voice session.
-        _suppress_mcp_noise()
+    _EXIT_COMMANDS = frozenset(
-        # Suppress MCP async-generator teardown tracebacks on exit.
+        {
-        _install_quiet_asyncgen_hooks()
+            "goodbye",
+            "exit",
+            "quit",
+            "stop",
+            "goodbye timmy",
+            "stop listening",
+        }
+    )
 
+    def _log_banner(self) -> None:
+        """Log the startup banner with STT/TTS/LLM configuration."""
         tts_label = (
             "macOS say"
             if self.config.use_say_fallback
@@ -393,52 +437,50 @@ class VoiceLoop:
             "  Press Ctrl-C to exit.\n" + "=" * 60
         )
 
+    def _is_hallucination(self, text: str) -> bool:
+        """Return True if *text* is a known Whisper hallucination."""
+        return not text or text.lower() in self._WHISPER_HALLUCINATIONS
+
+    def _is_exit_command(self, text: str) -> bool:
+        """Return True if the user asked to stop the voice session."""
+        return text.lower().strip().rstrip(".!") in self._EXIT_COMMANDS
+
+    def _process_turn(self, text: str) -> None:
+        """Handle a single listen-think-speak turn after transcription."""
+        sys.stdout.write(f"\n  👤 You: {text}\n")
+        sys.stdout.flush()
+
+        response = self._think(text)
+        sys.stdout.write(f"  🤖 Timmy: {response}\n")
+        sys.stdout.flush()
+
+        self._speak(response)
+
+    def run(self) -> None:
+        """Run the voice loop. Blocks until Ctrl-C."""
+        self._ensure_piper()
+        _suppress_mcp_noise()
+        _install_quiet_asyncgen_hooks()
+        self._log_banner()
+
         self._running = True
 
         try:
             while self._running:
-                # 1. LISTEN — record until silence
                 audio = self._record_utterance()
                 if audio is None:
                     continue
 
-                # 2. TRANSCRIBE — Whisper STT
                 text = self._transcribe(audio)
-                if not text or text.lower() in (
+                if self._is_hallucination(text):
-                    "you",
-                    "thanks.",
-                    "thank you.",
-                    "bye.",
-                    "",
-                    "thanks for watching!",
-                    "thank you for watching!",
-                ):
-                    # Whisper hallucinations on silence/noise
                     logger.debug("Ignoring likely Whisper hallucination: '%s'", text)
                     continue
 
-                sys.stdout.write(f"\n  👤 You: {text}\n")
+                if self._is_exit_command(text):
-                sys.stdout.flush()
-
-                # Exit commands
-                if text.lower().strip().rstrip(".!") in (
-                    "goodbye",
-                    "exit",
-                    "quit",
-                    "stop",
-                    "goodbye timmy",
-                    "stop listening",
-                ):
                     logger.info("👋 Goodbye!")
                     break
 
-                # 3. THINK — send to Timmy
+                self._process_turn(text)
-                response = self._think(text)
-                sys.stdout.write(f"  🤖 Timmy: {response}\n")
-                sys.stdout.flush()
-
-                # 4. SPEAK — TTS output
-                self._speak(response)
 
         except KeyboardInterrupt:
             logger.info("👋 Voice loop stopped.")

tests/timmy/test_voice_loop.py

@@ -15,7 +15,7 @@ except ImportError:
     np = None
 
 try:
-    from timmy.voice_loop import VoiceConfig, VoiceLoop, _strip_markdown
+    from timmy.voice_loop import VoiceConfig, VoiceLoop, _rms, _strip_markdown
 except ImportError:
     pass  # pytestmark will skip all tests anyway
 
@@ -236,6 +236,7 @@ class TestHallucinationFilter:
     """Whisper tends to hallucinate on silence/noise. The loop should filter these."""
 
     def test_known_hallucinations_filtered(self):
+        loop = VoiceLoop()
         hallucinations = [
             "you",
             "thanks.",
@@ -243,33 +244,35 @@ class TestHallucinationFilter:
             "Bye.",
             "Thanks for watching!",
             "Thank you for watching!",
+            "",
         ]
         for text in hallucinations:
-            assert text.lower() in (
+            assert loop._is_hallucination(text), f"'{text}' should be filtered"
-                "you",
+
-                "thanks.",
+    def test_real_speech_not_filtered(self):
-                "thank you.",
+        loop = VoiceLoop()
-                "bye.",
+        assert not loop._is_hallucination("Hello Timmy")
-                "",
+        assert not loop._is_hallucination("What time is it?")
-                "thanks for watching!",
-                "thank you for watching!",
-            ), f"'{text}' should be filtered"
 
 
 class TestExitCommands:
     """Voice loop should recognize exit commands."""
 
     def test_exit_commands(self):
+        loop = VoiceLoop()
         exits = ["goodbye", "exit", "quit", "stop", "goodbye timmy", "stop listening"]
         for cmd in exits:
-            assert cmd.lower().strip().rstrip(".!") in (
+            assert loop._is_exit_command(cmd), f"'{cmd}' should be an exit command"
-                "goodbye",
+
-                "exit",
+    def test_exit_with_punctuation(self):
-                "quit",
+        loop = VoiceLoop()
-                "stop",
+        assert loop._is_exit_command("goodbye!")
-                "goodbye timmy",
+        assert loop._is_exit_command("stop.")
-                "stop listening",
+
-            ), f"'{cmd}' should be an exit command"
+    def test_non_exit_commands(self):
+        loop = VoiceLoop()
+        assert not loop._is_exit_command("hello")
+        assert not loop._is_exit_command("what time is it")
 
 
 class TestPlayAudio:
@@ -333,3 +336,28 @@ class TestSpeakSetsFlag:
 
         # After speak
         assert loop._speaking is False
+
+
+class TestRms:
+    def test_rms_of_silence(self):
+        block = np.zeros(1600, dtype=np.float32)
+        assert _rms(block) == 0.0
+
+    def test_rms_of_signal(self):
+        block = np.ones(1600, dtype=np.float32) * 0.5
+        assert abs(_rms(block) - 0.5) < 1e-5
+
+
+class TestFinalizeUtterance:
+    def test_returns_none_for_empty(self):
+        assert VoiceLoop._finalize_utterance([], min_blocks=5, sample_rate=16000) is None
+
+    def test_returns_none_below_min(self):
+        chunks = [np.zeros(1600, dtype=np.float32) for _ in range(3)]
+        assert VoiceLoop._finalize_utterance(chunks, min_blocks=5, sample_rate=16000) is None
+
+    def test_concatenates_chunks(self):
+        chunks = [np.ones(1600, dtype=np.float32) for _ in range(5)]
+        result = VoiceLoop._finalize_utterance(chunks, min_blocks=3, sample_rate=16000)
+        assert result is not None
+        assert len(result) == 8000