fix(context): degrade the prompt visibly instead of failing the cycle (#1438) (#1440)

* wip(context): uniform degradation ladder for prompt cap overflow (#1438)

Work-in-progress snapshot committed by the operator after the agent's
session was terminated by a gateway 504. Not reviewed, not test-run.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(context): make the ladder's degradation visible, fair, and correctly named

Four defects in the WIP snapshot:

- Both degradation warnings used %-style placeholders while the project
  logs through loguru, so the alert emitted its own format string:
  "starved=%s shortfall=%d cap=%d". An alert that carries no data reads
  the same as one that never fired.
- The rung recorded on the non-strict path was "uniform_trim", but that
  branch line-trims and then drops whole sections in iteration order. It
  is now "line_trim": prompt_fit_rung must not count a positional
  mechanism and the ladder as one thing.
- The budget was a flat available//n, so a 50-char section reserved the
  same allowance as a 40,000-char one and the slack was unusable. Budgets
  are now water-filled: an entry shorter than its share keeps everything
  and the remainder is redistributed. The allocation is keyed on length,
  not order — permuting the entries permutes the budgets identically.
- A trimmed entry was cut with no marker, so the loss was visible only in
  our log and not in the artifact the model reads. Each trimmed entry now
  carries "[trimmed N chars]" inside its own budget.

Tests: _fair_budgets is pinned against permutation (the property a
tail-slice implementation fails while still "fitting"), two over-budget
entries of very different length are asserted to receive the same
allowance, and the trim note is asserted to name the loss.

Corrects one assertion rather than the code: the subagent test required
the whole returned prompt under the cap, but #1379 appends the operator
charter after the fit, deliberately outside it. The test now bounds the
capped portion and pins the charter tail explicitly.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: ozand

nanobot/agent/context.py

@@ -78,16 +78,17 @@ def build_system_prompt(
         excluded_skill_names: list[str] | None = None,
         loop_profile: bool = False,
         strict: bool | None = None,
+        degrade_on_overflow: bool = False,
     ) -> str:
         """Build the system prompt from identity, bootstrap files, skills, and memory.
 
         Prompt ordering is identity, bootstrap, active skills, skills
         catalogue, then memory. Under the cap (#1300):
 
-        * ``strict`` (default for the loop profile): only bootstrap ``## ``
+        * ``strict`` (default for the loop profile): only bootstrap ``## ```
           sections carrying :data:`DROPPABLE_MARKER` may be dropped, whole,
-          largest first. If the rest still does not fit, raise
-          :class:`SystemPromptOverflowError` — never choose survivors by position.
+          largest first. If the rest still does not fit, the prompt degrades
+          uniformly and visibly — never choose survivors by position.
         * non-strict (interactive sessions): the pre-#1300 behaviour, bootstrap
           trimmed first at complete-line boundaries, loss logged.
 
@@ -128,7 +129,9 @@ def build_system_prompt(
 
         memory = self.memory.get_memory_context(loop=loop_profile)
         sections.append(("memory", f"# Memory\n\n{memory}" if memory else ""))
-        return self._fit_system_prompt(sections, strict=strict)
+        return self._fit_system_prompt(
+            sections, strict=strict, degrade_on_overflow=degrade_on_overflow,
+        )
 
     SECTION_SEPARATOR = "\n\n---\n\n"
 
@@ -275,15 +278,96 @@ def _droppable_reserve_chars(self, sections: list[tuple[str, str]]) -> int:
         units = self._split_bootstrap_sections(sections[index][1])
         return sum(len(text) for _, text in units if self.DROPPABLE_MARKER in text)
 
-    def _fit_system_prompt(self, sections: list[tuple[str, str]], strict: bool = False) -> str:
+    @staticmethod
+    def _entry_names_only(sections: list[tuple[str, str]]) -> list[tuple[str, str]]:
+        """Keep each assembled entry's identifier while dropping its body."""
+        return [(name, f"# {name}") for name, content in sections if content]
+
+    TRIM_NOTE = "\n\n[trimmed {n} chars]"
+
+    @staticmethod
+    def _fair_budgets(lengths: list[int], available: int) -> list[int]:
+        """Split ``available`` across entries by equal share, water-filling.
+
+        An entry shorter than its share keeps all of its content, and what it
+        does not use is redistributed to the entries still over budget. A flat
+        ``available // n`` would gut a long entry to hand unusable slack to a
+        short one.
+
+        The allocation is keyed on LENGTH, never on order: permuting the input
+        permutes the output identically. That is what "no survivor chosen by
+        position" means here — a shorter entry is never cut so a longer one can
+        survive whole, and where the input arrives in the list decides nothing.
+        """
+        budgets = [0] * len(lengths)
+        pending = list(range(len(lengths)))
+        remaining = max(0, available)
+        while pending:
+            share = remaining // len(pending)
+            if share <= 0:
+                break
+            settled = [i for i in pending if lengths[i] <= share]
+            if not settled:
+                # Everyone left is over the share: they all take it, equally.
+                for i in pending:
+                    budgets[i] = share
+                break
+            for i in settled:
+                budgets[i] = lengths[i]
+                remaining -= lengths[i]
+                pending.remove(i)
+        return budgets
+
+    def _uniform_trim(
+        self,
+        sections: list[tuple[str, str]],
+        cap: int,
+    ) -> tuple[list[tuple[str, str]], int]:
+        """Trim each non-empty entry to its share of one shared budget.
+
+        No survivor is selected by position and the assembled prompt is never
+        sliced as one string. A section is an atomic entry: its contents may
+        be shortened, but it cannot be split into separately-selected pieces.
+
+        A trimmed entry carries :data:`TRIM_NOTE` inside its own budget, so the
+        loss is visible in the artifact the model reads and not only in the log
+        line we emit. Degradation nobody can see reads exactly like degradation
+        that never happened.
+        """
+        entries = [(name, content) for name, content in sections if content]
+        if not entries:
+            return [], 0
+        separator_chars = len(self.SECTION_SEPARATOR) * max(0, len(entries) - 1)
+        budgets = self._fair_budgets(
+            [len(content) for _, content in entries], max(0, cap - separator_chars)
+        )
+        trimmed: list[tuple[str, str]] = []
+        for (name, content), budget in zip(entries, budgets):
+            if len(content) <= budget:
+                trimmed.append((name, content))
+                continue
+            note = self.TRIM_NOTE.format(n=len(content) - budget)
+            kept = max(0, budget - len(note))
+            # If the note itself does not fit the budget, the body is already
+            # down to noise: keep the truncated body rather than only a note.
+            trimmed.append((name, content[:kept] + note if kept else content[:budget]))
+        shortfall = max(0, len(self._join_sections(sections)) - cap)
+        return trimmed, shortfall
+
+    def _fit_system_prompt(
+        self,
+        sections: list[tuple[str, str]],
+        strict: bool = False,
+        degrade_on_overflow: bool = False,
+    ) -> str:
         """Fit sections under the cap and record the outcome in :attr:`last_fit`.
 
         Strict (#1300): the only content the cap may remove is a bootstrap
         section the operator marked :data:`DROPPABLE_MARKER`, removed whole,
         largest first. Position never decides. If critical content still does
-        not fit, :class:`SystemPromptOverflowError` is raised — an under-specified
-        prompt is a failed build, not a quieter one. The decision recorded
-        here: the cap never drops a critical section.
+        not fit, the prompt uses the uniform degradation ladder: trim every
+        entry to the same budget, then fall back to names only. The decision
+        recorded here: the cap never drops a critical section by position.
 
         Non-strict (interactive sessions): the pre-#1300 behaviour — bootstrap
         is trimmed first at complete-line boundaries (#1191), then the later
@@ -300,6 +384,18 @@ def _fit_system_prompt(self, sections: list[tuple[str, str]], strict: bool = Fal
         fit: dict[str, Any] = {"cap": cap, "strict": strict, "dropped": []}
         joined = self._record_fit(fit, section_names, sections)
         if len(joined) <= cap:
+            occupancy = len(joined) / cap if cap else 1.0
+            fit.update(
+                rung="full",
+                shortfall=0,
+                starved=[],
+                occupancy_alert=occupancy >= 0.92,
+            )
+            if occupancy >= 0.92:
+                logger.warning(
+                    "System prompt occupancy alert: rung=full occupancy={:.1%} cap={} chars={}",
+                    occupancy, cap, len(joined),
+                )
             fit["droppable_reserve_chars"] = self._droppable_reserve_chars(sections)
             self.last_fit = fit
             return joined
@@ -310,6 +406,47 @@ def _fit_system_prompt(self, sections: list[tuple[str, str]], strict: bool = Fal
             droppable_reserve_chars = self._droppable_reserve_chars(sections)
             fit.update(dropped=dropped, droppable_reserve_chars=droppable_reserve_chars)
             self.last_fit = fit
+            if len(prompt) > cap and degrade_on_overflow:
+                shortfall = len(prompt) - cap
+                degraded, _ = self._uniform_trim(sections, cap)
+                uniform_prompt = self._record_fit(fit, section_names, degraded)
+                if len(uniform_prompt) <= cap and all(content for _, content in degraded):
+                    starved = [name for name, content in sections if len(content) > len(dict(degraded).get(name, ""))]
+                    fit.update(
+                        rung="uniform_trim",
+                        shortfall=shortfall,
+                        starved=starved,
+                        occupancy_alert=False,
+                        dropped=dropped,
+                        droppable_reserve_chars=0,
+                    )
+                    self.last_fit = fit
+                    logger.warning(
+                        "System prompt cap degradation: rung=uniform_trim "
+                        "starved={} shortfall={} cap={}",
+                        ",".join(starved) or "none", shortfall, cap,
+                    )
+                    return uniform_prompt
+
+                names_only = self._entry_names_only(sections)
+                names_prompt = self._record_fit(fit, section_names, names_only)
+                if len(names_prompt) <= cap:
+                    starved = [name for name, content in sections if content]
+                    fit.update(
+                        rung="names_only",
+                        shortfall=shortfall,
+                        starved=starved,
+                        occupancy_alert=False,
+                        dropped=dropped,
+                        droppable_reserve_chars=0,
+                    )
+                    self.last_fit = fit
+                    logger.warning(
+                        "System prompt cap degradation: rung=names_only "
+                        "starved={} shortfall={} cap={}",
+                        ",".join(starved) or "none", shortfall, cap,
+                    )
+                    return names_prompt
             if len(prompt) > cap:
                 raise SystemPromptOverflowError(
                     over_by=len(prompt) - cap, cap=cap,
@@ -348,6 +485,14 @@ def _fit_system_prompt(self, sections: list[tuple[str, str]], strict: bool = Fal
             logger.warning("System prompt cap dropped content: {}", details)
         prompt = self._record_fit(fit, section_names, sections)
         fit.update(
+            # NOT `uniform_trim`: this branch line-trims and then drops whole
+            # sections in iteration order. Labelling a positional mechanism
+            # with the ladder's name would make `prompt_fit_rung` count two
+            # different behaviours as one.
+            rung="line_trim" if dropped_chars else "full",
+            shortfall=0,
+            starved=list(dropped_chars),
+            occupancy_alert=(len(prompt) / cap >= 0.92 if cap else True),
             dropped=[{"section": n, "chars": c, "how": "line-trim"} for n, c in dropped_chars.items()],
             droppable_reserve_chars=self._droppable_reserve_chars(sections),
         )

nanobot/agent/subagent.py

@@ -704,6 +704,7 @@ def _build_subagent_prompt(self) -> str:
             prompt = builder.build_system_prompt(
                 excluded_skill_names=self._excluded_skill_names or None,
                 loop_profile=True,
+                degrade_on_overflow=True,
             )
         finally:
             self.last_prompt_fit = builder.last_fit

nanobot/runtime/bridge.py

@@ -2959,9 +2959,13 @@ async def _evaluate_candidate(cand_cycle_id: str, do_integration: bool, meas_met
             # here; a critical one that does not fit raises above instead.
             _prompt_fit = getattr(mgr, 'last_prompt_fit', None)
             if isinstance(_prompt_fit, dict):
+                _prompt_fit_rung = _prompt_fit.get('rung')
                 append_event(STATE_DIR, {
                     'phase': 'system_prompt', 'cycle_id': _cycle_id,
                     'chars': _prompt_fit.get('chars'), 'cap': _prompt_fit.get('cap'),
+                    'rung': _prompt_fit.get('rung'),
+                    'shortfall': _prompt_fit.get('shortfall', 0),
+                    'occupancy_alert': bool(_prompt_fit.get('occupancy_alert', False)),
                     # #1379: the per-section breakdown on EVERY row, not only
                     # at overflow — a legitimately empty section is 0, not
                     # absent. ``None`` only if the builder recorded nothing
@@ -3735,6 +3739,7 @@ async def _evaluate_candidate(cand_cycle_id: str, do_integration: bool, meas_met
             'subagent_task_id': locals().get('_subagent_task_id', None),
             'executor_llm_error': locals().get('_executor_llm_error_text', ''),
             'system_prompt_overflow': locals().get('_system_prompt_overflow_text', ''),
+            'prompt_fit_rung': locals().get('_prompt_fit_rung'),
             'origin_main_observed': locals().get('_origin_main_observed', locals().get('main_sha_before', ''))
         }
 
@@ -3836,6 +3841,7 @@ async def _evaluate_candidate(cand_cycle_id: str, do_integration: bool, meas_met
     _subagent_task_id = _res.get('subagent_task_id')
     _executor_llm_error_text = str(_res.get('executor_llm_error') or '')
     _system_prompt_overflow_text = str(_res.get('system_prompt_overflow') or '')
+    _prompt_fit_rung = None
     commits_pushed = cycle_commit_count if _integrated else 0
     import subprocess as _sp
 
@@ -3959,6 +3965,7 @@ async def _evaluate_candidate(cand_cycle_id: str, do_integration: bool, meas_met
         # dead executor is countable from this row alone.
         executor_llm_error=bool(_executor_llm_error_text),
         lane=req.get('lane') or None,
+        prompt_fit_rung=_res.get('prompt_fit_rung'),
     )
     # #721: post-cycle tag at the terminal HEAD, same outcome value as the
     # ledger row above. Integrated -> main_sha_after (shared checkout stayed on

nanobot/runtime/cycle_ledger.py

@@ -239,6 +239,7 @@ def record_cycle_outcome(
     verdict_reason: str | None = None,
     executor_llm_error: bool = False,
     lane: str | None = None,
+    prompt_fit_rung: str | None = None,
 ) -> None:
     """Write the terminal, exactly-once-per-cycle row with an enum ``outcome``.
 
@@ -294,6 +295,8 @@ def record_cycle_outcome(
         # written without ``lane`` (every pre-#1411 call site) is byte-
         # identical to before.
         row["lane"] = str(lane)
+    if prompt_fit_rung:
+        row["prompt_fit_rung"] = str(prompt_fit_rung)[:40]
     if files_changed is not None:
         try:
             from nanobot.runtime.demand import classify_change_tier

tests/test_context_prompt_fit.py

@@ -71,7 +71,7 @@ def test_strict_drops_only_declared_sections_largest_first_and_records_them(tmp_
 
 
 def test_strict_refuses_when_critical_sections_do_not_fit(tmp_path, monkeypatch):
-    """The decision recorded for #1300: the cap never drops a critical section."""
+    """Direct strict callers retain the old refusal contract."""
     monkeypatch.setattr(ContextBuilder, "MAX_SYSTEM_PROMPT_CHARS", 4_000)
     bootstrap = _section("Working knowledge", 40) + _section("Optional", 4, droppable=True) + _section("Standard test runner", 40)
     builder = _builder(tmp_path, bootstrap)
@@ -87,7 +87,7 @@ def test_strict_refuses_when_critical_sections_do_not_fit(tmp_path, monkeypatch)
 
 
 def test_strict_never_trims_lines_inside_a_section(tmp_path, monkeypatch):
-    """Position-based line trimming is the defect; strict mode must not fall back to it."""
+    """Position-based line trimming is the defect; direct strict mode must not fall back to it."""
     monkeypatch.setattr(ContextBuilder, "MAX_SYSTEM_PROMPT_CHARS", 3_000)
     builder = _builder(tmp_path, _section("Working knowledge", 80))
     with pytest.raises(SystemPromptOverflowError):
@@ -176,11 +176,63 @@ def test_subagent_prompt_is_strict_and_exposes_the_fit(tmp_path, monkeypatch):
     mgr.workspace = tmp_path
     mgr._excluded_skill_names = []
     mgr.system_context = "# Immutable operator charter\n\ncharter"
-    with pytest.raises(SystemPromptOverflowError):
-        mgr._build_subagent_prompt()
+    prompt = mgr._build_subagent_prompt()
     assert isinstance(mgr.last_prompt_fit, dict) and mgr.last_prompt_fit["strict"] is True
+    assert mgr.last_prompt_fit["rung"] == "uniform_trim"
+    # #1379: the operator charter is appended AFTER the fit and is deliberately
+    # outside the cap, so the capped portion is what the ladder bounds — not
+    # the returned string. Asserting the whole string would be asserting the
+    # charter away.
+    charter_tail = ContextBuilder.SECTION_SEPARATOR + mgr.system_context
+    assert prompt.endswith(charter_tail)
+    assert len(prompt) - len(charter_tail) <= 3_000
+    assert mgr.last_prompt_fit["chars"] <= 3_000
 
     monkeypatch.setenv(ContextBuilder.SYSTEM_PROMPT_CAP_ENV, "60000")
     prompt = mgr._build_subagent_prompt()
     assert prompt.endswith("# Immutable operator charter\n\ncharter") and "## Working knowledge" in prompt
     assert mgr.last_prompt_fit["dropped"] == []
+
+
+def test_fair_budgets_are_keyed_on_length_not_position():
+    """Permuting the entries must permute the budgets identically.
+
+    This is the property that separates the ladder from positional
+    truncation: where an entry sits in the list decides nothing. A tail-slice
+    implementation would pass a "does it fit" assertion but fail this one.
+    """
+    lengths = [10, 4_000, 40, 4_000]
+    budgets = ContextBuilder._fair_budgets(lengths, 2_000)
+    reversed_budgets = ContextBuilder._fair_budgets(lengths[::-1], 2_000)
+    assert budgets == reversed_budgets[::-1]
+    # The two short entries fit outright; the two long ones share what is left
+    # and receive the SAME budget as each other, not a prefix-first split.
+    assert budgets[0] == 10 and budgets[2] == 40
+    assert budgets[1] == budgets[3]
+    assert sum(budgets) <= 2_000
+
+
+def test_uniform_trim_gives_over_budget_entries_the_same_allowance(tmp_path):
+    """Two entries of very different length get the same budget when both overflow."""
+    builder = _builder(tmp_path, "")
+    sections = [("short", "s" * 5_000), ("long", "l" * 50_000)]
+    trimmed, shortfall = builder._uniform_trim(sections, 6_000)
+    kept = dict(trimmed)
+    assert len(kept["short"]) == len(kept["long"]), (
+        "a 10x length difference must not buy a larger allowance"
+    )
+    assert shortfall > 0
+    assert builder.TRIM_NOTE.split("{")[0].strip() in kept["long"], (
+        "the loss must be visible in the artifact the model reads"
+    )
+
+
+def test_trimmed_entry_reports_the_characters_it_lost(tmp_path):
+    builder = _builder(tmp_path, "")
+    sections = [("only", "x" * 10_000)]
+    trimmed, _ = builder._uniform_trim(sections, 1_000)
+    body = dict(trimmed)["only"]
+    assert len(body) <= 1_000
+    lost = 10_000 - (1_000 - len(builder.TRIM_NOTE.format(n=0)))
+    assert "[trimmed" in body and "chars]" in body
+    assert str(lost)[:2] in body, "the note names how much went, not just that something did"