feat: make perception_cache confidence threshold configurable

Move hardcoded 0.85 threshold from perception_cache.py to config.py
as perception_confidence_threshold setting with default 0.85.

Fixes #1259

docs/research/autoresearch-h1-baseline.md

@@ -0,0 +1,132 @@
+# Autoresearch H1 — M3 Max Baseline
+
+**Status:** Baseline established (Issue #905)
+**Hardware:** Apple M3 Max · 36 GB unified memory
+**Date:** 2026-03-23
+**Refs:** #905 · #904 (parent) · #881 (M3 Max compute) · #903 (MLX benchmark)
+
+---
+
+## Setup
+
+### Prerequisites
+
+```bash
+# Install MLX (Apple Silicon — definitively faster than llama.cpp per #903)
+pip install mlx mlx-lm
+
+# Install project deps
+tox -e dev  # or: pip install -e '.[dev]'
+```
+
+### Clone & prepare
+
+`prepare_experiment` in `src/timmy/autoresearch.py` handles the clone.
+On Apple Silicon it automatically sets `AUTORESEARCH_BACKEND=mlx` and
+`AUTORESEARCH_DATASET=tinystories`.
+
+```python
+from timmy.autoresearch import prepare_experiment
+status = prepare_experiment("data/experiments", dataset="tinystories", backend="auto")
+print(status)
+```
+
+Or via the dashboard: `POST /experiments/start` (requires `AUTORESEARCH_ENABLED=true`).
+
+### Configuration (`.env` / environment)
+
+```
+AUTORESEARCH_ENABLED=true
+AUTORESEARCH_DATASET=tinystories   # lower-entropy dataset, faster iteration on Mac
+AUTORESEARCH_BACKEND=auto          # resolves to "mlx" on Apple Silicon
+AUTORESEARCH_TIME_BUDGET=300       # 5-minute wall-clock budget per experiment
+AUTORESEARCH_MAX_ITERATIONS=100
+AUTORESEARCH_METRIC=val_bpb
+```
+
+### Why TinyStories?
+
+Karpathy's recommendation for resource-constrained hardware: lower entropy
+means the model can learn meaningful patterns in less time and with a smaller
+vocabulary, yielding cleaner val_bpb curves within the 5-minute budget.
+
+---
+
+## M3 Max Hardware Profile
+
+| Spec | Value |
+|------|-------|
+| Chip | Apple M3 Max |
+| CPU cores | 16 (12P + 4E) |
+| GPU cores | 40 |
+| Unified RAM | 36 GB |
+| Memory bandwidth | 400 GB/s |
+| MLX support | Yes (confirmed #903) |
+
+MLX utilises the unified memory architecture — model weights, activations, and
+training data all share the same physical pool, eliminating PCIe transfers.
+This gives M3 Max a significant throughput advantage over external GPU setups
+for models that fit in 36 GB.
+
+---
+
+## Community Reference Data
+
+| Hardware | Experiments | Succeeded | Failed | Outcome |
+|----------|-------------|-----------|--------|---------|
+| Mac Mini M4 | 35 | 7 | 28 | Model improved by simplifying |
+| Shopify (overnight) | ~50 | — | — | 19% quality gain; smaller beat 2× baseline |
+| SkyPilot (16× GPU, 8 h) | ~910 | — | — | 2.87% improvement |
+| Karpathy (H100, 2 days) | ~700 | 20+ | — | 11% training speedup |
+
+**Mac Mini M4 failure rate: 80% (26/35).** Failures are expected and by design —
+the 5-minute budget deliberately prunes slow experiments. The 20% success rate
+still yielded an improved model.
+
+---
+
+## Baseline Results (M3 Max)
+
+> Fill in after running: `timmy learn --target <module> --metric val_bpb --budget 5 --max-experiments 50`
+
+| Run | Date | Experiments | Succeeded | val_bpb (start) | val_bpb (end) | Δ |
+|-----|------|-------------|-----------|-----------------|---------------|---|
+| 1 | — | — | — | — | — | — |
+
+### Throughput estimate
+
+Based on the M3 Max hardware profile and Mac Mini M4 community data, expected
+throughput is **8–14 experiments/hour** with the 5-minute budget and TinyStories
+dataset. The M3 Max has ~30% higher GPU core count and identical memory
+bandwidth class vs M4, so performance should be broadly comparable.
+
+---
+
+## Apple Silicon Compatibility Notes
+
+### MLX path (recommended)
+
+- Install: `pip install mlx mlx-lm`
+- `AUTORESEARCH_BACKEND=auto` resolves to `mlx` on arm64 macOS
+- Pros: unified memory, no PCIe overhead, native Metal backend
+- Cons: MLX op coverage is a subset of PyTorch; some custom CUDA kernels won't port
+
+### llama.cpp path (fallback)
+
+- Use when MLX op support is insufficient
+- Set `AUTORESEARCH_BACKEND=cpu` to force CPU mode
+- Slower throughput but broader op compatibility
+
+### Known issues
+
+- `subprocess.TimeoutExpired` is the normal termination path — autoresearch
+  treats timeout as a completed-but-pruned experiment, not a failure
+- Large batch sizes may trigger OOM if other processes hold unified memory;
+  set `PYTORCH_MPS_HIGH_WATERMARK_RATIO=0.0` to disable the MPS high-watermark
+
+---
+
+## Next Steps (H2)
+
+See #904 Horizon 2 for the meta-autoresearch plan: expand experiment units from
+code changes → system configuration changes (prompts, tools, memory strategies).

scripts/loop_guard.py

@@ -240,9 +240,33 @@ def compute_backoff(consecutive_idle: int) -> int:
     return min(BACKOFF_BASE * (BACKOFF_MULTIPLIER ** consecutive_idle), BACKOFF_MAX)
 
 
+def seed_cycle_result(item: dict) -> None:
+    """Pre-seed cycle_result.json with the top queue item.
+
+    Only writes if cycle_result.json does not already exist — never overwrites
+    agent-written data.  This ensures cycle_retro.py can always resolve the
+    issue number even when the dispatcher (claude-loop, gemini-loop, etc.) does
+    not write cycle_result.json itself.
+    """
+    if CYCLE_RESULT_FILE.exists():
+        return  # Agent already wrote its own result — leave it alone
+
+    seed = {
+        "issue": item.get("issue"),
+        "type": item.get("type", "unknown"),
+    }
+    try:
+        CYCLE_RESULT_FILE.parent.mkdir(parents=True, exist_ok=True)
+        CYCLE_RESULT_FILE.write_text(json.dumps(seed) + "\n")
+        print(f"[loop-guard] Seeded cycle_result.json with issue #{seed['issue']}")
+    except OSError as exc:
+        print(f"[loop-guard] WARNING: Could not seed cycle_result.json: {exc}")
+
+
 def main() -> int:
     wait_mode = "--wait" in sys.argv
     status_mode = "--status" in sys.argv
+    pick_mode = "--pick" in sys.argv
 
     state = load_idle_state()
 
@@ -269,6 +293,17 @@ def main() -> int:
         state["consecutive_idle"] = 0
         state["last_idle_at"] = 0
         save_idle_state(state)
+
+        # Pre-seed cycle_result.json so cycle_retro.py can resolve issue=
+        # even when the dispatcher doesn't write the file itself.
+        seed_cycle_result(ready[0])
+
+        if pick_mode:
+            # Emit the top issue number to stdout for shell script capture.
+            issue = ready[0].get("issue")
+            if issue is not None:
+                print(issue)
+
         return 0
 
     # Queue empty — apply backoff

src/config.py

@@ -387,6 +387,11 @@ class Settings(BaseSettings):
     autoresearch_time_budget: int = 300  # seconds per experiment run
     autoresearch_max_iterations: int = 100
     autoresearch_metric: str = "val_bpb"  # metric to optimise (lower = better)
+    # M3 Max / Apple Silicon tuning (Issue #905).
+    # dataset: "tinystories" (default, lower-entropy, recommended for Mac) or "openwebtext".
+    autoresearch_dataset: str = "tinystories"
+    # backend: "auto" detects MLX on Apple Silicon; "cpu" forces CPU fallback.
+    autoresearch_backend: str = "auto"
 
     # ── Weekly Narrative Summary ───────────────────────────────────────
     # Generates a human-readable weekly summary of development activity.
@@ -450,6 +455,10 @@ class Settings(BaseSettings):
     # Background meditation interval in seconds (0 = disabled).
     scripture_meditation_interval: int = 0
 
+    # ── Perception Cache ───────────────────────────────────────────────
+    # Minimum confidence threshold for template matching in perception cache.
+    perception_confidence_threshold: float = 0.85
+
     def _compute_repo_root(self) -> str:
         """Auto-detect repo root if not set."""
         if self.repo_root:

src/dashboard/app.py

@@ -42,19 +42,19 @@ from dashboard.routes.hermes import router as hermes_router
 from dashboard.routes.loop_qa import router as loop_qa_router
 from dashboard.routes.memory import router as memory_router
 from dashboard.routes.mobile import router as mobile_router
-from dashboard.routes.nexus import router as nexus_router
 from dashboard.routes.models import api_router as models_api_router
 from dashboard.routes.models import router as models_router
+from dashboard.routes.nexus import router as nexus_router
 from dashboard.routes.quests import router as quests_router
 from dashboard.routes.scorecards import router as scorecards_router
 from dashboard.routes.sovereignty_metrics import router as sovereignty_metrics_router
 from dashboard.routes.sovereignty_ws import router as sovereignty_ws_router
-from dashboard.routes.three_strike import router as three_strike_router
 from dashboard.routes.spark import router as spark_router
 from dashboard.routes.system import router as system_router
 from dashboard.routes.tasks import router as tasks_router
 from dashboard.routes.telegram import router as telegram_router
 from dashboard.routes.thinking import router as thinking_router
+from dashboard.routes.three_strike import router as three_strike_router
 from dashboard.routes.tools import router as tools_router
 from dashboard.routes.tower import router as tower_router
 from dashboard.routes.voice import router as voice_router

src/dashboard/routes/nexus.py

@@ -12,7 +12,7 @@ Routes:
 
 import asyncio
 import logging
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 
 from fastapi import APIRouter, Form, Request
 from fastapi.responses import HTMLResponse
@@ -39,7 +39,7 @@ _nexus_log: list[dict] = []
 
 
 def _ts() -> str:
-    return datetime.now(timezone.utc).strftime("%H:%M:%S")
+    return datetime.now(UTC).strftime("%H:%M:%S")
 
 
 def _append_log(role: str, content: str) -> None:
@@ -94,9 +94,7 @@ async def nexus_chat(request: Request, message: str = Form(...)):
 
     # Fetch semantically relevant memories to surface in the sidebar
     try:
-        memory_hits = await asyncio.to_thread(
-            search_memories, query=message, limit=4
-        )
+        memory_hits = await asyncio.to_thread(search_memories, query=message, limit=4)
     except Exception as exc:
         logger.warning("Nexus memory search failed: %s", exc)
         memory_hits = []

src/dashboard/routes/three_strike.py

@@ -101,9 +101,7 @@ async def record_strike(body: RecordRequest) -> dict[str, Any]:
 
 
 @router.post("/{category}/{key}/automation")
-async def register_automation(
-    category: str, key: str, body: AutomationRequest
-) -> dict[str, bool]:
+async def register_automation(category: str, key: str, body: AutomationRequest) -> dict[str, bool]:
     """Register an automation artifact to unblock a (category, key) pair."""
     detector = get_detector()
     detector.register_automation(category, key, body.artifact_path)

src/infrastructure/router/cascade.py

@@ -16,7 +16,10 @@ from dataclasses import dataclass, field
 from datetime import UTC, datetime
 from enum import Enum
 from pathlib import Path
-from typing import Any
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from infrastructure.router.classifier import TaskComplexity
 
 from config import settings
 

src/infrastructure/router/classifier.py

@@ -13,7 +13,7 @@ from enum import Enum
 class TaskComplexity(Enum):
     """Task complexity tier for model routing."""
 
-    SIMPLE = "simple"   # Qwen3-8B Q6_K: routine, latency-sensitive
+    SIMPLE = "simple"  # Qwen3-8B Q6_K: routine, latency-sensitive
     COMPLEX = "complex"  # Qwen3-14B Q5_K_M: quality-sensitive, multi-step
 
 
@@ -118,12 +118,15 @@ def classify_task(messages: list[dict]) -> TaskComplexity:
         return TaskComplexity.SIMPLE
 
     # Concatenate all user-turn content for analysis
-    user_content = " ".join(
-        msg.get("content", "")
-        for msg in messages
-        if msg.get("role") in ("user", "human")
-        and isinstance(msg.get("content"), str)
-    ).lower().strip()
+    user_content = (
+        " ".join(
+            msg.get("content", "")
+            for msg in messages
+            if msg.get("role") in ("user", "human") and isinstance(msg.get("content"), str)
+        )
+        .lower()
+        .strip()
+    )
 
     if not user_content:
         return TaskComplexity.SIMPLE

src/timmy/autoresearch.py

@@ -8,7 +8,7 @@ Flow:
   1. prepare_experiment  — clone repo + run data prep
   2. run_experiment      — execute train.py with wall-clock timeout
   3. evaluate_result     — compare metric against baseline
-  4. experiment_loop     — orchestrate the full cycle
+  4. SystemExperiment    — orchestrate the full cycle via class interface
 
 All subprocess calls are guarded with timeouts for graceful degradation.
 """
@@ -17,9 +17,12 @@ from __future__ import annotations
 
 import json
 import logging
+import os
+import platform
 import re
 import subprocess
 import time
+from collections.abc import Callable
 from pathlib import Path
 from typing import Any
 
@@ -29,15 +32,61 @@ DEFAULT_REPO = "https://github.com/karpathy/autoresearch.git"
 _METRIC_RE = re.compile(r"val_bpb[:\s]+([0-9]+\.?[0-9]*)")
 
 
+# ── Higher-is-better metric names ────────────────────────────────────────────
+_HIGHER_IS_BETTER = frozenset({"unit_pass_rate", "coverage"})
+
+
+def is_apple_silicon() -> bool:
+    """Return True when running on Apple Silicon (M-series chip)."""
+    return platform.system() == "Darwin" and platform.machine() == "arm64"
+
+
+def _build_experiment_env(
+    dataset: str = "tinystories",
+    backend: str = "auto",
+) -> dict[str, str]:
+    """Build environment variables for an autoresearch subprocess.
+
+    Args:
+        dataset: Dataset name forwarded as ``AUTORESEARCH_DATASET``.
+            ``"tinystories"`` is recommended for Apple Silicon (lower entropy,
+            faster iteration).
+        backend: Inference backend forwarded as ``AUTORESEARCH_BACKEND``.
+            ``"auto"`` enables MLX on Apple Silicon; ``"cpu"`` forces CPU.
+
+    Returns:
+        Merged environment dict (inherits current process env).
+    """
+    env = os.environ.copy()
+    env["AUTORESEARCH_DATASET"] = dataset
+
+    if backend == "auto":
+        env["AUTORESEARCH_BACKEND"] = "mlx" if is_apple_silicon() else "cuda"
+    else:
+        env["AUTORESEARCH_BACKEND"] = backend
+
+    return env
+
+
 def prepare_experiment(
     workspace: Path,
     repo_url: str = DEFAULT_REPO,
+    dataset: str = "tinystories",
+    backend: str = "auto",
 ) -> str:
     """Clone autoresearch repo and run data preparation.
 
+    On Apple Silicon the ``dataset`` defaults to ``"tinystories"`` (lower
+    entropy, faster iteration) and ``backend`` to ``"auto"`` which resolves to
+    MLX.  Both values are forwarded as ``AUTORESEARCH_DATASET`` /
+    ``AUTORESEARCH_BACKEND`` environment variables so that ``prepare.py`` and
+    ``train.py`` can adapt their behaviour without CLI changes.
+
     Args:
         workspace: Directory to set up the experiment in.
         repo_url: Git URL for the autoresearch repository.
+        dataset: Dataset name; ``"tinystories"`` is recommended on Mac.
+        backend: Inference backend; ``"auto"`` picks MLX on Apple Silicon.
 
     Returns:
         Status message describing what was prepared.
@@ -59,6 +108,14 @@ def prepare_experiment(
     else:
         logger.info("Autoresearch repo already present at %s", repo_dir)
 
+    env = _build_experiment_env(dataset=dataset, backend=backend)
+    if is_apple_silicon():
+        logger.info(
+            "Apple Silicon detected — dataset=%s backend=%s",
+            env["AUTORESEARCH_DATASET"],
+            env["AUTORESEARCH_BACKEND"],
+        )
+
     # Run prepare.py (data download + tokeniser training)
     prepare_script = repo_dir / "prepare.py"
     if prepare_script.exists():
@@ -69,6 +126,7 @@ def prepare_experiment(
             text=True,
             cwd=str(repo_dir),
             timeout=300,
+            env=env,
         )
         if result.returncode != 0:
             return f"Preparation failed: {result.stderr.strip()[:500]}"
@@ -81,6 +139,8 @@ def run_experiment(
     workspace: Path,
     timeout: int = 300,
     metric_name: str = "val_bpb",
+    dataset: str = "tinystories",
+    backend: str = "auto",
 ) -> dict[str, Any]:
     """Run a single training experiment with a wall-clock timeout.
 
@@ -88,6 +148,9 @@ def run_experiment(
         workspace: Experiment workspace (contains autoresearch/ subdir).
         timeout: Maximum wall-clock seconds for the run.
         metric_name: Name of the metric to extract from stdout.
+        dataset: Dataset forwarded to the subprocess via env var.
+        backend: Inference backend forwarded via env var (``"auto"`` → MLX on
+            Apple Silicon, CUDA otherwise).
 
     Returns:
         Dict with keys: metric (float|None), log (str), duration_s (int),
@@ -105,6 +168,7 @@ def run_experiment(
             "error": f"train.py not found in {repo_dir}",
         }
 
+    env = _build_experiment_env(dataset=dataset, backend=backend)
     start = time.monotonic()
     try:
         result = subprocess.run(
@@ -113,6 +177,7 @@ def run_experiment(
             text=True,
             cwd=str(repo_dir),
             timeout=timeout,
+            env=env,
         )
         duration = int(time.monotonic() - start)
         output = result.stdout + result.stderr
@@ -125,7 +190,7 @@ def run_experiment(
             "log": output[-2000:],  # Keep last 2k chars
             "duration_s": duration,
             "success": result.returncode == 0,
-            "error": None if result.returncode == 0 else f"Exit code {result.returncode}",
+            "error": (None if result.returncode == 0 else f"Exit code {result.returncode}"),
         }
     except subprocess.TimeoutExpired:
         duration = int(time.monotonic() - start)
@@ -237,10 +302,6 @@ def _extract_coverage(output: str) -> float | None:
     return None
 
 
-# ── Higher-is-better metric names ────────────────────────────────────────────
-_HIGHER_IS_BETTER = frozenset({"unit_pass_rate", "coverage"})
-
-
 class SystemExperiment:
     """An autoresearch experiment targeting a specific module with a configurable metric.
 
@@ -254,6 +315,10 @@ class SystemExperiment:
             Any other value is forwarded to :func:`_extract_metric`.
         budget_minutes: Wall-clock budget per experiment (default 5 min).
         workspace: Working directory for subprocess calls.  Defaults to ``cwd``.
+        revert_on_failure: Whether to revert changes on failed experiments.
+        hypothesis: Optional natural language hypothesis for the experiment.
+        metric_fn: Optional callable for custom metric extraction.
+            If provided, overrides built-in metric extraction.
     """
 
     def __init__(
@@ -262,11 +327,19 @@ class SystemExperiment:
         metric: str = "unit_pass_rate",
         budget_minutes: int = 5,
         workspace: Path | None = None,
+        revert_on_failure: bool = True,
+        hypothesis: str = "",
+        metric_fn: Callable[[str], float | None] | None = None,
     ) -> None:
         self.target = target
         self.metric = metric
         self.budget_seconds = budget_minutes * 60
         self.workspace = Path(workspace) if workspace else Path.cwd()
+        self.revert_on_failure = revert_on_failure
+        self.hypothesis = hypothesis
+        self.metric_fn = metric_fn
+        self.results: list[dict[str, Any]] = []
+        self.baseline: float | None = None
 
     # ── Hypothesis generation ─────────────────────────────────────────────────
 
@@ -341,7 +414,7 @@ class SystemExperiment:
                 "log": output[-3000:],
                 "duration_s": duration,
                 "success": result.returncode == 0,
-                "error": None if result.returncode == 0 else f"Exit code {result.returncode}",
+                "error": (None if result.returncode == 0 else f"Exit code {result.returncode}"),
             }
         except subprocess.TimeoutExpired:
             duration = int(time.monotonic() - start)
@@ -353,10 +426,24 @@ class SystemExperiment:
                 "error": f"Budget exceeded after {self.budget_seconds}s",
             }
         except OSError as exc:
-            return {"metric": None, "log": "", "duration_s": 0, "success": False, "error": str(exc)}
+            return {
+                "metric": None,
+                "log": "",
+                "duration_s": 0,
+                "success": False,
+                "error": str(exc),
+            }
 
     def _extract_tox_metric(self, output: str) -> float | None:
         """Dispatch to the correct metric extractor based on *self.metric*."""
+        # Use custom metric function if provided
+        if self.metric_fn is not None:
+            try:
+                return self.metric_fn(output)
+            except Exception as exc:
+                logger.warning("Custom metric_fn failed: %s", exc)
+                return None
+
         if self.metric == "unit_pass_rate":
             return _extract_pass_rate(output)
         if self.metric == "coverage":
@@ -391,12 +478,24 @@ class SystemExperiment:
 
     # ── Git phase ─────────────────────────────────────────────────────────────
 
+    def create_branch(self, branch_name: str) -> bool:
+        """Create and checkout a new git branch. Returns True on success."""
+        try:
+            subprocess.run(
+                ["git", "checkout", "-b", branch_name],
+                cwd=str(self.workspace),
+                check=True,
+                timeout=30,
+            )
+            return True
+        except subprocess.CalledProcessError as exc:
+            logger.warning("Git branch creation failed: %s", exc)
+            return False
+
     def commit_changes(self, message: str) -> bool:
         """Stage and commit all changes.  Returns True on success."""
         try:
-            subprocess.run(
-                ["git", "add", "-A"], cwd=str(self.workspace), check=True, timeout=30
-            )
+            subprocess.run(["git", "add", "-A"], cwd=str(self.workspace), check=True, timeout=30)
             subprocess.run(
                 ["git", "commit", "-m", message],
                 cwd=str(self.workspace),
@@ -421,3 +520,126 @@ class SystemExperiment:
         except subprocess.CalledProcessError as exc:
             logger.warning("Git revert failed: %s", exc)
             return False
+
+    # ── Full experiment loop ──────────────────────────────────────────────────
+
+    def run(
+        self,
+        tox_env: str = "unit",
+        model: str = "qwen3:30b",
+        program_content: str = "",
+        max_iterations: int = 1,
+        dry_run: bool = False,
+        create_branch: bool = False,
+    ) -> dict[str, Any]:
+        """Run the full experiment loop: hypothesis → edit → tox → evaluate → commit/revert.
+
+        This method encapsulates the complete experiment cycle, running multiple
+        iterations until an improvement is found or max_iterations is reached.
+
+        Args:
+            tox_env: Tox environment to run (default "unit").
+            model: Ollama model for Aider edits (default "qwen3:30b").
+            program_content: Research direction for hypothesis generation.
+            max_iterations: Maximum number of experiment iterations.
+            dry_run: If True, only generate hypotheses without making changes.
+            create_branch: If True, create a new git branch for the experiment.
+
+        Returns:
+            Dict with keys: ``success`` (bool), ``final_metric`` (float|None),
+            ``baseline`` (float|None), ``iterations`` (int), ``results`` (list).
+        """
+        if create_branch:
+            branch_name = f"autoresearch/{self.target.replace('/', '-')}-{int(time.time())}"
+            self.create_branch(branch_name)
+
+        baseline: float | None = self.baseline
+        final_metric: float | None = None
+        success = False
+
+        for iteration in range(1, max_iterations + 1):
+            logger.info("Experiment iteration %d/%d", iteration, max_iterations)
+
+            # Generate hypothesis
+            hypothesis = self.hypothesis or self.generate_hypothesis(program_content)
+            logger.info("Hypothesis: %s", hypothesis)
+
+            # In dry-run mode, just record the hypothesis and continue
+            if dry_run:
+                result_record = {
+                    "iteration": iteration,
+                    "hypothesis": hypothesis,
+                    "metric": None,
+                    "baseline": baseline,
+                    "assessment": "Dry-run: no changes made",
+                    "success": True,
+                    "duration_s": 0,
+                }
+                self.results.append(result_record)
+                continue
+
+            # Apply edit
+            edit_result = self.apply_edit(hypothesis, model=model)
+            edit_failed = "not available" in edit_result or edit_result.startswith("Aider error")
+            if edit_failed:
+                logger.warning("Edit phase failed: %s", edit_result)
+
+            # Run evaluation
+            tox_result = self.run_tox(tox_env=tox_env)
+            metric = tox_result["metric"]
+
+            # Evaluate result
+            assessment = self.evaluate(metric, baseline)
+            logger.info("Assessment: %s", assessment)
+
+            # Store result
+            result_record = {
+                "iteration": iteration,
+                "hypothesis": hypothesis,
+                "metric": metric,
+                "baseline": baseline,
+                "assessment": assessment,
+                "success": tox_result["success"],
+                "duration_s": tox_result["duration_s"],
+            }
+            self.results.append(result_record)
+
+            # Set baseline on first successful run
+            if metric is not None and baseline is None:
+                baseline = metric
+                self.baseline = baseline
+                final_metric = metric
+                continue
+
+            # Determine if we should commit or revert
+            should_commit = False
+            if tox_result["success"] and metric is not None and baseline is not None:
+                if self.is_improvement(metric, baseline):
+                    should_commit = True
+                    final_metric = metric
+                    baseline = metric
+                    self.baseline = baseline
+                    success = True
+
+            if should_commit:
+                commit_msg = f"autoresearch: improve {self.metric} on {self.target}\n\n{hypothesis}"
+                if self.commit_changes(commit_msg):
+                    logger.info("Changes committed")
+                else:
+                    self.revert_changes()
+                    logger.warning("Commit failed, changes reverted")
+            elif self.revert_on_failure:
+                self.revert_changes()
+                logger.info("Changes reverted (no improvement)")
+
+            # Early exit if we found an improvement
+            if success:
+                break
+
+        return {
+            "success": success,
+            "final_metric": final_metric,
+            "baseline": self.baseline,
+            "iterations": len(self.results),
+            "results": self.results,
+        }

src/timmy/cli.py

@@ -347,7 +347,10 @@ def interview(
         # Force agent creation by calling chat once with a warm-up prompt
         try:
             loop.run_until_complete(
-                chat("Hello, Timmy. We're about to start your interview.", session_id="interview")
+                chat(
+                    "Hello, Timmy. We're about to start your interview.",
+                    session_id="interview",
+                )
             )
         except Exception as exc:
             typer.echo(f"Warning: Initialization issue — {exc}", err=True)
@@ -410,11 +413,17 @@ def down():
 @app.command()
 def voice(
     whisper_model: str = typer.Option(
-        "base.en", "--whisper", "-w", help="Whisper model: tiny.en, base.en, small.en, medium.en"
+        "base.en",
+        "--whisper",
+        "-w",
+        help="Whisper model: tiny.en, base.en, small.en, medium.en",
     ),
     use_say: bool = typer.Option(False, "--say", help="Use macOS `say` instead of Piper TTS"),
     threshold: float = typer.Option(
-        0.015, "--threshold", "-t", help="Mic silence threshold (RMS). Lower = more sensitive."
+        0.015,
+        "--threshold",
+        "-t",
+        help="Mic silence threshold (RMS). Lower = more sensitive.",
     ),
     silence: float = typer.Option(1.5, "--silence", help="Seconds of silence to end recording"),
     backend: str | None = _BACKEND_OPTION,
@@ -457,7 +466,8 @@ def route(
 @app.command()
 def focus(
     topic: str | None = typer.Argument(
-        None, help='Topic to focus on (e.g. "three-phase loop"). Omit to show current focus.'
+        None,
+        help='Topic to focus on (e.g. "three-phase loop"). Omit to show current focus.',
     ),
     clear: bool = typer.Option(False, "--clear", "-c", help="Clear focus and return to broad mode"),
 ):
@@ -604,7 +614,8 @@ def learn(
 
     if target is None:
         typer.echo(
-            "Error: --target is required. Specify the module or file to optimise.", err=True
+            "Error: --target is required. Specify the module or file to optimise.",
+            err=True,
         )
         raise typer.Exit(1)
 
@@ -621,54 +632,50 @@ def learn(
         typer.echo("  (dry-run — no changes will be made)")
     typer.echo()
 
-    baseline: float | None = None
+    def _progress_callback(iteration: int, max_iter: int, message: str) -> None:
+        """Print progress updates during experiment iterations."""
+        if iteration > 0:
+            prefix = typer.style(f"[{iteration}/{max_iter}]", bold=True)
+            typer.echo(f"{prefix} {message}")
 
     try:
-        for i in range(1, max_experiments + 1):
-            typer.echo(typer.style(f"[{i}/{max_experiments}]", bold=True), nl=False)
+        # Run the full experiment loop via the SystemExperiment class
+        result = experiment.run(
+            tox_env=tox_env,
+            model=model,
+            program_content=program_content,
+            max_iterations=max_experiments,
+            dry_run=dry_run,
+            create_branch=False,  # CLI mode: work on current branch
+        )
 
-            hypothesis = experiment.generate_hypothesis(program_content)
-            typer.echo(f" {hypothesis}")
+        # Display results for each iteration
+        for i, record in enumerate(experiment.results, 1):
+            _progress_callback(i, max_experiments, record["hypothesis"])
 
             if dry_run:
                 continue
 
-            # Edit phase
+            # Edit phase result
             typer.echo("  → editing …", nl=False)
-            edit_result = experiment.apply_edit(hypothesis, model=model)
-            if "not available" in edit_result or edit_result.startswith("Aider error"):
-                typer.echo(f" skipped ({edit_result.split(':')[0]})")
+            if record.get("edit_failed"):
+                typer.echo(f" skipped ({record.get('edit_result', 'unknown')})")
             else:
                 typer.echo(" done")
 
-            # Evaluate phase
-            typer.echo("  → running tox …", nl=False)
-            tox_result = experiment.run_tox(tox_env=tox_env)
-            typer.echo(f" {tox_result['duration_s']}s")
+            # Evaluate phase result
+            duration = record.get("duration_s", 0)
+            typer.echo(f"  → running tox … {duration}s")
 
-            assessment = experiment.evaluate(tox_result["metric"], baseline)
+            # Assessment
+            assessment = record.get("assessment", "No assessment")
             typer.echo(f"  → {assessment}")
 
-            if tox_result["metric"] is not None and baseline is None:
-                baseline = tox_result["metric"]
-
-            if tox_result["success"] and tox_result["metric"] is not None and baseline is not None:
-                if experiment.is_improvement(tox_result["metric"], baseline):
-                    commit_msg = (
-                        f"autoresearch: improve {metric} on {target} — {assessment}"
-                    )
-                    if experiment.commit_changes(commit_msg):
-                        typer.echo("  → committed")
-                        baseline = tox_result["metric"]
-                    else:
-                        experiment.revert_changes()
-                        typer.echo("  → commit failed, reverted")
-                else:
-                    experiment.revert_changes()
-                    typer.echo("  → reverted (no improvement)")
-            elif not tox_result["success"]:
-                experiment.revert_changes()
-                typer.echo(f"  → reverted ({tox_result['error']})")
+            # Outcome
+            if record.get("committed"):
+                typer.echo("  → committed")
+            elif record.get("reverted"):
+                typer.echo("  → reverted (no improvement)")
 
             typer.echo()
 
@@ -677,8 +684,8 @@ def learn(
         raise typer.Exit(0) from None
 
     typer.echo(typer.style("Autoresearch complete.", bold=True))
-    if baseline is not None:
-        typer.echo(f"Final {metric}: {baseline:.4f}")
+    if result.get("baseline") is not None:
+        typer.echo(f"Final {metric}: {result['baseline']:.4f}")
 
 
 def main():

src/timmy/memory/embeddings.py

@@ -7,10 +7,11 @@ Also includes vector similarity utilities (cosine similarity, keyword overlap).
 """
 
 import hashlib
+import json
 import logging
 import math
-import json
-import httpx # Import httpx for Ollama API calls
+
+import httpx  # Import httpx for Ollama API calls
 
 from config import settings
 
@@ -20,14 +21,21 @@ logger = logging.getLogger(__name__)
 EMBEDDING_MODEL = None
 EMBEDDING_DIM = 384  # MiniLM dimension, will be overridden if Ollama model has different dim
 
+
 class OllamaEmbedder:
     """Mimics SentenceTransformer interface for Ollama."""
+
     def __init__(self, model_name: str, ollama_url: str):
         self.model_name = model_name
         self.ollama_url = ollama_url
-        self.dimension = 0 # Will be updated after first call
+        self.dimension = 0  # Will be updated after first call
 
-    def encode(self, sentences: str | list[str], convert_to_numpy: bool = False, normalize_embeddings: bool = True) -> list[list[float]] | list[float]:
+    def encode(
+        self,
+        sentences: str | list[str],
+        convert_to_numpy: bool = False,
+        normalize_embeddings: bool = True,
+    ) -> list[list[float]] | list[float]:
         """Generate embeddings using Ollama."""
         if isinstance(sentences, str):
             sentences = [sentences]
@@ -43,9 +51,9 @@ class OllamaEmbedder:
                 response.raise_for_status()
                 embedding = response.json()["embedding"]
                 if not self.dimension:
-                    self.dimension = len(embedding) # Set dimension on first successful call
+                    self.dimension = len(embedding)  # Set dimension on first successful call
                     global EMBEDDING_DIM
-                    EMBEDDING_DIM = self.dimension # Update global EMBEDDING_DIM
+                    EMBEDDING_DIM = self.dimension  # Update global EMBEDDING_DIM
                 all_embeddings.append(embedding)
             except httpx.RequestError as exc:
                 logger.error("Ollama embeddings request failed: %s", exc)
@@ -59,6 +67,7 @@ class OllamaEmbedder:
             return all_embeddings[0]
         return all_embeddings
 
+
 def _get_embedding_model():
     """Lazy-load embedding model, preferring Ollama if configured."""
     global EMBEDDING_MODEL
@@ -69,8 +78,13 @@ def _get_embedding_model():
             return EMBEDDING_MODEL
 
         if settings.timmy_embedding_backend == "ollama":
-            logger.info("MemorySystem: Using Ollama for embeddings with model %s", settings.ollama_embedding_model)
-            EMBEDDING_MODEL = OllamaEmbedder(settings.ollama_embedding_model, settings.normalized_ollama_url)
+            logger.info(
+                "MemorySystem: Using Ollama for embeddings with model %s",
+                settings.ollama_embedding_model,
+            )
+            EMBEDDING_MODEL = OllamaEmbedder(
+                settings.ollama_embedding_model, settings.normalized_ollama_url
+            )
             # We don't know the dimension until after the first call, so keep it default for now.
             # It will be updated dynamically in OllamaEmbedder.encode
             return EMBEDDING_MODEL
@@ -79,7 +93,7 @@ def _get_embedding_model():
                 from sentence_transformers import SentenceTransformer
 
                 EMBEDDING_MODEL = SentenceTransformer("all-MiniLM-L6-v2")
-                EMBEDDING_DIM = 384 # Reset to MiniLM dimension
+                EMBEDDING_DIM = 384  # Reset to MiniLM dimension
                 logger.info("MemorySystem: Loaded local embedding model (all-MiniLM-L6-v2)")
             except ImportError:
                 logger.warning("MemorySystem: sentence-transformers not installed, using fallback")
@@ -107,13 +121,12 @@ def embed_text(text: str) -> list[float]:
     if model and model is not False:
         embedding = model.encode(text)
         # Ensure it's a list of floats, not numpy array
-        if hasattr(embedding, 'tolist'):
+        if hasattr(embedding, "tolist"):
             return embedding.tolist()
         return embedding
     return _simple_hash_embedding(text)
 
 
-
 def cosine_similarity(a: list[float], b: list[float]) -> float:
     """Calculate cosine similarity between two vectors."""
     dot = sum(x * y for x, y in zip(a, b, strict=False))

src/timmy/memory_system.py

@@ -1318,11 +1318,11 @@ def memory_store(topic: str, report: str, type: str = "research") -> str:
     try:
         # Dedup check for facts and research — skip if similar exists
         if type in ("fact", "research"):
-            existing = search_memories(
-                full_content, limit=3, context_type=type, min_relevance=0.75
-            )
+            existing = search_memories(full_content, limit=3, context_type=type, min_relevance=0.75)
             if existing:
-                return f"Similar {type} already stored (id={existing[0].id[:8]}). Skipping duplicate."
+                return (
+                    f"Similar {type} already stored (id={existing[0].id[:8]}). Skipping duplicate."
+                )
 
         entry = store_memory(
             content=full_content,

src/timmy/sovereignty/perception_cache.py

@@ -8,6 +8,8 @@ from typing import Any
 import cv2
 import numpy as np
 
+from config import settings
+
 
 @dataclass
 class Template:
@@ -43,7 +45,7 @@ class PerceptionCache:
                 best_match_confidence = max_val
                 best_match_name = template.name
 
-        if best_match_confidence > 0.85:  # TODO: Make this configurable per template
+        if best_match_confidence > settings.perception_confidence_threshold:
             return CacheResult(
                 confidence=best_match_confidence, state={"template_name": best_match_name}
             )

src/timmy/sovereignty/three_strike.py

@@ -222,9 +222,7 @@ class ThreeStrikeStore:
             ThreeStrikeError: On the third (or later) strike with no automation.
         """
         if category not in CATEGORIES:
-            raise ValueError(
-                f"Unknown category '{category}'. Valid: {sorted(CATEGORIES)}"
-            )
+            raise ValueError(f"Unknown category '{category}'. Valid: {sorted(CATEGORIES)}")
 
         now = datetime.now(UTC).isoformat()
         meta_json = json.dumps(metadata or {})
@@ -404,9 +402,7 @@ class ThreeStrikeStore:
         """Return all strike records ordered by last seen (most recent first)."""
         try:
             with closing(self._connect()) as conn:
-                rows = conn.execute(
-                    "SELECT * FROM strikes ORDER BY last_seen DESC"
-                ).fetchall()
+                rows = conn.execute("SELECT * FROM strikes ORDER BY last_seen DESC").fetchall()
             return [
                 StrikeRecord(
                     category=r["category"],

src/timmy/tools/__init__.py

@@ -20,12 +20,12 @@ Sub-modules:
 # ``from timmy.tools import <symbol>`` continue to work unchanged.
 
 from timmy.tools._base import (
+    _AGNO_TOOLS_AVAILABLE,
+    _TOOL_USAGE,
     AgentTools,
     PersonaTools,
     ToolStats,
-    _AGNO_TOOLS_AVAILABLE,
     _ImportError,
-    _TOOL_USAGE,
     _track_tool_usage,
     get_tool_stats,
 )

src/timmy/tools/_base.py

@@ -11,10 +11,10 @@ logger = logging.getLogger(__name__)
 # Lazy imports to handle test mocking
 _ImportError = None
 try:
-    from agno.tools import Toolkit
-    from agno.tools.file import FileTools
-    from agno.tools.python import PythonTools
-    from agno.tools.shell import ShellTools
+    from agno.tools import Toolkit  # noqa: F401
+    from agno.tools.file import FileTools  # noqa: F401
+    from agno.tools.python import PythonTools  # noqa: F401
+    from agno.tools.shell import ShellTools  # noqa: F401
 
     _AGNO_TOOLS_AVAILABLE = True
 except ImportError as e:
@@ -41,7 +41,7 @@ class AgentTools:
 
     agent_id: str
     agent_name: str
-    toolkit: "Toolkit"
+    toolkit: Toolkit
     available_tools: list[str] = field(default_factory=list)
 
 

src/timmy/tools/_registry.py

@@ -16,11 +16,11 @@ from pathlib import Path
 
 from timmy.tools._base import (
     _AGNO_TOOLS_AVAILABLE,
-    _ImportError,
     FileTools,
     PythonTools,
     ShellTools,
     Toolkit,
+    _ImportError,
 )
 from timmy.tools.file_tools import (
     _make_smart_read_file,
@@ -90,10 +90,10 @@ def _register_grok_tool(toolkit: Toolkit) -> None:
 def _register_memory_tools(toolkit: Toolkit) -> None:
     """Register memory search, write, and forget tools."""
     try:
-        from timmy.memory_system import memory_forget, memory_read, memory_search, memory_write
+        from timmy.memory_system import memory_forget, memory_read, memory_search, memory_store
 
         toolkit.register(memory_search, name="memory_search")
-        toolkit.register(memory_write, name="memory_write")
+        toolkit.register(memory_store, name="memory_write")
         toolkit.register(memory_read, name="memory_read")
         toolkit.register(memory_forget, name="memory_forget")
     except (ImportError, AttributeError) as exc:
@@ -363,7 +363,7 @@ AGENT_TOOLKITS: dict[str, Callable[[], Toolkit]] = {
 }
 
 
-def get_tools_for_agent(agent_id: str, base_dir: str | Path | None = None) -> "Toolkit | None":
+def get_tools_for_agent(agent_id: str, base_dir: str | Path | None = None) -> Toolkit | None:
     """Get the appropriate toolkit for an agent.
 
     Args:

src/timmy/tools/file_tools.py

@@ -13,16 +13,16 @@ from pathlib import Path
 
 from timmy.tools._base import (
     _AGNO_TOOLS_AVAILABLE,
-    _ImportError,
     FileTools,
     PythonTools,
     Toolkit,
+    _ImportError,
 )
 
 logger = logging.getLogger(__name__)
 
 
-def _make_smart_read_file(file_tools: "FileTools") -> Callable:
+def _make_smart_read_file(file_tools: FileTools) -> Callable:
     """Wrap FileTools.read_file so directories auto-list their contents.
 
     When the user (or the LLM) passes a directory path to read_file,

src/timmy/tools/system_tools.py

@@ -17,11 +17,11 @@ from pathlib import Path
 
 from timmy.tools._base import (
     _AGNO_TOOLS_AVAILABLE,
-    _ImportError,
     FileTools,
     PythonTools,
     ShellTools,
     Toolkit,
+    _ImportError,
 )
 from timmy.tools.file_tools import _make_smart_read_file
 

tests/dashboard/test_nexus.py

@@ -49,8 +49,10 @@ def test_nexus_chat_posts_message(client):
 
 def test_nexus_teach_stores_fact(client):
     """POST /nexus/teach should persist a fact and return confirmation."""
-    with patch("dashboard.routes.nexus.store_personal_fact") as mock_store, \
-         patch("dashboard.routes.nexus.recall_personal_facts_with_ids", return_value=[]):
+    with (
+        patch("dashboard.routes.nexus.store_personal_fact") as mock_store,
+        patch("dashboard.routes.nexus.recall_personal_facts_with_ids", return_value=[]),
+    ):
         mock_store.return_value = None
         response = client.post("/nexus/teach", data={"fact": "Timmy loves Python"})
     assert response.status_code == 200

tests/infrastructure/test_router_classifier.py

@@ -1,7 +1,5 @@
 """Tests for Qwen3 dual-model task complexity classifier."""
 
-import pytest
-
 from infrastructure.router.classifier import TaskComplexity, classify_task
 
 

tests/loop/test_loop_guard_seed.py

@@ -0,0 +1,144 @@
+"""Tests for loop_guard.seed_cycle_result and --pick mode.
+
+The seed fixes the cycle-metrics dead-pipeline bug (#1250):
+loop_guard pre-seeds cycle_result.json so cycle_retro.py can always
+resolve issue= even when the dispatcher doesn't write the file.
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+from unittest.mock import patch
+
+import pytest
+import scripts.loop_guard as lg
+
+
+@pytest.fixture(autouse=True)
+def _isolate(tmp_path, monkeypatch):
+    """Redirect loop_guard paths to tmp_path for isolation."""
+    monkeypatch.setattr(lg, "QUEUE_FILE", tmp_path / "queue.json")
+    monkeypatch.setattr(lg, "IDLE_STATE_FILE", tmp_path / "idle_state.json")
+    monkeypatch.setattr(lg, "CYCLE_RESULT_FILE", tmp_path / "cycle_result.json")
+    monkeypatch.setattr(lg, "GITEA_API", "http://test:3000/api/v1")
+    monkeypatch.setattr(lg, "REPO_SLUG", "owner/repo")
+
+
+# ── seed_cycle_result ──────────────────────────────────────────────────
+
+
+def test_seed_writes_issue_and_type(tmp_path):
+    """seed_cycle_result writes issue + type to cycle_result.json."""
+    item = {"issue": 42, "type": "bug", "title": "Fix the thing", "ready": True}
+    lg.seed_cycle_result(item)
+
+    data = json.loads((tmp_path / "cycle_result.json").read_text())
+    assert data == {"issue": 42, "type": "bug"}
+
+
+def test_seed_does_not_overwrite_existing(tmp_path):
+    """If cycle_result.json already exists, seed_cycle_result leaves it alone."""
+    existing = {"issue": 99, "type": "feature", "tests_passed": 123}
+    (tmp_path / "cycle_result.json").write_text(json.dumps(existing))
+
+    lg.seed_cycle_result({"issue": 1, "type": "bug"})
+
+    data = json.loads((tmp_path / "cycle_result.json").read_text())
+    assert data["issue"] == 99, "Existing file must not be overwritten"
+
+
+def test_seed_missing_issue_field(tmp_path):
+    """Item with no issue key — seed still writes without crashing."""
+    lg.seed_cycle_result({"type": "unknown"})
+    data = json.loads((tmp_path / "cycle_result.json").read_text())
+    assert data["issue"] is None
+
+
+def test_seed_default_type_when_absent(tmp_path):
+    """Item with no type key defaults to 'unknown'."""
+    lg.seed_cycle_result({"issue": 7})
+    data = json.loads((tmp_path / "cycle_result.json").read_text())
+    assert data["type"] == "unknown"
+
+
+def test_seed_oserror_is_graceful(tmp_path, monkeypatch, capsys):
+    """OSError during seed logs a warning but does not raise."""
+    monkeypatch.setattr(lg, "CYCLE_RESULT_FILE", tmp_path / "no_dir" / "cycle_result.json")
+
+    from pathlib import Path
+
+    def failing_mkdir(self, *args, **kwargs):
+        raise OSError("no space left")
+
+    monkeypatch.setattr(Path, "mkdir", failing_mkdir)
+
+    # Should not raise
+    lg.seed_cycle_result({"issue": 5, "type": "bug"})
+
+    captured = capsys.readouterr()
+    assert "WARNING" in captured.out
+
+
+# ── main() integration ─────────────────────────────────────────────────
+
+
+def _write_queue(tmp_path, items):
+    tmp_path.mkdir(parents=True, exist_ok=True)
+    lg.QUEUE_FILE.parent.mkdir(parents=True, exist_ok=True)
+    lg.QUEUE_FILE.write_text(json.dumps(items))
+
+
+def test_main_seeds_cycle_result_when_work_found(tmp_path, monkeypatch):
+    """main() seeds cycle_result.json with top queue item on ready queue."""
+    _write_queue(tmp_path, [{"issue": 10, "type": "feature", "ready": True}])
+    monkeypatch.setattr(lg, "_fetch_open_issue_numbers", lambda: None)
+
+    with patch.object(sys, "argv", ["loop_guard"]):
+        rc = lg.main()
+
+    assert rc == 0
+    data = json.loads((tmp_path / "cycle_result.json").read_text())
+    assert data["issue"] == 10
+
+
+def test_main_no_seed_when_queue_empty(tmp_path, monkeypatch):
+    """main() does not create cycle_result.json when queue is empty."""
+    _write_queue(tmp_path, [])
+    monkeypatch.setattr(lg, "_fetch_open_issue_numbers", lambda: None)
+
+    with patch.object(sys, "argv", ["loop_guard"]):
+        rc = lg.main()
+
+    assert rc == 1
+    assert not (tmp_path / "cycle_result.json").exists()
+
+
+def test_main_pick_mode_prints_issue(tmp_path, monkeypatch, capsys):
+    """--pick flag prints the top issue number to stdout."""
+    _write_queue(tmp_path, [{"issue": 55, "type": "bug", "ready": True}])
+    monkeypatch.setattr(lg, "_fetch_open_issue_numbers", lambda: None)
+
+    with patch.object(sys, "argv", ["loop_guard", "--pick"]):
+        rc = lg.main()
+
+    assert rc == 0
+    captured = capsys.readouterr()
+    # The issue number must appear as a line in stdout
+    lines = captured.out.strip().splitlines()
+    assert str(55) in lines
+
+
+def test_main_pick_mode_empty_queue_no_output(tmp_path, monkeypatch, capsys):
+    """--pick with empty queue exits 1, doesn't print an issue number."""
+    _write_queue(tmp_path, [])
+    monkeypatch.setattr(lg, "_fetch_open_issue_numbers", lambda: None)
+
+    with patch.object(sys, "argv", ["loop_guard", "--pick"]):
+        rc = lg.main()
+
+    assert rc == 1
+    captured = capsys.readouterr()
+    # No bare integer line printed
+    for line in captured.out.strip().splitlines():
+        assert not line.strip().isdigit(), f"Unexpected issue number in output: {line!r}"

tests/timmy/test_autoresearch.py

@@ -6,6 +6,52 @@ from unittest.mock import MagicMock, patch
 import pytest
 
 
+class TestAppleSiliconHelpers:
+    """Tests for is_apple_silicon() and _build_experiment_env()."""
+
+    def test_is_apple_silicon_true_on_arm64_darwin(self):
+        from timmy.autoresearch import is_apple_silicon
+
+        with (
+            patch("timmy.autoresearch.platform.system", return_value="Darwin"),
+            patch("timmy.autoresearch.platform.machine", return_value="arm64"),
+        ):
+            assert is_apple_silicon() is True
+
+    def test_is_apple_silicon_false_on_linux(self):
+        from timmy.autoresearch import is_apple_silicon
+
+        with (
+            patch("timmy.autoresearch.platform.system", return_value="Linux"),
+            patch("timmy.autoresearch.platform.machine", return_value="x86_64"),
+        ):
+            assert is_apple_silicon() is False
+
+    def test_build_env_auto_resolves_mlx_on_apple_silicon(self):
+        from timmy.autoresearch import _build_experiment_env
+
+        with patch("timmy.autoresearch.is_apple_silicon", return_value=True):
+            env = _build_experiment_env(dataset="tinystories", backend="auto")
+
+        assert env["AUTORESEARCH_BACKEND"] == "mlx"
+        assert env["AUTORESEARCH_DATASET"] == "tinystories"
+
+    def test_build_env_auto_resolves_cuda_on_non_apple(self):
+        from timmy.autoresearch import _build_experiment_env
+
+        with patch("timmy.autoresearch.is_apple_silicon", return_value=False):
+            env = _build_experiment_env(dataset="openwebtext", backend="auto")
+
+        assert env["AUTORESEARCH_BACKEND"] == "cuda"
+        assert env["AUTORESEARCH_DATASET"] == "openwebtext"
+
+    def test_build_env_explicit_backend_not_overridden(self):
+        from timmy.autoresearch import _build_experiment_env
+
+        env = _build_experiment_env(dataset="tinystories", backend="cpu")
+        assert env["AUTORESEARCH_BACKEND"] == "cpu"
+
+
 class TestPrepareExperiment:
     """Tests for prepare_experiment()."""
 
@@ -44,6 +90,24 @@ class TestPrepareExperiment:
 
         assert "failed" in result.lower()
 
+    def test_prepare_passes_env_to_prepare_script(self, tmp_path):
+        from timmy.autoresearch import prepare_experiment
+
+        repo_dir = tmp_path / "autoresearch"
+        repo_dir.mkdir()
+        (repo_dir / "prepare.py").write_text("pass")
+
+        with patch("timmy.autoresearch.subprocess.run") as mock_run:
+            mock_run.return_value = MagicMock(returncode=0, stdout="", stderr="")
+            prepare_experiment(tmp_path, dataset="tinystories", backend="cpu")
+
+        # The prepare.py call is the second call (first is skipped since repo exists)
+        prepare_call = mock_run.call_args
+        assert prepare_call.kwargs.get("env") is not None or prepare_call[1].get("env") is not None
+        call_kwargs = prepare_call.kwargs if prepare_call.kwargs else prepare_call[1]
+        assert call_kwargs["env"]["AUTORESEARCH_DATASET"] == "tinystories"
+        assert call_kwargs["env"]["AUTORESEARCH_BACKEND"] == "cpu"
+
 
 class TestRunExperiment:
     """Tests for run_experiment()."""
@@ -349,3 +413,107 @@ class TestSystemExperiment:
             success = exp.revert_changes()
 
         assert success is False
+
+    def test_create_branch_success(self, tmp_path):
+        from timmy.autoresearch import SystemExperiment
+
+        exp = SystemExperiment(target="x.py", workspace=tmp_path)
+        with patch("timmy.autoresearch.subprocess.run") as mock_run:
+            mock_run.return_value = MagicMock(returncode=0)
+            success = exp.create_branch("feature/test-branch")
+
+        assert success is True
+        # Verify correct git command was called
+        mock_run.assert_called_once()
+        call_args = mock_run.call_args[0][0]
+        assert "checkout" in call_args
+        assert "-b" in call_args
+        assert "feature/test-branch" in call_args
+
+    def test_create_branch_failure(self, tmp_path):
+        import subprocess
+
+        from timmy.autoresearch import SystemExperiment
+
+        exp = SystemExperiment(target="x.py", workspace=tmp_path)
+        with patch("timmy.autoresearch.subprocess.run") as mock_run:
+            mock_run.side_effect = subprocess.CalledProcessError(1, "git")
+            success = exp.create_branch("feature/test-branch")
+
+        assert success is False
+
+    def test_run_dry_run_mode(self, tmp_path):
+        """Test that run() in dry_run mode only generates hypotheses."""
+        from timmy.autoresearch import SystemExperiment
+
+        exp = SystemExperiment(target="x.py", workspace=tmp_path)
+        result = exp.run(max_iterations=3, dry_run=True, program_content="Test program")
+
+        assert result["iterations"] == 3
+        assert result["success"] is False  # No actual experiments run
+        assert len(exp.results) == 3
+        # Each result should have a hypothesis
+        for record in exp.results:
+            assert "hypothesis" in record
+
+    def test_run_with_custom_metric_fn(self, tmp_path):
+        """Test that custom metric_fn is used for metric extraction."""
+        from timmy.autoresearch import SystemExperiment
+
+        def custom_metric_fn(output: str) -> float | None:
+            match = __import__("re").search(r"custom_metric:\s*([0-9.]+)", output)
+            return float(match.group(1)) if match else None
+
+        exp = SystemExperiment(
+            target="x.py",
+            workspace=tmp_path,
+            metric="custom",
+            metric_fn=custom_metric_fn,
+        )
+
+        with patch("timmy.autoresearch.subprocess.run") as mock_run:
+            mock_run.return_value = MagicMock(
+                returncode=0,
+                stdout="custom_metric: 42.5\nother output",
+                stderr="",
+            )
+            tox_result = exp.run_tox()
+
+        assert tox_result["metric"] == pytest.approx(42.5)
+
+    def test_run_single_iteration_success(self, tmp_path):
+        """Test a successful single iteration that finds an improvement."""
+        from timmy.autoresearch import SystemExperiment
+
+        exp = SystemExperiment(target="x.py", workspace=tmp_path)
+
+        with patch("timmy.autoresearch.subprocess.run") as mock_run:
+            # Mock tox returning a passing test with metric
+            mock_run.return_value = MagicMock(
+                returncode=0,
+                stdout="10 passed in 1.23s",
+                stderr="",
+            )
+            result = exp.run(max_iterations=1, tox_env="unit")
+
+        assert result["iterations"] == 1
+        assert len(exp.results) == 1
+        assert exp.results[0]["metric"] == pytest.approx(100.0)
+
+    def test_run_stores_baseline_on_first_success(self, tmp_path):
+        """Test that baseline is set after first successful iteration."""
+        from timmy.autoresearch import SystemExperiment
+
+        exp = SystemExperiment(target="x.py", workspace=tmp_path)
+        assert exp.baseline is None
+
+        with patch("timmy.autoresearch.subprocess.run") as mock_run:
+            mock_run.return_value = MagicMock(
+                returncode=0,
+                stdout="8 passed in 1.23s",
+                stderr="",
+            )
+            exp.run(max_iterations=1)
+
+        assert exp.baseline == pytest.approx(100.0)
+        assert exp.results[0]["baseline"] is None  # First run has no baseline

tests/timmy/test_semantic_memory.py

@@ -572,7 +572,9 @@ class TestMemoryStore:
 
         mock_vector_store["store"].reset_mock()
         # Test with 'research'
-        result = memory_store(topic="Similar research", report="Similar research content", type="research")
+        result = memory_store(
+            topic="Similar research", report="Similar research content", type="research"
+        )
         assert "similar" in result.lower() or "duplicate" in result.lower()
         mock_vector_store["store"].assert_not_called()
 
@@ -600,7 +602,9 @@ class TestMemoryStore:
         valid_types = ["fact", "conversation", "document", "research"]
         for ctx_type in valid_types:
             mock_vector_store["store"].reset_mock()
-            memory_store(topic=f"Topic for {ctx_type}", report=f"Content for {ctx_type}", type=ctx_type)
+            memory_store(
+                topic=f"Topic for {ctx_type}", report=f"Content for {ctx_type}", type=ctx_type
+            )
             mock_vector_store["store"].assert_called_once()
 
     def test_memory_store_strips_report_and_adds_topic(self, mock_vector_store):

tests/timmy/test_three_strike.py

@@ -190,7 +190,7 @@ class TestThreeStrikeStore:
 
     @pytest.mark.unit
     def test_get_events_respects_limit(self, store):
-        for i in range(5):
+        for _ in range(5):
             try:
                 store.record("vlm_prompt_edit", "el")
             except ThreeStrikeError:

tests/timmy/test_three_strike_routes.py

@@ -1,11 +1,21 @@
 """Integration tests for the three-strike dashboard routes.
 
 Refs: #962
+
+Uses unique keys per test (uuid4) so parallel xdist workers and repeated
+runs never collide on shared SQLite state.
 """
 
+import uuid
+
 import pytest
 
 
+def _uid() -> str:
+    """Return a short unique suffix for test keys."""
+    return uuid.uuid4().hex[:8]
+
+
 class TestThreeStrikeRoutes:
     @pytest.mark.unit
     def test_list_strikes_returns_200(self, client):
@@ -24,9 +34,10 @@ class TestThreeStrikeRoutes:
 
     @pytest.mark.unit
     def test_record_strike_first(self, client):
+        key = f"test_btn_{_uid()}"
         response = client.post(
             "/sovereignty/three-strike/record",
-            json={"category": "vlm_prompt_edit", "key": "test_btn"},
+            json={"category": "vlm_prompt_edit", "key": key},
         )
         assert response.status_code == 200
         data = response.json()
@@ -43,14 +54,15 @@ class TestThreeStrikeRoutes:
 
     @pytest.mark.unit
     def test_third_strike_returns_409(self, client):
+        key = f"push_route_{_uid()}"
         for _ in range(2):
             client.post(
                 "/sovereignty/three-strike/record",
-                json={"category": "deployment_step", "key": "push_route_test"},
+                json={"category": "deployment_step", "key": key},
             )
         response = client.post(
             "/sovereignty/three-strike/record",
-            json={"category": "deployment_step", "key": "push_route_test"},
+            json={"category": "deployment_step", "key": key},
         )
         assert response.status_code == 409
         data = response.json()
@@ -60,7 +72,7 @@ class TestThreeStrikeRoutes:
     @pytest.mark.unit
     def test_register_automation_returns_success(self, client):
         response = client.post(
-            "/sovereignty/three-strike/deployment_step/some_key/automation",
+            f"/sovereignty/three-strike/deployment_step/auto_{_uid()}/automation",
             json={"artifact_path": "scripts/auto.sh"},
         )
         assert response.status_code == 200
@@ -68,15 +80,14 @@ class TestThreeStrikeRoutes:
 
     @pytest.mark.unit
     def test_get_events_returns_200(self, client):
+        key = f"events_{_uid()}"
         client.post(
             "/sovereignty/three-strike/record",
-            json={"category": "vlm_prompt_edit", "key": "events_test_key"},
-        )
-        response = client.get(
-            "/sovereignty/three-strike/vlm_prompt_edit/events_test_key/events"
+            json={"category": "vlm_prompt_edit", "key": key},
         )
+        response = client.get(f"/sovereignty/three-strike/vlm_prompt_edit/{key}/events")
         assert response.status_code == 200
         data = response.json()
         assert data["category"] == "vlm_prompt_edit"
-        assert data["key"] == "events_test_key"
+        assert data["key"] == key
         assert len(data["events"]) >= 1

tests/unit/test_paperclip.py

@@ -310,7 +310,9 @@ class TestResearchOrchestrator:
         mock_llm_client = MagicMock()
         mock_llm_client.completion = AsyncMock(return_value=mock_llm_response)
 
-        with patch("timmy.paperclip.google_web_search", new=AsyncMock(return_value=mock_search_results)):
+        with patch(
+            "timmy.paperclip.google_web_search", new=AsyncMock(return_value=mock_search_results)
+        ):
             with patch("timmy.paperclip.get_llm_client", return_value=mock_llm_client):
                 report = await orchestrator.run_research_pipeline("test query")
 
@@ -358,7 +360,10 @@ class TestResearchOrchestrator:
             orchestrator.run_research_pipeline = AsyncMock(return_value=mock_report)
             orchestrator.post_gitea_comment = AsyncMock()
 
-            with patch("timmy.paperclip.triage_research_report", new=AsyncMock(return_value=mock_triage_results)):
+            with patch(
+                "timmy.paperclip.triage_research_report",
+                new=AsyncMock(return_value=mock_triage_results),
+            ):
                 result = await orchestrator.run({"issue_number": 42})
 
             assert "Research complete for issue #42" in result
@@ -500,7 +505,9 @@ class TestPaperclipPoller:
 
         assert poller.client.update_task_status.call_count == 2
         poller.client.update_task_status.assert_any_call("task-1", "running")
-        poller.client.update_task_status.assert_any_call("task-1", "completed", "Research completed successfully")
+        poller.client.update_task_status.assert_any_call(
+            "task-1", "completed", "Research completed successfully"
+        )
         poller.orchestrator.run.assert_called_once_with({"issue_number": 42})
 
     @pytest.mark.asyncio

tests/unit/test_vassal_agent_health.py

@@ -336,7 +336,12 @@ async def test_check_agent_health_no_token():
     """Returns idle status gracefully when Gitea token is absent."""
     from timmy.vassal.agent_health import check_agent_health
 
-    status = await check_agent_health("claude")
+    mock_settings = MagicMock()
+    mock_settings.gitea_enabled = True
+    mock_settings.gitea_token = ""  # explicitly no token → early return
+
+    with patch("config.settings", mock_settings):
+        status = await check_agent_health("claude")
     # Should not raise; returns idle (no active issues discovered)
     assert isinstance(status, AgentStatus)
     assert status.agent == "claude"
@@ -478,7 +483,12 @@ async def test_check_agent_health_fetch_exception(monkeypatch):
 async def test_get_full_health_report_returns_both_agents():
     from timmy.vassal.agent_health import get_full_health_report
 
-    report = await get_full_health_report()
+    mock_settings = MagicMock()
+    mock_settings.gitea_enabled = False  # disabled → no network calls
+    mock_settings.gitea_token = ""
+
+    with patch("config.settings", mock_settings):
+        report = await get_full_health_report()
     agent_names = {a.agent for a in report.agents}
     assert "claude" in agent_names
     assert "kimi" in agent_names
@@ -488,7 +498,12 @@ async def test_get_full_health_report_returns_both_agents():
 async def test_get_full_health_report_structure():
     from timmy.vassal.agent_health import get_full_health_report
 
-    report = await get_full_health_report()
+    mock_settings = MagicMock()
+    mock_settings.gitea_enabled = False  # disabled → no network calls
+    mock_settings.gitea_token = ""
+
+    with patch("config.settings", mock_settings):
+        report = await get_full_health_report()
     assert isinstance(report, AgentHealthReport)
     assert len(report.agents) == 2
 

tests/unit/test_vassal_orchestration_loop.py

@@ -10,6 +10,29 @@ from timmy.vassal.orchestration_loop import VassalCycleRecord, VassalOrchestrato
 
 pytestmark = pytest.mark.unit
 
+
+# ---------------------------------------------------------------------------
+# Helpers — prevent real network calls under xdist parallel execution
+# ---------------------------------------------------------------------------
+
+
+def _disabled_settings() -> MagicMock:
+    """Settings mock with Gitea disabled — backlog + agent health skip HTTP."""
+    s = MagicMock()
+    s.gitea_enabled = False
+    s.gitea_token = ""
+    s.vassal_stuck_threshold_minutes = 120
+    return s
+
+
+def _fast_snapshot() -> MagicMock:
+    """Minimal SystemSnapshot mock — no disk warnings, Ollama not probed."""
+    snap = MagicMock()
+    snap.warnings = []
+    snap.disk.percent_used = 0.0
+    return snap
+
+
 # ---------------------------------------------------------------------------
 # VassalCycleRecord
 # ---------------------------------------------------------------------------
@@ -74,7 +97,15 @@ async def test_run_cycle_completes_without_services():
     clear_dispatch_registry()
     orch = VassalOrchestrator(cycle_interval=300)
 
-    record = await orch.run_cycle()
+    with (
+        patch("config.settings", _disabled_settings()),
+        patch(
+            "timmy.vassal.house_health.get_system_snapshot",
+            new_callable=AsyncMock,
+            return_value=_fast_snapshot(),
+        ),
+    ):
+        record = await orch.run_cycle()
 
     assert isinstance(record, VassalCycleRecord)
     assert record.cycle_id == 1
@@ -95,8 +126,16 @@ async def test_run_cycle_increments_cycle_count():
     clear_dispatch_registry()
     orch = VassalOrchestrator()
 
-    await orch.run_cycle()
-    await orch.run_cycle()
+    with (
+        patch("config.settings", _disabled_settings()),
+        patch(
+            "timmy.vassal.house_health.get_system_snapshot",
+            new_callable=AsyncMock,
+            return_value=_fast_snapshot(),
+        ),
+    ):
+        await orch.run_cycle()
+        await orch.run_cycle()
 
     assert orch.cycle_count == 2
     assert len(orch.history) == 2
@@ -109,7 +148,15 @@ async def test_get_status_after_cycle():
     clear_dispatch_registry()
     orch = VassalOrchestrator()
 
-    await orch.run_cycle()
+    with (
+        patch("config.settings", _disabled_settings()),
+        patch(
+            "timmy.vassal.house_health.get_system_snapshot",
+            new_callable=AsyncMock,
+            return_value=_fast_snapshot(),
+        ),
+    ):
+        await orch.run_cycle()
     status = orch.get_status()
 
     assert status["cycle_count"] == 1
@@ -183,10 +230,18 @@ async def test_run_cycle_records_backlog_error():
     clear_dispatch_registry()
     orch = VassalOrchestrator()
 
-    with patch(
-        "timmy.vassal.backlog.fetch_open_issues",
-        new_callable=AsyncMock,
-        side_effect=ConnectionError("gitea unreachable"),
+    with (
+        patch(
+            "timmy.vassal.backlog.fetch_open_issues",
+            new_callable=AsyncMock,
+            side_effect=ConnectionError("gitea unreachable"),
+        ),
+        patch("config.settings", _disabled_settings()),
+        patch(
+            "timmy.vassal.house_health.get_system_snapshot",
+            new_callable=AsyncMock,
+            return_value=_fast_snapshot(),
+        ),
     ):
         record = await orch.run_cycle()
 
@@ -202,10 +257,18 @@ async def test_run_cycle_records_agent_health_error():
     clear_dispatch_registry()
     orch = VassalOrchestrator()
 
-    with patch(
-        "timmy.vassal.agent_health.get_full_health_report",
-        new_callable=AsyncMock,
-        side_effect=RuntimeError("health check failed"),
+    with (
+        patch(
+            "timmy.vassal.agent_health.get_full_health_report",
+            new_callable=AsyncMock,
+            side_effect=RuntimeError("health check failed"),
+        ),
+        patch("config.settings", _disabled_settings()),
+        patch(
+            "timmy.vassal.house_health.get_system_snapshot",
+            new_callable=AsyncMock,
+            return_value=_fast_snapshot(),
+        ),
     ):
         record = await orch.run_cycle()
 
@@ -221,10 +284,13 @@ async def test_run_cycle_records_house_health_error():
     clear_dispatch_registry()
     orch = VassalOrchestrator()
 
-    with patch(
-        "timmy.vassal.house_health.get_system_snapshot",
-        new_callable=AsyncMock,
-        side_effect=OSError("disk check failed"),
+    with (
+        patch(
+            "timmy.vassal.house_health.get_system_snapshot",
+            new_callable=AsyncMock,
+            side_effect=OSError("disk check failed"),
+        ),
+        patch("config.settings", _disabled_settings()),
     ):
         record = await orch.run_cycle()
 
@@ -255,7 +321,10 @@ async def test_run_cycle_counts_dispatched_issues():
         patch(
             "timmy.vassal.backlog.fetch_open_issues",
             new_callable=AsyncMock,
-            return_value=[{"number": i, "title": f"Issue {i}", "labels": [], "assignees": []} for i in range(1, 4)],
+            return_value=[
+                {"number": i, "title": f"Issue {i}", "labels": [], "assignees": []}
+                for i in range(1, 4)
+            ],
         ),
         patch(
             "timmy.vassal.backlog.triage_issues",
@@ -291,7 +360,10 @@ async def test_run_cycle_respects_max_dispatch_cap():
         patch(
             "timmy.vassal.backlog.fetch_open_issues",
             new_callable=AsyncMock,
-            return_value=[{"number": i, "title": f"Issue {i}", "labels": [], "assignees": []} for i in range(1, 6)],
+            return_value=[
+                {"number": i, "title": f"Issue {i}", "labels": [], "assignees": []}
+                for i in range(1, 6)
+            ],
         ),
         patch(
             "timmy.vassal.backlog.triage_issues",
@@ -301,6 +373,12 @@ async def test_run_cycle_respects_max_dispatch_cap():
             "timmy.vassal.dispatch.dispatch_issue",
             new_callable=AsyncMock,
         ),
+        patch("config.settings", _disabled_settings()),
+        patch(
+            "timmy.vassal.house_health.get_system_snapshot",
+            new_callable=AsyncMock,
+            return_value=_fast_snapshot(),
+        ),
     ):
         record = await orch.run_cycle()
 
@@ -320,6 +398,8 @@ def test_resolve_interval_uses_explicit_value():
 
 def test_resolve_interval_falls_back_to_300():
     orch = VassalOrchestrator()
-    with patch("timmy.vassal.orchestration_loop.VassalOrchestrator._resolve_interval") as mock_resolve:
+    with patch(
+        "timmy.vassal.orchestration_loop.VassalOrchestrator._resolve_interval"
+    ) as mock_resolve:
         mock_resolve.return_value = 300.0
         assert orch._resolve_interval() == 300.0