Compare commits

...

1 Commits

Author SHA1 Message Date
Timmy
d57d818a5c fix(#296): poka-yoke context overflow guard — 85%/95% hard thresholds
Some checks failed
Forge CI / smoke-and-build (pull_request) Failing after 1m12s
85% auto-compress, 95% block tools, /context-status (aliases: /ctx /context)
Refs #296
2026-04-14 07:37:51 -04:00
4 changed files with 42 additions and 2 deletions

View File

@@ -134,10 +134,28 @@ class ContextCompressor:
return tokens >= self.threshold_tokens
def should_compress_preflight(self, messages: List[Dict[str, Any]]) -> bool:
"""Quick pre-flight check using rough estimate (before API call)."""
rough_estimate = estimate_messages_tokens_rough(messages)
return rough_estimate >= self.threshold_tokens
WARNING_THRESHOLD = 0.85
CRITICAL_THRESHOLD = 0.95
def get_context_usage_percent(self, prompt_tokens=None):
t = prompt_tokens if prompt_tokens is not None else self.last_prompt_tokens
return min(100.0, t / self.context_length * 100) if self.context_length > 0 else 0.0
def get_usage_level(self, prompt_tokens=None):
p = self.get_context_usage_percent(prompt_tokens) / 100
return "critical" if p >= self.CRITICAL_THRESHOLD else "warning" if p >= self.WARNING_THRESHOLD else "normal"
def should_auto_compress(self, prompt_tokens=None):
t = prompt_tokens if prompt_tokens is not None else self.last_prompt_tokens
return t >= int(self.context_length * self.WARNING_THRESHOLD)
def should_block_tools(self, prompt_tokens=None):
t = prompt_tokens if prompt_tokens is not None else self.last_prompt_tokens
return t >= int(self.context_length * self.CRITICAL_THRESHOLD)
def get_status(self) -> Dict[str, Any]:
"""Get current compression status for display/logging."""
return {
@@ -146,6 +164,9 @@ class ContextCompressor:
"context_length": self.context_length,
"usage_percent": min(100, (self.last_prompt_tokens / self.context_length * 100)) if self.context_length else 0,
"compression_count": self.compression_count,
"usage_level": self.get_usage_level(),
"warning_threshold_tokens": int(self.context_length * self.WARNING_THRESHOLD),
"critical_threshold_tokens": int(self.context_length * self.CRITICAL_THRESHOLD),
}
# ------------------------------------------------------------------

18
cli.py
View File

@@ -4658,6 +4658,8 @@ def _upload_0x0st(content: str) -> str | None:
self._handle_reasoning_command(cmd_original)
elif canonical == "compress":
self._manual_compress()
elif canonical == "context-status":
self._show_context_status()
elif canonical == "usage":
self._show_usage()
elif canonical == "insights":
@@ -5474,6 +5476,22 @@ def _upload_0x0st(content: str) -> str | None:
except Exception as e:
print(f" ❌ Compression failed: {e}")
def _show_context_status(self):
if not self.agent: print("(._.) No active agent."); return
c = getattr(self.agent, "context_compressor", None)
if not c: print("(._.) No compressor."); return
from agent.model_metadata import estimate_messages_tokens_rough
s = c.get_status()
real = s.get("last_prompt_tokens", 0) or (estimate_messages_tokens_rough(self.conversation_history) if self.conversation_history else 0)
cl = s.get("context_length", 1)
pct = real/cl*100 if cl else 0
lv = s.get("usage_level", "normal")
em = {"normal":"","warning":"⚠️","critical":"🔴"}.get(lv,"")
bar = ""*int(40*pct/100)+""*(40-int(40*pct/100))
print(f"\n📊 Context: {em} {lv.upper()} | {real:,}/{cl:,} ({pct:.1f}%) | {s.get('compression_count',0)}x compressed")
print(f" [{bar}] {pct:.1f}% | Remaining: {max(0,cl-real):,}")
print(f" Thresholds: config={s.get('threshold_tokens',0):,} | warn(85%)={s.get('warning_threshold_tokens',0):,} | crit(95%)={s.get('critical_threshold_tokens',0):,}")
def _show_usage(self):
"""Show cumulative token usage for the current session."""
if not self.agent:

View File

@@ -60,6 +60,7 @@ COMMAND_REGISTRY: list[CommandDef] = [
CommandDef("branch", "Branch the current session (explore a different path)", "Session",
aliases=("fork",), args_hint="[name]"),
CommandDef("compress", "Manually compress conversation context", "Session"),
CommandDef("context-status", "Show context usage, compression history, and remaining budget", "Session", aliases=("ctx", "context")),
CommandDef("rollback", "List or restore filesystem checkpoints", "Session",
args_hint="[number]"),
CommandDef("stop", "Kill all running background processes", "Session"),

View File

@@ -5931,7 +5931,7 @@ class AIAgent:
if messages and messages[-1].get("_flush_sentinel") == _sentinel:
messages.pop()
def _compress_context(self, messages: list, system_message: str, *, approx_tokens: int = None, task_id: str = "default") -> tuple:
def _compress_context(self, messages: list, system_message: str, *, approx_tokens: int = None, task_id: str = "default", overflow_triggered: bool = False) -> tuple:
"""Compress conversation context and split the session in SQLite.
Returns: