Restore model-aware compaction headroom

Status	File	+	-
M	`src/loader/agent/loop.py`	1	1
M	`src/loader/llm/ollama.py`	2	0
M	`src/loader/runtime/capabilities.py`	54	2
M	`src/loader/runtime/compaction.py`	37	4
M	`src/loader/runtime/public_shell.py`	16	2
M	`src/loader/runtime/safeguard_services.py`	3	5
M	`tests/test_capabilities.py`	14	0
M	`tests/test_compaction.py`	41	0
M	`tests/test_runtime_public_shell.py`	31	0
M	`tests/test_safeguard_services.py`	9	0

src/loader/agent/loop.pymodified

      workflow_mode_override: str | None = None
      stream: bool = True  # Stream LLM responses for real-time output
      session_rotate_after_bytes: int = 256 * 1024
 -    session_auto_compaction_input_tokens_threshold: int = 12_000  # ~75% of default 16K context
 +    session_auto_compaction_input_tokens_threshold: int = 100_000  # Upper bound; runtime clamps to model capacity
      session_compaction_keep_last_messages: int = 4
      # Reasoning stages configuration

src/loader/llm/ollama.pymodified

              response.raise_for_status()
              self._model_details_cache = response.json()
              self._model_details_loaded_for = self.model
 +            self._capability_profile = None
          except Exception:
              self._model_details_cache = None
              self._model_details_loaded_for = self.model
 +            self._capability_profile = None
          return self._model_details_cache

src/loader/runtime/capabilities.pymodified

      return False
 +def _coerce_positive_int(value: Any) -> int | None:
 +    """Return one positive integer when the input looks numeric."""
++
 +    try:
 +        number = int(value)
 +    except (TypeError, ValueError):
 +        return None
 +    if number <= 0:
 +        return None
 +    return number
++
++
 +def _infer_context_window(model_details: dict[str, Any] | None) -> int | None:
 +    """Infer one model context window from Ollama model metadata."""
++
 +    if not isinstance(model_details, dict):
 +        return None
++
 +    candidates: list[int] = []
++
 +    details = model_details.get("details")
 +    if isinstance(details, dict):
 +        context_length = _coerce_positive_int(details.get("context_length"))
 +        if context_length is not None:
 +            candidates.append(context_length)
++
 +    model_info = model_details.get("model_info")
 +    if isinstance(model_info, dict):
 +        for key, value in model_info.items():
 +            if str(key).endswith(".context_length"):
 +                context_length = _coerce_positive_int(value)
 +                if context_length is not None:
 +                    candidates.append(context_length)
++
 +    return max(candidates) if candidates else None
++
++
  def resolve_capability_profile(
      model_name: str,
      *,
 . heuristic fallback using model details / family tokens
      """
 +    inferred_context_window = _infer_context_window(model_details)
++
      if override is not None:
 -        return override
 +        if inferred_context_window is None:
 +            return override
 +        return CapabilityProfile(
 +            model_name=override.model_name,
 +            supports_native_tools=override.supports_native_tools,
 +            supports_streaming=override.supports_streaming,
 +            context_window=inferred_context_window,
 +            preferred_tool_call_format=override.preferred_tool_call_format,
 +            verification_strictness=override.verification_strictness,
 +            notes=list(override.notes),
 +        )
      normalized = model_name.lower().strip()
      # Try full name first, then without :tag (e.g. "deepseek-r1:14b" -> "deepseek-r1")
                  model_name=model_name,
                  supports_native_tools=known.supports_native_tools,
                  supports_streaming=known.supports_streaming,
 -                context_window=known.context_window,
 +                context_window=inferred_context_window or known.context_window,
                  preferred_tool_call_format=known.preferred_tool_call_format,
                  verification_strictness=known.verification_strictness,
                  notes=list(known.notes),
          return _profile(
              model_name,
              supports_native_tools=True,
 +            context_window=inferred_context_window or 8192,
              preferred_tool_call_format="native",
              verification_strictness="standard",
              notes=["Resolved from model family heuristic."],
          return _profile(
              model_name,
              supports_native_tools=False,
 +            context_window=inferred_context_window or 8192,
              preferred_tool_call_format="json_tag",
              verification_strictness="standard",
              notes=["Resolved from conservative no-native-tools heuristic."],
      return _profile(
          model_name,
          supports_native_tools=False,
 +        context_window=inferred_context_window or 8192,
          preferred_tool_call_format="json_tag",
          verification_strictness="standard",
          notes=["Unknown model family; defaulting to safe ReAct-style tool use."],

src/loader/runtime/compaction.pymodified

  from ..llm.base import Message, Role
  DEFAULT_AUTO_COMPACTION_INPUT_TOKENS_THRESHOLD = 100_000
 +MIN_AUTO_COMPACTION_INPUT_TOKENS_THRESHOLD = 12_000
  DEFAULT_COMPACTION_KEEP_LAST_MESSAGES = 4
  DEFAULT_MAX_CHARS = 1_200
  DEFAULT_MAX_LINES = 24
      return max(1, total_chars // 4)
 +def resolve_auto_compaction_input_tokens_threshold(
 +    configured_threshold: int,
 +    *,
 +    context_window: int | None = None,
 +) -> int:
 +    """Resolve one compaction threshold from config and model context."""
++
 +    threshold = max(1, int(configured_threshold))
 +    if context_window is None or context_window <= 0:
 +        return threshold
++
 +    context_bound = max(
 +        MIN_AUTO_COMPACTION_INPUT_TOKENS_THRESHOLD,
 +        int(context_window * 0.75),
 +    )
 +    context_bound = min(DEFAULT_AUTO_COMPACTION_INPUT_TOKENS_THRESHOLD, context_bound)
 +    return min(threshold, context_bound)
++
++
  def compress_summary(
      summary: str,
      budget: SummaryCompressionBudget | None = None,
      user_messages = [
          _collapse_inline_whitespace(message.content)
          for message in messages
 -        if message.role == Role.USER and message.content.strip()
 +        if (
 +            message.role == Role.USER
 +            and message.content.strip()
 +            and not _is_compacted_context_message(message.content)
 +        )
+     ]
      assistant_messages = [
          _collapse_inline_whitespace(message.content)
          for message in messages
 -        if message.role == Role.ASSISTANT and message.content.strip()
 +        if (
 +            message.role == Role.ASSISTANT
 +            and message.content.strip()
 +            and not _is_compacted_context_message(message.content)
 +        )
+     ]
      tool_names = [
          tool_call.name
          f"- Recent user requests: {recent_requests}",
+     ]
      if previous_summary:
 -        previous_line = _collapse_inline_whitespace(previous_summary.splitlines()[0])
 -        lines.append(f"- Previously compacted context: {previous_line}")
 +        lines.append("- Previously compacted context retained.")
      lines.extend(
+         [
              f"- Newly compacted context: {len(messages)} earlier message(s) summarized.",
      pattern = re.compile(r"(?:/|\.{1,2}/|[A-Za-z0-9_.-]+/)[A-Za-z0-9_./-]+\.[A-Za-z0-9]+")
      files: list[str] = []
      for message in messages:
 +        if _is_compacted_context_message(message.content):
 +            continue
          for match in pattern.findall(message.content):
              if match not in files:
                  files.append(match)
      return " ".join(line.split())
 +def _is_compacted_context_message(content: str) -> bool:
 +    return content.lstrip().startswith("[COMPACTED CONTEXT]")
++
++
  def _truncate_line(line: str, max_chars: int) -> str:
      if max_chars <= 0 or len(line) <= max_chars:
          return line

src/loader/runtime/public_shell.pymodified

  from ..llm.base import Message, Role
  from ..tools.base import ToolRegistry
  from .capabilities import CapabilityProfile, resolve_backend_capability_profile
 +from .compaction import resolve_auto_compaction_input_tokens_threshold
  from .dod import DefinitionOfDoneStore
  from .events import AgentEvent, TurnSummary
  from .launcher import build_runtime_launcher
          runtime_owner_path=owner_metadata["owner_path"],
          rotate_after_bytes=owner.config.session_rotate_after_bytes,
          auto_compaction_input_tokens_threshold=(
 -            owner.config.session_auto_compaction_input_tokens_threshold
 +            _resolve_owner_auto_compaction_threshold(owner)
          ),
          compaction_keep_last_messages=owner.config.session_compaction_keep_last_messages,
          system_message_factory=owner._get_system_message,
          session_id=session_id,
          rotate_after_bytes=owner.config.session_rotate_after_bytes,
          auto_compaction_input_tokens_threshold=(
 -            owner.config.session_auto_compaction_input_tokens_threshold
 +            _resolve_owner_auto_compaction_threshold(owner)
          ),
          compaction_keep_last_messages=owner.config.session_compaction_keep_last_messages,
+     )
      if refresh.prompt_reset_required:
          owner._system_message = None
      owner._use_react = None
 +    if hasattr(owner, "session") and owner.session is not None:
 +        owner.session.auto_compaction_input_tokens_threshold = (
 +            _resolve_owner_auto_compaction_threshold(owner)
 +        )
      return refresh
 +def _resolve_owner_auto_compaction_threshold(owner: RuntimeShellOwner) -> int:
 +    """Clamp one owner's compaction threshold to the active model context."""
++
 +    return resolve_auto_compaction_input_tokens_threshold(
 +        owner.config.session_auto_compaction_input_tokens_threshold,
 +        context_window=owner.capability_profile.context_window,
 +    )
++
++
  def build_runtime_system_message(
      *,
      registry: ToolRegistry,

src/loader/runtime/safeguard_services.pymodified

              if isinstance(hunks, list) and self.would_duplicate_patch(file_path, hunks):
                  return True, f"Same patch already applied to: {file_path}"
 -        elif tool_name == "bash":
 -            command = arguments.get("command", "")
 -            if self.would_duplicate_command(command):
 -                return True, f"Command already executed: {command[:50]}..."
+-
 +        # Bash commands intentionally skip exact-command dedupe here.
 +        # Re-running the same shell probe after a filesystem change is often valid,
 +        # and higher-level loop detection is a safer backstop than blocking `ls`.
          return False, ""
      def record_tool_call(self, tool_name: str, arguments: dict) -> None:

tests/test_capabilities.pymodified

      assert "heuristic" in resolved.notes[0].lower()
 +def test_model_details_context_window_overrides_registry_default() -> None:
 +    resolved = resolve_capability_profile(
 +        "gpt-oss:20b",
 +        model_details={
 +            "model_info": {
 +                "gptoss.context_length": 131072,
 +            }
 +        },
 +    )
++
 +    assert resolved.context_window == 131072
 +    assert resolved.supports_native_tools
++
++
  def test_unknown_models_default_to_safe_react_profile() -> None:
      resolved = resolve_capability_profile("mystery-model")

tests/test_compaction.pymodified

  from loader.llm.base import Message, Role
  from loader.runtime.compaction import (
      SummaryCompressionBudget,
 +    build_session_summary,
      compact_session_messages,
      compress_summary,
 +    resolve_auto_compaction_input_tokens_threshold,
+ )
+     ]
      assert result.messages[0].content.startswith("[COMPACTED CONTEXT]")
      assert "Continuation instructions:" in result.messages[0].content
++
++
 +def test_build_session_summary_skips_nested_compacted_context_content() -> None:
 +    messages = [
 +        Message(
 +            role=Role.USER,
 +            content=(
 +                "[COMPACTED CONTEXT]\nConversation summary:\n"
 +                "- Scope: older work\n- Current work: old state"
 +            ),
 +        ),
 +        Message(role=Role.ASSISTANT, content="Read the chapter index."),
 +        Message(role=Role.USER, content="Update the chapter links."),
 +    ]
++
 +    summary = build_session_summary(
 +        messages,
 +        previous_summary="[COMPACTED CONTEXT]\nConversation summary:\n- Scope: older work",
 +        current_task="Repair the table of contents links",
 +    )
++
 +    assert "Recent user requests: [COMPACTED CONTEXT]" not in summary
 +    assert "Pending work: [COMPACTED CONTEXT]" not in summary
 +    assert "- Previously compacted context retained." in summary
++
++
 +def test_resolve_auto_compaction_threshold_uses_context_window_as_upper_bound() -> None:
 +    assert resolve_auto_compaction_input_tokens_threshold(
 +        100_000,
 +        context_window=131_072,
 +    ) == 98_304
 +    assert resolve_auto_compaction_input_tokens_threshold(
 +        100_000,
 +        context_window=262_144,
 +    ) == 100_000
 +    assert resolve_auto_compaction_input_tokens_threshold(
 +        100_000,
 +        context_window=8_192,
 +    ) == 12_000

tests/test_runtime_public_shell.pymodified

  from loader.agent.loop import AgentConfig
  from loader.llm.base import CompletionResponse, Message, Role, StreamChunk
 +from loader.runtime.capabilities import CapabilityProfile
  from loader.runtime.completion_trace import CompletionTraceEntry
  from loader.runtime.dod import DefinitionOfDoneStore, create_definition_of_done
  from loader.runtime.public_shell import (
      assert handle._use_react is None
 +def test_refresh_runtime_shell_capability_profile_reclamps_session_threshold(
 +    temp_dir: Path,
 +) -> None:
 +    class ProfiledBackend(ScriptedBackend):
 +        def __init__(self) -> None:
 +            super().__init__(supports_native_tools=True)
 +            self.context_window = 8192
++
 +        def capability_profile(self) -> CapabilityProfile:
 +            return CapabilityProfile(
 +                model_name="qwen3-coder:30b",
 +                supports_native_tools=True,
 +                supports_streaming=True,
 +                context_window=self.context_window,
 +                preferred_tool_call_format="native",
 +                verification_strictness="standard",
 +                notes=["scripted"],
 +            )
++
 +    backend = ProfiledBackend()
 +    handle = _runtime_handle(temp_dir, backend=backend)
++
 +    assert handle.session.auto_compaction_input_tokens_threshold == 12_000
++
 +    backend.context_window = 131_072
 +    refresh_runtime_shell_capability_profile(handle)
++
 +    assert handle.session.auto_compaction_input_tokens_threshold == 98_304
++
++
  def test_create_runtime_session_install_builds_restored_shell_state(
      temp_dir: Path,
  ) -> None:

tests/test_safeguard_services.pymodified

      assert description == "Repeating pattern detected (2x): read → grep"
 +def test_action_tracker_allows_repeated_bash_commands() -> None:
 +    tracker = ActionTracker()
 +    arguments = {"command": "ls -la ~/Loader/guides/fortran/chapters/"}
++
 +    tracker.record_tool_call("bash", arguments)
++
 +    assert tracker.check_tool_call("bash", arguments) == (False, "")
++
++
  def test_pre_action_validator_blocks_patch_without_hunks() -> None:
      validator = PreActionValidator()

tenseleyflow/loader / `20e3ff4`

10 changed files