fix: deep focus mode — single-problem context for Timmy (#409)

Co-authored-by: Kimi Agent <kimi@timmy.local>
Co-committed-by: Kimi Agent <kimi@timmy.local>

src/config.py

@@ -238,6 +238,10 @@ class Settings(BaseSettings):
     # Fallback to server when browser model is unavailable or too slow.
     browser_model_fallback: bool = True
 
+    # ── Deep Focus Mode ─────────────────────────────────────────────
+    # "deep" = single-problem context; "broad" = default multi-task.
+    focus_mode: Literal["deep", "broad"] = "broad"
+
     # ── Default Thinking ──────────────────────────────────────────────
     # When enabled, the agent starts an internal thought loop on server start.
     thinking_enabled: bool = True

src/timmy/cli.py

@@ -416,5 +416,40 @@ def route(
         typer.echo("→ orchestrator (no pattern match)")
 
 
+@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.'
+    ),
+    clear: bool = typer.Option(False, "--clear", "-c", help="Clear focus and return to broad mode"),
+):
+    """Set deep-focus mode on a single problem.
+
+    When focused, Timmy prioritizes the active topic in all responses
+    and deprioritizes unrelated context. Focus persists across sessions.
+
+    Examples:
+        timmy focus "three-phase loop"   # activate deep focus
+        timmy focus                       # show current focus
+        timmy focus --clear               # return to broad mode
+    """
+    from timmy.focus import focus_manager
+
+    if clear:
+        focus_manager.clear()
+        typer.echo("Focus cleared — back to broad mode.")
+        return
+
+    if topic:
+        focus_manager.set_topic(topic)
+        typer.echo(f'Deep focus activated: "{topic}"')
+    else:
+        # Show current focus status
+        if focus_manager.is_focused():
+            typer.echo(f'Deep focus: "{focus_manager.get_topic()}"')
+        else:
+            typer.echo("No active focus (broad mode).")
+
+
 def main():
     app()

src/timmy/focus.py

@@ -0,0 +1,105 @@
+"""Deep focus mode — single-problem context for Timmy.
+
+Persists focus state to a JSON file so Timmy can maintain narrow,
+deep attention on one problem across session restarts.
+
+Usage:
+    from timmy.focus import focus_manager
+
+    focus_manager.set_topic("three-phase loop")
+    topic = focus_manager.get_topic()       # "three-phase loop"
+    ctx   = focus_manager.get_focus_context()  # prompt injection string
+    focus_manager.clear()
+"""
+
+import json
+import logging
+from pathlib import Path
+
+logger = logging.getLogger(__name__)
+
+_DEFAULT_STATE_DIR = Path.home() / ".timmy"
+_STATE_FILE = "focus.json"
+
+
+class FocusManager:
+    """Manages deep-focus state with file-backed persistence."""
+
+    def __init__(self, state_dir: Path | None = None) -> None:
+        self._state_dir = state_dir or _DEFAULT_STATE_DIR
+        self._state_file = self._state_dir / _STATE_FILE
+        self._topic: str | None = None
+        self._mode: str = "broad"
+        self._load()
+
+    # ── Public API ────────────────────────────────────────────────
+
+    def get_topic(self) -> str | None:
+        """Return the current focus topic, or None if unfocused."""
+        return self._topic
+
+    def get_mode(self) -> str:
+        """Return 'deep' or 'broad'."""
+        return self._mode
+
+    def is_focused(self) -> bool:
+        """True when deep-focus is active with a topic set."""
+        return self._mode == "deep" and self._topic is not None
+
+    def set_topic(self, topic: str) -> None:
+        """Activate deep focus on a specific topic."""
+        self._topic = topic.strip()
+        self._mode = "deep"
+        self._save()
+        logger.info("Focus: deep-focus set → %r", self._topic)
+
+    def clear(self) -> None:
+        """Return to broad (unfocused) mode."""
+        old = self._topic
+        self._topic = None
+        self._mode = "broad"
+        self._save()
+        logger.info("Focus: cleared (was %r)", old)
+
+    def get_focus_context(self) -> str:
+        """Return a prompt-injection string for the current focus state.
+
+        When focused, this tells the model to prioritize the topic.
+        When broad, returns an empty string (no injection).
+        """
+        if not self.is_focused():
+            return ""
+        return (
+            f"[DEEP FOCUS MODE] You are currently in deep-focus mode on: "
+            f'"{self._topic}". '
+            f"Prioritize this topic in your responses. Surface related memories "
+            f"and prior conversation about this topic first. Deprioritize "
+            f"unrelated context. Stay focused — depth over breadth."
+        )
+
+    # ── Persistence ───────────────────────────────────────────────
+
+    def _load(self) -> None:
+        """Load focus state from disk."""
+        if not self._state_file.exists():
+            return
+        try:
+            data = json.loads(self._state_file.read_text())
+            self._topic = data.get("topic")
+            self._mode = data.get("mode", "broad")
+        except Exception as exc:
+            logger.warning("Focus: failed to load state: %s", exc)
+
+    def _save(self) -> None:
+        """Persist focus state to disk."""
+        try:
+            self._state_dir.mkdir(parents=True, exist_ok=True)
+            self._state_file.write_text(
+                json.dumps({"topic": self._topic, "mode": self._mode}, indent=2)
+            )
+        except Exception as exc:
+            logger.warning("Focus: failed to save state: %s", exc)
+
+
+# Module-level singleton
+focus_manager = FocusManager()

src/timmy/session.py

@@ -106,6 +106,9 @@ async def chat(message: str, session_id: str | None = None) -> str:
     # Pre-processing: extract user facts
     _extract_facts(message)
 
+    # Inject deep-focus context when active
+    message = _prepend_focus_context(message)
+
     # Run with session_id so Agno retrieves history from SQLite
     try:
         run = await agent.arun(message, stream=False, session_id=sid)
@@ -165,6 +168,9 @@ async def chat_with_tools(message: str, session_id: str | None = None):
 
     _extract_facts(message)
 
+    # Inject deep-focus context when active
+    message = _prepend_focus_context(message)
+
     try:
         run_output = await agent.arun(message, stream=False, session_id=sid)
         # Record Timmy response after getting it
@@ -303,6 +309,19 @@ def _extract_facts(message: str) -> None:
         logger.debug("Session: Fact extraction skipped: %s", exc)
 
 
+def _prepend_focus_context(message: str) -> str:
+    """Prepend deep-focus context to a message when focus mode is active."""
+    try:
+        from timmy.focus import focus_manager
+
+        ctx = focus_manager.get_focus_context()
+        if ctx:
+            return f"{ctx}\n\n{message}"
+    except Exception as exc:
+        logger.debug("Focus context injection skipped: %s", exc)
+    return message
+
+
 def _clean_response(text: str) -> str:
     """Remove hallucinated tool calls and chain-of-thought narration.
 

tests/timmy/test_focus.py

@@ -0,0 +1,113 @@
+"""Tests for timmy.focus — deep focus mode state management."""
+
+import json
+
+import pytest
+
+
+@pytest.fixture
+def focus_mgr(tmp_path):
+    """Create a FocusManager with a temporary state directory."""
+    from timmy.focus import FocusManager
+
+    return FocusManager(state_dir=tmp_path)
+
+
+class TestFocusManager:
+    """Unit tests for FocusManager."""
+
+    def test_default_state_is_broad(self, focus_mgr):
+        assert focus_mgr.get_mode() == "broad"
+        assert focus_mgr.get_topic() is None
+        assert not focus_mgr.is_focused()
+
+    def test_set_topic_activates_deep_focus(self, focus_mgr):
+        focus_mgr.set_topic("three-phase loop")
+        assert focus_mgr.get_topic() == "three-phase loop"
+        assert focus_mgr.get_mode() == "deep"
+        assert focus_mgr.is_focused()
+
+    def test_clear_returns_to_broad(self, focus_mgr):
+        focus_mgr.set_topic("bitcoin strategy")
+        focus_mgr.clear()
+        assert focus_mgr.get_topic() is None
+        assert focus_mgr.get_mode() == "broad"
+        assert not focus_mgr.is_focused()
+
+    def test_topic_strips_whitespace(self, focus_mgr):
+        focus_mgr.set_topic("  padded topic  ")
+        assert focus_mgr.get_topic() == "padded topic"
+
+    def test_focus_context_when_focused(self, focus_mgr):
+        focus_mgr.set_topic("memory architecture")
+        ctx = focus_mgr.get_focus_context()
+        assert "DEEP FOCUS MODE" in ctx
+        assert "memory architecture" in ctx
+
+    def test_focus_context_when_broad(self, focus_mgr):
+        assert focus_mgr.get_focus_context() == ""
+
+    def test_persistence_across_instances(self, tmp_path):
+        from timmy.focus import FocusManager
+
+        mgr1 = FocusManager(state_dir=tmp_path)
+        mgr1.set_topic("persistent problem")
+
+        # New instance should load persisted state
+        mgr2 = FocusManager(state_dir=tmp_path)
+        assert mgr2.get_topic() == "persistent problem"
+        assert mgr2.is_focused()
+
+    def test_clear_persists(self, tmp_path):
+        from timmy.focus import FocusManager
+
+        mgr1 = FocusManager(state_dir=tmp_path)
+        mgr1.set_topic("will be cleared")
+        mgr1.clear()
+
+        mgr2 = FocusManager(state_dir=tmp_path)
+        assert not mgr2.is_focused()
+        assert mgr2.get_topic() is None
+
+    def test_state_file_is_valid_json(self, tmp_path, focus_mgr):
+        focus_mgr.set_topic("json check")
+        state_file = tmp_path / "focus.json"
+        assert state_file.exists()
+        data = json.loads(state_file.read_text())
+        assert data["topic"] == "json check"
+        assert data["mode"] == "deep"
+
+    def test_missing_state_file_is_fine(self, tmp_path):
+        """FocusManager gracefully handles missing state file."""
+        from timmy.focus import FocusManager
+
+        mgr = FocusManager(state_dir=tmp_path / "nonexistent")
+        assert not mgr.is_focused()
+
+
+class TestPrependFocusContext:
+    """Tests for the session-level focus injection helper."""
+
+    def test_no_injection_when_unfocused(self, tmp_path, monkeypatch):
+        from timmy.focus import FocusManager
+
+        mgr = FocusManager(state_dir=tmp_path)
+        monkeypatch.setattr("timmy.focus.focus_manager", mgr)
+
+        from timmy.session import _prepend_focus_context
+
+        assert _prepend_focus_context("hello") == "hello"
+
+    def test_injection_when_focused(self, tmp_path, monkeypatch):
+        from timmy.focus import FocusManager
+
+        mgr = FocusManager(state_dir=tmp_path)
+        mgr.set_topic("test topic")
+        monkeypatch.setattr("timmy.focus.focus_manager", mgr)
+
+        from timmy.session import _prepend_focus_context
+
+        result = _prepend_focus_context("hello")
+        assert "DEEP FOCUS MODE" in result
+        assert "test topic" in result
+        assert result.endswith("hello")