feat(loop): SLM/LLM subtask complexity routing

core/loop/slm_routing.py

@@ -0,0 +1,152 @@
+"""P2-F1 (GenAI lesson 19): SLM/LLM task-complexity routing.
+
+Lesson 19: small language models (SLM — Mistral 7B, Phi-3) fit local /
+edge / low-cost niches. DeepCode already routes by reasoning effort
+(``core.providers.reasoning``) and uses a small classifier model for risk
+gating; this module adds an explicit *subtask-class* router: high-frequency,
+low-complexity subtasks (tool-result cleanup, summarization, classification)
+should ride the SLM path, while deep reasoning stays on the LLM path.
+
+Pure decision mechanism: ``route_subtask(task_class, ...) -> RoutingDecision``
+with env-tunable model overrides. No I/O.
+"""
+
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+from typing import Literal
+
+# Subtask classes with an inherent complexity tier (lesson 19: route by task
+# complexity, not by caller identity).
+SUBTASK_SIMPLE = "simple"  # classification, extraction, cleanup, formatting
+SUBTASK_MEDIUM = "medium"  # summarization, translation, structured rewrite
+SUBTASK_COMPLEX = "complex"  # planning, debugging, multi-step reasoning
+
+# Default tier per class (SLM for simple/medium; LLM for complex).
+_TIER_BY_CLASS = {
+    SUBTASK_SIMPLE: "slm",
+    SUBTASK_MEDIUM: "slm",
+    SUBTASK_COMPLEX: "llm",
+}
+
+_KNOWN_CLASSES = frozenset(_TIER_BY_CLASS)
+
+# Env override: DEEPCODE_SLM_MODEL / DEEPCODE_LLM_MODEL — the router is
+# environment-driven so deployments pick their own SLM/LLM pair.
+# DEEPCODE_SLM_ROUTING=0 disables SLM routing (everything → llm tier).
+
+
+@dataclass(frozen=True, slots=True)
+class RoutingDecision:
+    """One subtask's routing decision."""
+
+    task_class: str
+    tier: Literal["slm", "llm"]
+    model: str | None
+    reason: str
+    override: bool = False
+
+    def to_dict(self) -> dict[str, str | None | bool]:
+        return {
+            "task_class": self.task_class,
+            "tier": self.tier,
+            "model": self.model,
+            "reason": self.reason,
+            "override": self.override,
+        }
+
+
+def slm_routing_enabled() -> bool:
+    """Whether SLM routing is on (env ``DEEPCODE_SLM_ROUTING``; default on)."""
+    value = os.environ.get("DEEPCODE_SLM_ROUTING", "").strip().lower()
+    if not value:
+        return True
+    return value not in {"0", "false", "off", "no"}
+
+
+def slm_model() -> str | None:
+    """Configured SLM model id (env ``DEEPCODE_SLM_MODEL``), or None."""
+    value = os.environ.get("DEEPCODE_SLM_MODEL", "").strip()
+    return value or None
+
+
+def llm_model() -> str | None:
+    """Configured LLM model id (env ``DEEPCODE_LLM_MODEL``), or None."""
+    value = os.environ.get("DEEPCODE_LLM_MODEL", "").strip()
+    return value or None
+
+
+def route_subtask(
+    task_class: str,
+    *,
+    default_model: str | None = None,
+    slm_override: str | None = None,
+    llm_override: str | None = None,
+) -> RoutingDecision:
+    """Route one subtask to the SLM or LLM tier.
+
+    Parameters
+    ----------
+    task_class:
+        One of the ``SUBTASK_*`` constants (unknown classes default to
+        ``llm`` with a note — safer to over-provision than to under-reason).
+    default_model:
+        The session's current model; returned for the llm tier when no
+        explicit LLM override is set.
+    slm_override / llm_override:
+        Explicit model ids (win over env; env wins over None).
+    """
+    task_class = str(task_class or "").strip()
+    if task_class not in _KNOWN_CLASSES:
+        return RoutingDecision(
+            task_class=task_class or "unknown",
+            tier="llm",
+            model=llm_override or llm_model() or default_model,
+            reason=f"unknown task class {task_class!r}; defaulting to LLM",
+        )
+    if not slm_routing_enabled():
+        return RoutingDecision(
+            task_class=task_class,
+            tier="llm",
+            model=llm_override or llm_model() or default_model,
+            reason="SLM routing disabled (DEEPCODE_SLM_ROUTING=0)",
+            override=True,
+        )
+    tier = _TIER_BY_CLASS[task_class]
+    if tier == "slm":
+        model = slm_override or slm_model() or None
+        if model is None:
+            return RoutingDecision(
+                task_class=task_class,
+                tier="llm",
+                model=llm_override or llm_model() or default_model,
+                reason=(
+                    "SLM tier requested but DEEPCODE_SLM_MODEL unset; "
+                    "falling back to LLM"
+                ),
+            )
+        return RoutingDecision(
+            task_class=task_class,
+            tier="slm",
+            model=model,
+            reason=f"{task_class} subtask is low-complexity; routing to SLM",
+        )
+    return RoutingDecision(
+        task_class=task_class,
+        tier="llm",
+        model=llm_override or llm_model() or default_model,
+        reason=f"{task_class} subtask needs deep reasoning; routing to LLM",
+    )
+
+
+__all__ = [
+    "SUBTASK_COMPLEX",
+    "SUBTASK_MEDIUM",
+    "SUBTASK_SIMPLE",
+    "RoutingDecision",
+    "llm_model",
+    "route_subtask",
+    "slm_model",
+    "slm_routing_enabled",
+]

core/loop/slm_tasks.py

@@ -0,0 +1,102 @@
+"""P3-B (GenAI lesson 19): first consumer of the SLM subtask router.
+
+P2-F1 (:mod:`core.loop.slm_routing`) introduced a pure decision mechanism —
+``route_subtask(task_class, ...)`` — with no I/O and, until now, no caller.
+Lesson 19 says high-frequency, low-complexity subtasks (tool-result cleanup,
+summarization, classification) belong on the SLM tier. This module is the
+first real consumer: it turns that decision into a preview-shaping policy
+for oversized tool results.
+
+Scope
+-----
+A true SLM *generation* call needs a separate provider channel, which the
+runner does not carry. So the consumer here is deliberately decision-only
+(zero network, zero async): when the router says the cleanup is an SLM-grade
+task, the persisted-tool-result preview is shaped as a *clean*, dense digest
+of the head of the payload (structured drop of noise, keeps signal); when it
+is an LLM-grade task the raw truncated preview is kept as-is. Both shapes
+stay side-effect free and cheap; wiring an actual SLM generation call into
+this decision is left to a deployment that provides an SLM channel.
+
+Enable: follows ``DEEPCODE_SLM_ROUTING`` (see :mod:`core.loop.slm_routing`).
+"""
+
+from __future__ import annotations
+
+import re
+
+from core.loop.slm_routing import (
+    SUBTASK_MEDIUM,
+    route_subtask,
+    slm_routing_enabled,
+)
+
+# Maximum preview length after SLM-grade cleanup shaping.
+_CLEAN_PREVIEW_CHARS = 600
+
+# Lines that carry little decision signal in raw tool output (noise gutters).
+_NOISE_LINE_PATTERN = re.compile(
+    r"^\s*(?:ok|true|done|success|\[\s*\]|—+|-+|=+|\*+|#+|null|none)\s*$",
+    re.IGNORECASE,
+)
+
+
+def should_route_tool_cleanup() -> bool:
+    """Whether oversized tool-result cleanup should take the SLM-grade path.
+
+    Uses the subtask router's decision for the ``medium`` class
+    (summarization/cleanup). Unknown/disabled routing falls back to False so
+    the caller keeps the default raw-truncation behavior.
+    """
+    if not slm_routing_enabled():
+        return False
+    decision = route_subtask(SUBTASK_MEDIUM)
+    return decision.tier == "slm"
+
+
+def shape_slm_preview(text: str, limit: int = _CLEAN_PREVIEW_CHARS) -> str:
+    """Shrink ``text`` into a dense, noise-stripped preview.
+
+    Decision-only shaping: drops blank/noise lines and returns the surviving
+    head up to ``limit`` chars. Keeps JSON-ish payload structure by preserving
+    first-non-blank lines rather than blind character truncation.
+    """
+    if not text:
+        return text
+    lines: list[str] = []
+    for line in text.splitlines():
+        if _NOISE_LINE_PATTERN.match(line):
+            continue
+        lines.append(line)
+        if len("\n".join(lines)) >= limit:
+            break
+    preview = "\n".join(lines)[:limit]
+    return preview if preview.strip() else text[:limit]
+
+
+def choose_tool_result_preview(
+    text: str,
+    *,
+    default_preview: str,
+) -> str:
+    """Pick the preview shape for a persisted oversized tool result.
+
+    ``default_preview`` is the raw truncated head. When the SLM router says
+    cleanup belongs on the SLM tier, returns the shaped dense preview instead
+    (falls back to the default on any unexpected input).
+    """
+    if not isinstance(text, str) or not text:
+        return default_preview
+    try:
+        if should_route_tool_cleanup():
+            return shape_slm_preview(text)
+    except Exception:  # never let routing errors break result handling
+        return default_preview
+    return default_preview
+
+
+__all__ = [
+    "choose_tool_result_preview",
+    "shape_slm_preview",
+    "should_route_tool_cleanup",
+]