diff --git a/src/config.py b/src/config.py index 1c5bfc4..7ba85b3 100644 --- a/src/config.py +++ b/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 diff --git a/src/timmy/cli.py b/src/timmy/cli.py index db97527..d228991 100644 --- a/src/timmy/cli.py +++ b/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() diff --git a/src/timmy/focus.py b/src/timmy/focus.py new file mode 100644 index 0000000..9bf348f --- /dev/null +++ b/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() diff --git a/src/timmy/session.py b/src/timmy/session.py index 56be853..5a3ce23 100644 --- a/src/timmy/session.py +++ b/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. diff --git a/tests/timmy/test_focus.py b/tests/timmy/test_focus.py new file mode 100644 index 0000000..69072db --- /dev/null +++ b/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")