WIP: Claude Code progress on #1346

Automated salvage commit — agent session ended (exit 124).
Work in progress, may need continuation.

pyproject.toml

@@ -167,3 +167,6 @@ directory = "htmlcov"
 
 [tool.coverage.xml]
 output = "coverage.xml"
+
+[tool.mypy]
+mypy_path = "src"

src/config.py

@@ -56,13 +56,6 @@ class Settings(BaseSettings):
     # Set to 0 to use model defaults.
     ollama_num_ctx: int = 32768
 
-    # Maximum models loaded simultaneously in Ollama — override with OLLAMA_MAX_LOADED_MODELS
-    # Set to 2 so Qwen3-8B and Qwen3-14B can stay hot concurrently (~17 GB combined).
-    # Requires Ollama ≥ 0.1.33.  Export this to the Ollama process environment:
-    #   OLLAMA_MAX_LOADED_MODELS=2 ollama serve
-    # or add it to your systemd/launchd unit before starting the harness.
-    ollama_max_loaded_models: int = 2
-
     # Fallback model chains — override with FALLBACK_MODELS / VISION_FALLBACK_MODELS
     # as comma-separated strings, e.g. FALLBACK_MODELS="qwen3:8b,qwen2.5:14b"
     # Or edit config/providers.yaml → fallback_chains for the canonical source.

src/dashboard/routes/tools.py

@@ -13,9 +13,9 @@ from timmy.tools import get_all_available_tools
 
 router = APIRouter(tags=["tools"])
 
-_AgentView = namedtuple("AgentView", ["name", "status", "tools", "stats"])
-_ToolView = namedtuple("ToolView", ["name", "description"])
-_Stats = namedtuple("Stats", ["total_calls"])
+_AgentView = namedtuple("_AgentView", ["name", "status", "tools", "stats"])
+_ToolView = namedtuple("_ToolView", ["name", "description"])
+_Stats = namedtuple("_Stats", ["total_calls"])
 
 
 def _build_agent_tools():

src/infrastructure/presence.py

@@ -48,7 +48,7 @@ def _get_familiar_state() -> dict:
 BARK_STYLES = {"speech", "thought", "whisper", "shout"}
 
 
-def produce_bark(agent_id: str, text: str, reply_to: str = None, style: str = "speech") -> dict:
+def produce_bark(agent_id: str, text: str, reply_to: str | None = None, style: str = "speech") -> dict:
     """Format a chat response as a Matrix bark message.
 
     Barks appear as floating text above agents in the Matrix 3D world with
@@ -102,7 +102,7 @@ def produce_bark(agent_id: str, text: str, reply_to: str = None, style: str = "s
 
 
 def produce_thought(
-    agent_id: str, thought_text: str, thought_id: int, chain_id: str = None
+    agent_id: str, thought_text: str, thought_id: int, chain_id: str | None = None
 ) -> dict:
     """Format a thinking engine thought as a Matrix thought message.
 

src/infrastructure/sovereignty_metrics.py

@@ -70,12 +70,12 @@ class SovereigntyAlert:
 
 
 # Graduation targets from issue #981
-GRADUATION_TARGETS = {
+GRADUATION_TARGETS: dict[str, dict[str, float]] = {
     "cache_hit_rate": {"week1": 0.10, "month1": 0.40, "month3": 0.80, "graduation": 0.90},
     "api_cost": {"week1": 1.50, "month1": 0.50, "month3": 0.10, "graduation": 0.01},
     "time_to_report": {"week1": 180.0, "month1": 30.0, "month3": 5.0, "graduation": 1.0},
     "human_involvement": {"week1": 1.0, "month1": 0.5, "month3": 0.25, "graduation": 0.0},
-    "local_artifacts": {"week1": 6, "month1": 30, "month3": 100, "graduation": 500},
+    "local_artifacts": {"week1": 6.0, "month1": 30.0, "month3": 100.0, "graduation": 500.0},
 }
 
 

src/infrastructure/visitor.py

@@ -46,12 +46,13 @@ class VisitorRegistry:
     """
 
     _instance: "VisitorRegistry | None" = None
+    _visitors: dict[str, VisitorState]
 
     def __new__(cls) -> "VisitorRegistry":
         """Singleton constructor."""
         if cls._instance is None:
             cls._instance = super().__new__(cls)
-            cls._instance._visitors: dict[str, VisitorState] = {}
+            cls._instance._visitors = {}
         return cls._instance
 
     def add(

src/infrastructure/world/hardening/recovery.py

@@ -123,10 +123,8 @@ class RecoveryManager:
         if snapshot_id is None:
             snap_data = history[0]  # most recent
         else:
-            snap_data = next(
-                (s for s in history if s["snapshot_id"] == snapshot_id),
-                None,
-            )
+            matches = [s for s in history if s["snapshot_id"] == snapshot_id]
+            snap_data = matches[0] if matches else None
 
         if snap_data is None:
             logger.warning("RecoveryManager: snapshot %s not found", snapshot_id)

src/timmy/cli.py

@@ -177,35 +177,6 @@ def think(
     timmy.print_response(f"Think carefully about: {topic}", stream=True, session_id=_CLI_SESSION_ID)
 
 
-def _read_message_input(message: list[str]) -> str:
-    """Join CLI arguments and read from stdin when appropriate."""
-    message_str = " ".join(message)
-
-    if message_str == "-" or not _is_interactive():
-        try:
-            stdin_content = sys.stdin.read().strip()
-        except (KeyboardInterrupt, EOFError):
-            stdin_content = ""
-        if stdin_content:
-            message_str = stdin_content
-        elif message_str == "-":
-            typer.echo("No input provided via stdin.", err=True)
-            raise typer.Exit(1)
-
-    return message_str
-
-
-def _resolve_session_id(session_id: str | None, new_session: bool) -> str:
-    """Return the effective session ID based on CLI flags."""
-    import uuid
-
-    if session_id is not None:
-        return session_id
-    if new_session:
-        return str(uuid.uuid4())
-    return _CLI_SESSION_ID
-
-
 @app.command()
 def chat(
     message: list[str] = typer.Argument(

src/timmy/research_triage.py

@@ -349,7 +349,7 @@ async def triage_research_report(
         logger.info("No action items extracted from research report")
         return []
 
-    results = []
+    results: list[dict] = []
     for item in items:
         if dry_run:
             results.append({"action_item": item, "gitea_issue": None})

src/timmy/session.py

@@ -249,8 +249,8 @@ class _ErrorRunOutput:
     def __init__(self, message: str):
         self.content = message
         self.status = "ERROR"
-        self.requirements = []
-        self.tools = []
+        self.requirements: list = []
+        self.tools: list = []
 
     @property
     def active_requirements(self):

src/timmy/session_logger.py

@@ -46,7 +46,7 @@ class SessionLogger:
             content: The message content
             confidence: Optional confidence score (0.0 to 1.0)
         """
-        entry = {
+        entry: dict = {
             "type": "message",
             "role": role,
             "content": content,

src/timmy/sovereignty/session_report.py

@@ -36,7 +36,7 @@ except Exception:  # ImportError or circular import during early startup
 try:
     from infrastructure.sovereignty_metrics import GRADUATION_TARGETS, get_sovereignty_store
 except Exception:
-    GRADUATION_TARGETS: dict = {}  # type: ignore[assignment]
+    GRADUATION_TARGETS: dict = {}  # type: ignore[assignment,no-redef]
     get_sovereignty_store = None  # type: ignore[assignment]
 
 logger = logging.getLogger(__name__)

tox.ini

@@ -41,6 +41,8 @@ description = Static type checking with mypy
 commands_pre =
 deps =
     mypy>=1.0.0
+    types-PyYAML
+    types-requests
 commands =
     mypy src --ignore-missing-imports --no-error-summary