Add assembly cap guardrails

config.py

@@ -27,6 +27,13 @@ class LCMConfig:
     # L3 deterministic truncate token limit
     l3_truncate_tokens: int = 512
 
+    # -- Assembly guardrails ---
+    # Hard cap for the assembled active context (0 = disabled)
+    max_assembly_tokens: int = 0
+    # Reserve this many tokens from the model context window before assembly
+    # (0 = disabled). Effective cap becomes context_length - reserve_tokens_floor.
+    reserve_tokens_floor: int = 0
+
     # -- Models ---
     summary_model: str = ""       # empty = use Hermes auxiliary model
 
@@ -50,6 +57,10 @@ def from_env(cls) -> "LCMConfig":
         c.context_threshold = _float("LCM_CONTEXT_THRESHOLD", c.context_threshold)
         c.incremental_max_depth = _int("LCM_INCREMENTAL_MAX_DEPTH", c.incremental_max_depth)
         c.condensation_fanin = _int("LCM_CONDENSATION_FANIN", c.condensation_fanin)
+        c.l2_budget_ratio = _float("LCM_L2_BUDGET_RATIO", c.l2_budget_ratio)
+        c.l3_truncate_tokens = _int("LCM_L3_TRUNCATE_TOKENS", c.l3_truncate_tokens)
+        c.max_assembly_tokens = _int("LCM_MAX_ASSEMBLY_TOKENS", c.max_assembly_tokens)
+        c.reserve_tokens_floor = _int("LCM_RESERVE_TOKENS_FLOOR", c.reserve_tokens_floor)
         c.summary_model = _str("LCM_SUMMARY_MODEL", c.summary_model)
         c.database_path = _str("LCM_DATABASE_PATH", c.database_path)
         c.new_session_retain_depth = _int("LCM_NEW_SESSION_RETAIN_DEPTH", c.new_session_retain_depth)

engine.py

@@ -465,6 +465,23 @@ def _assemble_context(self, system_msg: Dict[str, Any],
             )
         result.append(sys_msg)
 
+        assembly_cap = self._effective_assembly_token_cap()
+
+        tail_selected = tail_messages
+        summary_budget = None
+        if assembly_cap is not None:
+            used = count_message_tokens(sys_msg)
+            kept_tail_reversed: list[Dict[str, Any]] = []
+            tail_token_total = 0
+            for msg in reversed(tail_messages):
+                msg_tokens = count_message_tokens(msg)
+                if used + tail_token_total + msg_tokens > assembly_cap and kept_tail_reversed:
+                    break
+                kept_tail_reversed.append(msg)
+                tail_token_total += msg_tokens
+            tail_selected = list(reversed(kept_tail_reversed))
+            summary_budget = max(0, assembly_cap - used - tail_token_total)
+
         # Collect DAG summaries — highest depth first for context hierarchy
         all_nodes = self._dag.get_session_nodes(self._session_id)
         if all_nodes:
@@ -488,17 +505,55 @@ def _assemble_context(self, system_msg: Dict[str, Any],
                     )
 
             if summary_parts:
-                combined = "\n\n---\n\n".join(summary_parts)
                 # Choose role to avoid consecutive same-role
                 last_role = result[-1].get("role", "system")
                 summary_role = "assistant" if last_role != "assistant" else "user"
-                result.append({"role": summary_role, "content": combined})
+                selected_parts = summary_parts
+                if summary_budget is not None:
+                    selected_parts = []
+                    for part in summary_parts:
+                        candidate = "\n\n---\n\n".join(selected_parts + [part])
+                        candidate_msg = {"role": summary_role, "content": candidate}
+                        if count_message_tokens(candidate_msg) > summary_budget:
+                            break
+                        selected_parts.append(part)
+                if selected_parts:
+                    combined = "\n\n---\n\n".join(selected_parts)
+                    result.append({"role": summary_role, "content": combined})
 
         # Fresh tail
-        result.extend(tail_messages)
+        result.extend(tail_selected)
 
         return result
 
+    def _effective_assembly_token_cap(self) -> Optional[int]:
+        """Return the active assembly cap, if any.
+
+        Two knobs can constrain the assembled active context:
+        - max_assembly_tokens: explicit hard cap
+        - reserve_tokens_floor: keep headroom inside context_length
+        """
+        caps: list[int] = []
+
+        if self._config.max_assembly_tokens > 0:
+            caps.append(self._config.max_assembly_tokens)
+
+        if self.context_length > 0 and self._config.reserve_tokens_floor > 0:
+            reserve_cap = self.context_length - self._config.reserve_tokens_floor
+            if reserve_cap > 0:
+                caps.append(reserve_cap)
+            else:
+                logger.warning(
+                    "LCM reserve_tokens_floor=%d disables reserve-based assembly cap because context_length=%d",
+                    self._config.reserve_tokens_floor,
+                    self.context_length,
+                )
+
+        if not caps:
+            return None
+
+        return max(1, min(caps))
+
     # -- Internal: helpers -------------------------------------------------
 
     @staticmethod

tests/test_lcm_engine.py

@@ -1,6 +1,8 @@
 """Integration tests for the LCM engine."""
 
 import json
+import logging
+import time
 import pytest
 
 from agent.context_engine import ContextEngine
@@ -372,6 +374,191 @@ def test_no_delegation_timeout(self):
         assert not hasattr(config, "delegation_timeout_ms")
 
 
+class TestAssemblyGuardrails:
+    def test_max_assembly_tokens_caps_recent_tail(self, tmp_path, monkeypatch):
+        import importlib
+
+        config = LCMConfig(
+            fresh_tail_count=10,
+            database_path=str(tmp_path / "lcm_guardrail.db"),
+            max_assembly_tokens=60,
+        )
+        instance = LCMEngine(config=config)
+        instance._session_id = "guardrail-session"
+        instance.compression_count = 1
+
+        lcm_engine_module = importlib.import_module("hermes_lcm.engine")
+        monkeypatch.setattr(
+            lcm_engine_module,
+            "count_message_tokens",
+            lambda msg: len(msg.get("content", "")),
+        )
+
+        result = instance._assemble_context(
+            {"role": "system", "content": "s" * 10},
+            [
+                {"role": "user", "content": "a" * 20},
+                {"role": "assistant", "content": "b" * 20},
+                {"role": "user", "content": "c" * 20},
+            ],
+        )
+
+        assert [msg["content"] for msg in result[1:]] == ["b" * 20, "c" * 20]
+
+    def test_reserve_tokens_floor_caps_recent_tail(self, tmp_path, monkeypatch):
+        import importlib
+
+        config = LCMConfig(
+            fresh_tail_count=10,
+            database_path=str(tmp_path / "lcm_headroom.db"),
+            reserve_tokens_floor=40,
+        )
+        instance = LCMEngine(config=config)
+        instance._session_id = "guardrail-session"
+        instance.compression_count = 1
+        instance.context_length = 100
+
+        lcm_engine_module = importlib.import_module("hermes_lcm.engine")
+        monkeypatch.setattr(
+            lcm_engine_module,
+            "count_message_tokens",
+            lambda msg: len(msg.get("content", "")),
+        )
+
+        result = instance._assemble_context(
+            {"role": "system", "content": "s" * 10},
+            [
+                {"role": "user", "content": "a" * 20},
+                {"role": "assistant", "content": "b" * 20},
+                {"role": "user", "content": "c" * 20},
+            ],
+        )
+
+        assert [msg["content"] for msg in result[1:]] == ["b" * 20, "c" * 20]
+
+    def test_max_assembly_tokens_keeps_tail_contiguous_with_varied_sizes(self, tmp_path, monkeypatch):
+        import importlib
+
+        config = LCMConfig(
+            fresh_tail_count=10,
+            database_path=str(tmp_path / "lcm_guardrail_varied.db"),
+            max_assembly_tokens=70,
+        )
+        instance = LCMEngine(config=config)
+        instance._session_id = "guardrail-session"
+        instance.compression_count = 1
+
+        lcm_engine_module = importlib.import_module("hermes_lcm.engine")
+        monkeypatch.setattr(
+            lcm_engine_module,
+            "count_message_tokens",
+            lambda msg: len(msg.get("content", "")),
+        )
+
+        result = instance._assemble_context(
+            {"role": "system", "content": "s" * 10},
+            [
+                {"role": "user", "content": "a" * 10},
+                {"role": "assistant", "content": "b" * 45},
+                {"role": "user", "content": "c" * 20},
+            ],
+        )
+
+        assert [msg["content"] for msg in result[1:]] == ["c" * 20]
+
+    def test_summary_budget_keeps_summary_order_contiguous(self, tmp_path, monkeypatch):
+        import importlib
+        from hermes_lcm.dag import SummaryNode
+
+        config = LCMConfig(
+            fresh_tail_count=10,
+            database_path=str(tmp_path / "lcm_guardrail_summary.db"),
+            max_assembly_tokens=189,
+        )
+        instance = LCMEngine(config=config)
+        instance._session_id = "guardrail-session"
+        instance.compression_count = 1
+
+        lcm_engine_module = importlib.import_module("hermes_lcm.engine")
+        monkeypatch.setattr(
+            lcm_engine_module,
+            "count_message_tokens",
+            lambda msg: len(msg.get("content", "")),
+        )
+
+        instance._dag.add_node(SummaryNode(
+            session_id="guardrail-session", depth=2,
+            summary="A" * 15, token_count=15,
+            source_token_count=100, source_ids=[],
+            source_type="messages", created_at=time.time(),
+        ))
+        instance._dag.add_node(SummaryNode(
+            session_id="guardrail-session", depth=1,
+            summary="B" * 120, token_count=120,
+            source_token_count=200, source_ids=[],
+            source_type="messages", created_at=time.time(),
+        ))
+        instance._dag.add_node(SummaryNode(
+            session_id="guardrail-session", depth=0,
+            summary="C" * 10, token_count=10,
+            source_token_count=80, source_ids=[],
+            source_type="messages", created_at=time.time(),
+        ))
+
+        result = instance._assemble_context(
+            {"role": "system", "content": "s" * 10},
+            [{"role": "user", "content": "tail" * 10}],
+        )
+
+        assert len(result) == 3
+        summary_blob = result[1]["content"]
+        assert "A" * 15 in summary_blob
+        assert "B" * 120 not in summary_blob
+        assert "C" * 10 not in summary_blob
+
+    def test_max_assembly_tokens_keeps_newest_tail_message_even_if_it_alone_exceeds_cap(self, tmp_path, monkeypatch):
+        import importlib
+
+        config = LCMConfig(
+            fresh_tail_count=10,
+            database_path=str(tmp_path / "lcm_guardrail_newest.db"),
+            max_assembly_tokens=50,
+        )
+        instance = LCMEngine(config=config)
+        instance._session_id = "guardrail-session"
+        instance.compression_count = 1
+
+        lcm_engine_module = importlib.import_module("hermes_lcm.engine")
+        monkeypatch.setattr(
+            lcm_engine_module,
+            "count_message_tokens",
+            lambda msg: len(msg.get("content", "")),
+        )
+
+        result = instance._assemble_context(
+            {"role": "system", "content": "s" * 10},
+            [
+                {"role": "user", "content": "a" * 20},
+                {"role": "assistant", "content": "b" * 60},
+            ],
+        )
+
+        assert [msg["content"] for msg in result[1:]] == ["b" * 60]
+
+    def test_reserve_tokens_floor_warns_when_misconfigured(self, tmp_path, caplog):
+        config = LCMConfig(
+            database_path=str(tmp_path / "lcm_guardrail_warn.db"),
+            reserve_tokens_floor=100,
+        )
+        instance = LCMEngine(config=config)
+        instance.context_length = 100
+
+        with caplog.at_level(logging.WARNING, logger="hermes_lcm.engine"):
+            assert instance._effective_assembly_token_cap() is None
+
+        assert "reserve_tokens_floor=100 disables reserve-based assembly cap" in caplog.text
+
+
 class TestEngineTools:
     def test_handle_grep(self, engine):
         # Add some data