Enforce the hard context limit on the full request, with turn rollback

docs/DESIGN.md

@@ -515,7 +515,7 @@ hard_limit_tokens = floor(model_context_tokens * 0.90)
 
 `clamp(min, max, value)` means use `value` but never lower than `min` or higher than `max`.
 
-When automatic compaction is off, a turn is refused if it would cross `hard_limit_tokens`. That pre-turn gate counts the estimated prompt (system prompt + history, including history images) plus this turn's incoming text **and** its incoming images (`IMAGE_TOKEN_ESTIMATE` per image), so an image-heavy turn cannot slip past a limit that text-only turns respect. The term is zero for text-only turns, leaving them unchanged.
+A turn is refused if it would cross `hard_limit_tokens`. That pre-turn gate counts the full estimated request — the system prompt, the live profile's encoded tool schemas (`tool_schema_tokens`), and history (including history images) — plus this turn's incoming text **and** its incoming images (`IMAGE_TOKEN_ESTIMATE` per image), so an image-heavy turn cannot slip past a limit that text-only turns respect. The image term is zero for text-only turns, leaving them unchanged. With automatic compaction on, the runtime compacts first and refuses only when compaction cannot bring the projected prompt under the limit; with it off, it refuses directly. Counting the tool schemas closes a blind spot where a request that fit under the limit on paper still overran once the schemas were serialized to the model.
 
 Note the floor case: at the 8192-token fallback, after the system prompt, tool schemas, and behavior instructions, the working prompt budget is roughly 3-4k tokens. Small-context operation is a first-class mode, not a degraded one: shorter tool results, more aggressive truncation, and no long conversational tails.
 
@@ -2075,7 +2075,9 @@ V1 compaction was deliberately simple: oldest-first truncation. Selective token-
 2. Drop the oldest non-user messages outside the recent window. An assistant tool call takes its tool-result messages with it so no orphans confuse the model.
 3. Last resort: drop the oldest user messages, always keeping the newest one.
 
-No model call is involved — compaction is deterministic by design, matching the policy-first philosophy. Model-written summaries of dropped turns were considered and deliberately omitted. `/compact auto on|off` toggles automatic compaction (`[runtime] auto_compact`, default on); with it off, a turn that would exceed the hard limit is refused with guidance instead.
+**Bounded exception — hard-limit recovery.** The protection of the in-flight exchange in pass 1 has one deliberate, narrow exception. When a request assembled *inside* the tool loop still exceeds `hard_limit_tokens` after ordinary compaction — a single in-flight tool result large enough to overrun on its own, sitting in the protected recent window — the runtime makes one last-resort recovery attempt before refusing the turn: it force-digests **every** tool result, including the protected tail (`_force_digest_all_tools`), then compacts again and proceeds only if that brought the request under the limit. This is safe for the same reason pass 1 is: a digest is a head/tail excerpt with an omission marker, and snapshot staleness checks force a fresh read before any write, so no correctness depends on the exact digested text surviving. It is gated behind `auto_compact` — it only runs when automatic compaction is on. With compaction off, the runtime never rewrites the protected tail; it refuses and rolls the turn back instead (below).
+
+No model call is involved — compaction is deterministic by design, matching the policy-first philosophy. Model-written summaries of dropped turns were considered and deliberately omitted. `/compact auto on|off` toggles automatic compaction (`[runtime] auto_compact`, default on). A turn that would exceed the hard limit is refused with guidance either way — the difference is the recovery attempt above: with compaction on the runtime tries the force-digest recovery first (and, on the in-loop gate, rolls the in-flight turn back before refusing); with it off it refuses directly. The in-loop gate mirrors the pre-turn gate: it re-checks the full request before every model call, so a turn that only overruns after several tool results accumulate is caught mid-loop, the in-flight assistant turn is rolled back (from the last assistant message to the end of history), and the turn ends cleanly rather than sending an over-limit request.
 
 Even simple truncation must preserve:
 
@@ -2458,7 +2460,7 @@ The rebuild should stay light. The goal is a reliable local harness, not a frame
 | Memory system | Behavior/project memory, proposals, and optimization move to v2. V1 only reads `AGENTS.md`. Scheduled for v0.3.0 (settled 2026-06-11). |
 | Token-budget compaction | V1 uses oldest-first truncation; selective compaction is v2. Scheduled for v0.3.0 (settled 2026-06-11). |
 | `trusted-local` profile | Deferred from v1, and deferred again at the 2026-06-11 v2 scoping. Revisit for v3. |
-| Session resume | Shipped in v0.3.0 (settled 2026-06-11): append-only JSONL transcripts at `.shellpilot/sessions/<session-id>.jsonl`, written incrementally with secrets redacted; compaction trims memory, never the transcript. `shellpilot --resume [id]` restores the latest (or named) session's history; snapshots are never restored, so read-before-write forces fresh reads. `/export` renders the transcript to markdown. Tool-call arguments are redacted recursively (matching the audit log's `_redact_value` logic, now unified in `redact_structure` in `shellpilot/memory/redaction.py`) before they reach the JSONL transcript; `/export` inherits redaction by re-reading the transcript from disk. Fixed in v0.5.2. `session_markdown` re-applies redaction at export time so transcripts written before v0.5.2 (which may contain raw secrets on disk) cannot leak through `/export`; on-disk history is deliberately left untouched. Fixed in v0.5.2 review wave. Plan state now also restores on `--resume` (v0.6.0): an `active_plan` pointer in the transcript is read at boot; if the referenced plan sidecar is live (`proposed`/`active`/`blocked`), `PlanManager.restore` reinstates it (section 11.3). **Read-side traversal guard (v0.10.1):** `SessionStore.find` now rejects any session id whose resolved parent differs from the sessions directory, closing the `--resume ../../../../etc/x` path-traversal vector; the write path was already safe via `path.stem`. **Reconciliation records:** the transcript stays append-only, so mid-turn corrections are records rather than rewrites — on load, `replace_last_message` replaces the last *assistant* record (a mid-batch decline truncates the reply's remaining tool calls, section 14.6) and `truncate_last_turn` deletes from the last assistant record to the end (a mid-tool cancel, section 31.15); a record of either kind with no assistant message present, or an unknown record kind, is ignored. |
+| Session resume | Shipped in v0.3.0 (settled 2026-06-11): append-only JSONL transcripts at `.shellpilot/sessions/<session-id>.jsonl`, written incrementally with secrets redacted; compaction trims memory, never the transcript. `shellpilot --resume [id]` restores the latest (or named) session's history; snapshots are never restored, so read-before-write forces fresh reads. `/export` renders the transcript to markdown. Tool-call arguments are redacted recursively (matching the audit log's `_redact_value` logic, now unified in `redact_structure` in `shellpilot/memory/redaction.py`) before they reach the JSONL transcript; `/export` inherits redaction by re-reading the transcript from disk. Fixed in v0.5.2. `session_markdown` re-applies redaction at export time so transcripts written before v0.5.2 (which may contain raw secrets on disk) cannot leak through `/export`; on-disk history is deliberately left untouched. Fixed in v0.5.2 review wave. Plan state now also restores on `--resume` (v0.6.0): an `active_plan` pointer in the transcript is read at boot; if the referenced plan sidecar is live (`proposed`/`active`/`blocked`), `PlanManager.restore` reinstates it (section 11.3). **Read-side traversal guard (v0.10.1):** `SessionStore.find` now rejects any session id whose resolved parent differs from the sessions directory, closing the `--resume ../../../../etc/x` path-traversal vector; the write path was already safe via `path.stem`. **Reconciliation records:** the transcript stays append-only, so mid-turn corrections are records rather than rewrites — on load, `replace_last_message` replaces the last *assistant* record (a mid-batch decline truncates the reply's remaining tool calls, section 14.6), `truncate_last_turn` deletes from the last assistant record to the end (a mid-tool cancel, section 31.15), and `discard_last_message` pops the single trailing record (a user message that was written to the transcript and then refused by the hard context-limit gate, so `--resume` does not restore a stuck user turn with no reply). A `replace_last_message`/`truncate_last_turn` record with no assistant message present, a `discard_last_message` with an empty transcript, or an unknown record kind, is ignored. |
 | Agent raw shell | Do not expose `raw_shell` as an agent tool in v1. Keep Manual Shell for direct user-controlled `shell=True`. |
 | Capability packs (Skills v2) | v0.6.0 shipped instruction-only SKILL.md discovery; v0.7.0 extends it with deterministic trigger selection, four markdown-only builtins, read-only references/templates, script manifest discovery without execution, and enriched `/skills` + `/context` visibility (section 23). |
 | Capability packs (heavier: tools/handlers/permissions) | Design later after core tools are stable. v3 candidate (2026-06-11). |

shellpilot/cli/slash.py

@@ -1077,7 +1077,7 @@ def _context(self) -> None:
             "[green]yes[/green]",
             "",
         )
-        total = snapshot.est_system_tokens + tool_tokens + history_tokens
+        total = self._runtime.estimated_prompt_tokens()
         table.add_row(
             "TOTAL",
             f"of {budget.model_context_tokens} (compact at {budget.compact_at_tokens})",

shellpilot/persistence/sessions.py

@@ -145,6 +145,14 @@ def truncate_last_turn(self) -> None:
         """
         self._append({"type": "truncate_last_turn"})
 
+    def discard_last_message(self) -> None:
+        """Record that the trailing transcript message is discarded.
+
+        Used when a user turn is recorded and then refused (e.g. hard context
+        limit) so --resume does not keep a stuck user message with no reply.
+        """
+        self._append({"type": "discard_last_message"})
+
     def _message_record(self, kind: str, message: Message) -> dict[str, Any]:
         content = redact_secrets(message.content) if self._redact else message.content
         record: dict[str, Any] = {
@@ -215,6 +223,9 @@ def load(path: Path) -> LoadedSession:
                     if messages[i].role == "assistant":
                         del messages[i:]
                         break
+            elif kind == "discard_last_message":
+                if messages:
+                    messages.pop()
             elif kind in ("message", "replace_last_message"):
                 role = record.get("role")
                 if not role:

shellpilot/runtime/conversation.py

@@ -411,18 +411,27 @@ def tool_schema_tokens(self) -> int:
             for definition in self._registry.definitions_for_profile(profile)
         )
 
+    def _message_tokens(self, message: Message) -> int:
+        """Estimate one history message the way it is serialized for the model."""
+        total = estimate_tokens(message.content)
+        total += IMAGE_TOKEN_ESTIMATE * len(message.images)
+        for call in message.tool_calls:
+            total += estimate_tokens(
+                json.dumps({"function": {"name": call.name, "arguments": call.arguments}})
+            )
+        return total
+
     def history_token_estimate(self) -> tuple[int, int]:
-        """Estimated history tokens (incl. images) and message count, counted
-        exactly as estimated_prompt_tokens does for the /context breakdown."""
-        total = 0
-        for message in self._history:
-            total += estimate_tokens(message.content)
-            total += IMAGE_TOKEN_ESTIMATE * len(message.images)
+        """Estimated history tokens (content, images, tool-call args) and count."""
+        total = sum(self._message_tokens(message) for message in self._history)
         return total, len(self._history)
 
     def estimated_prompt_tokens(self) -> int:
+        """Complete request estimate: system + tool schemas + history."""
         history_tokens, _ = self.history_token_estimate()
-        return self._context_snapshot().est_system_tokens + history_tokens
+        return (
+            self._context_snapshot().est_system_tokens + self.tool_schema_tokens() + history_tokens
+        )
 
     def status(self) -> RuntimeStatus:
         return RuntimeStatus(
@@ -448,12 +457,9 @@ def compact_now(self) -> int:
         metadata live outside history and are never touched.
         """
         changed = 0
-        # System context is invariant during compaction (only self._history
-        # mutates), so estimate it once and re-test only the changing history sum.
-        system_tokens = self._context_snapshot().est_system_tokens
 
         def over() -> bool:
-            return system_tokens + self.history_token_estimate()[0] > self.budget.compact_at_tokens
+            return self.estimated_prompt_tokens() > self.budget.compact_at_tokens
 
         # Digestion may reach everything except the in-flight exchange (last 2
         # messages); snapshot staleness checks still force a fresh read before
@@ -490,6 +496,65 @@ def over() -> bool:
             changed += 1
         return changed
 
+    def _ensure_under_hard_limit(self) -> bool:
+        """Compact at the soft threshold when enabled; refuse past the hard limit."""
+        if (
+            self._settings.runtime.auto_compact
+            and self.estimated_prompt_tokens() > self.budget.compact_at_tokens
+        ):
+            adjusted = self.compact_now()
+            if adjusted:
+                self._ui.show_status(f"Compacted context: adjusted {adjusted} messages.")
+        if self.estimated_prompt_tokens() <= self.budget.hard_limit_tokens:
+            return True
+        self._ui.show_status(self._hard_limit_status())
+        return False
+
+    def _hard_limit_status(self) -> str:
+        """Status text when the hard context limit blocks a model call."""
+        if not self._history and self.estimated_prompt_tokens() > self.budget.hard_limit_tokens:
+            return (
+                "System prompt and tool schemas alone exceed the hard limit. "
+                "Raise context.model_context_tokens, or trim tools/AGENTS.md."
+            )
+        if self._settings.runtime.auto_compact:
+            return (
+                "Context is over the hard limit even after compaction. "
+                "Run /clear, or shorten the request."
+            )
+        return (
+            "Context is over the hard limit and automatic compaction is off. "
+            "Run /compact (or /clear), or turn it back on with /compact auto on."
+        )
+
+    def _discard_last_user_message(self) -> None:
+        """Undo a user message that was recorded then refused by the hard limit."""
+        if self._history and self._history[-1].role == "user":
+            self._history.pop()
+            if self._session is not None:
+                self._session.discard_last_message()
+
+    def _force_digest_all_tools(self) -> int:
+        """Digest every tool result, including the normally protected tail."""
+        changed = 0
+        for index, message in enumerate(self._history):
+            if message.role != "tool":
+                continue
+            digest = _digest_text(message.content)
+            if digest != message.content:
+                self._history[index] = Message(role="tool", content=digest)
+                changed += 1
+        return changed
+
+    def _rollback_in_flight_turn(self) -> None:
+        """Drop from the last assistant message to the end (overflow / cancel)."""
+        for index in range(len(self._history) - 1, -1, -1):
+            if self._history[index].role == "assistant":
+                del self._history[index:]
+                if self._session is not None:
+                    self._session.truncate_last_turn()
+                return
+
     def run_turn(
         self,
         text: str,
@@ -521,16 +586,22 @@ def run_turn(
             )
             return ""
 
-        if not self._settings.runtime.auto_compact and (
+        # Compact existing history first when enabled, then preflight the
+        # incoming turn so a refused request never sticks in history/session.
+        if (
+            self._settings.runtime.auto_compact
+            and self.estimated_prompt_tokens() > self.budget.compact_at_tokens
+        ):
+            adjusted = self.compact_now()
+            if adjusted:
+                self._ui.show_status(f"Compacted context: adjusted {adjusted} messages.")
+        projected = (
             self.estimated_prompt_tokens()
             + estimate_tokens(text)
             + IMAGE_TOKEN_ESTIMATE * len(images)
-            > self.budget.hard_limit_tokens
-        ):
-            self._ui.show_status(
-                "Context is over the hard limit and automatic compaction is off. "
-                "Run /compact (or /clear), or turn it back on with /compact auto on."
-            )
+        )
+        if projected > self.budget.hard_limit_tokens:
+            self._ui.show_status(self._hard_limit_status())
             return ""
 
         started = time.monotonic()
@@ -541,10 +612,9 @@ def run_turn(
                 audit_kwargs["images"] = len(images)
             self._audit.write("user_turn", **audit_kwargs)
         self._record(user(text, images=tuple(images)))
-        if self._settings.runtime.auto_compact:
-            adjusted = self.compact_now()
-            if adjusted:
-                self._ui.show_status(f"Compacted context: adjusted {adjusted} messages.")
+        if not self._ensure_under_hard_limit():
+            self._discard_last_user_message()
+            return ""
         content = self._tool_loop().content
         self._ui.turn_finished(self._turn_stats(time.monotonic() - started))
         return content
@@ -608,6 +678,7 @@ def _tool_loop(self) -> Message:
             profile=self._settings.runtime.security_profile,
             max_result_tokens=self.budget.max_tool_prompt_tokens,
             max_total_tokens=self.budget.max_total_tool_prompt_tokens,
+            max_command_prompt_tokens=self.budget.max_command_prompt_tokens,
             max_capture_chars=self.budget.max_command_capture_chars,
             command_timeout_seconds=self._settings.runtime.command_timeout_seconds,
             ask_approval=self._ui.ask_approval,
@@ -624,6 +695,26 @@ def _tool_loop(self) -> Message:
         consecutive_malformed = 0
 
         while True:
+            if (
+                self._settings.runtime.auto_compact
+                and self.estimated_prompt_tokens() > self.budget.compact_at_tokens
+            ):
+                adjusted = self.compact_now()
+                if adjusted:
+                    self._ui.show_status(f"Compacted context: adjusted {adjusted} messages.")
+            if self.estimated_prompt_tokens() > self.budget.hard_limit_tokens:
+                # In-flight tool results sit in the protected compaction window.
+                # Only force-digest them when automatic compaction is on; with it
+                # off, refuse and roll back without silently rewriting history.
+                recovered = False
+                if self._settings.runtime.auto_compact:
+                    if self._force_digest_all_tools():
+                        self.compact_now()
+                    recovered = self.estimated_prompt_tokens() <= self.budget.hard_limit_tokens
+                if not recovered:
+                    self._ui.show_status(self._hard_limit_status())
+                    self._rollback_in_flight_turn()
+                    return Message(role="assistant", content="")
             messages = [
                 Message(role="system", content=self._system_message_text()),
                 *self._history,