From 5dcf365c438fb666219e3453e6f33ea866f830d1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8F=AD=E6=89=AC?= Date: Wed, 5 Aug 2026 17:57:07 +0800 Subject: [PATCH 1/4] feat(context): scale the budget and truncation chain to 1M windows MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P0 + P1 of the compression relaxation plan. The compression *strategy* was fine — its ratio triggers already follow the window — so this fixes the inputs it works from: the denominator, and the serial truncation chain's tightest link. P0 — the capability registry must not shrink the window. llm_context_length is a configured *budget* and the registry holds model *capability*, so min() of the two is the right semantics. The problem was trusting family-wide patterns: r"qwen" carried the 131K that was current when the row was written, so any later 1M-class model in that family ran on 13% of its window — with every compression ratio computed against that wrong denominator. The failure mode was inverted too: an unrecognised model got the full default while a recognised one got clamped by stale data. ModelCapabilities now declares whether its context_length is `authoritative`. Version-specific rows are (they still cap an oversized budget, so configuring past a real limit is still caught); family rows and the default fallback are not, and defer to the configured budget. Family rows no longer assert a length at all — a test guards the table against re-adding one. A length learned from a real response counts as authoritative. Model names always outrun a static list, so this is a deliberate stance: keeping the table current is a convenience, never a correctness requirement. That is also why no entry was invented for newer models — a made-up number is exactly the defect being removed here. Overshooting a real limit is recoverable (the provider reports overflow and recovery compresses); silently running at a fraction of the window is not, because nothing surfaces it. Status reporting now uses the engine's effective budget rather than the configured one, so the bar cannot claim a window compression is not using. P1 — the tool-result budget could only shrink. The call site used min(max_tool_result_chars, context_length // 20), which pins every large window to the 3000-char base: on 1M that is 0.3% of the window, and since the truncation chain is serial it decided what reached the model no matter how much room was left. Replaced with adaptive_tool_result_chars(), which scales both ways — 1M now yields 25K, 128K stays ~3.2K (divisor chosen to avoid regressing existing setups), 32K still contracts, bounded and monotonic throughout. Evidence and trim ceilings were also low enough to bind almost immediately on 1M (8K/50K -> 24K/120K), and shell stdout/stderr capture moved to 40K/20K because build logs routinely exceed 10K and the truncated tail is usually where the failure is. Two governance tests asserted ceiling literals; they now bound against the constants, so the contract stays "bounded and monotonic" as ceilings rise. Verification: the first sabotage run passed, which exposed two blind spots in my own tests — the shipped family rows carry no length so min() was harmless against them, and the formula was tested without its call site. Added a stale non-authoritative entry case, an authoritative/non-authoritative pair that isolates the flag, a table guard, and two tests that drive _sync_engine_runtime_budget itself. Re-running the sabotage now fails 4 tests. Suite 1374 -> 1410 passing, ruff clean. Not addressed here (P2-P4): loss-free stages before LLM summarisation, cache prefix stability vs PCD tool-set churn, and externalising long-task state. --- src/leapflow/cli/context.py | 5 +- src/leapflow/config.py | 4 +- src/leapflow/daemon/_service_helpers.py | 6 + src/leapflow/engine/context_compressor.py | 31 ++- src/leapflow/engine/context_control.py | 5 +- src/leapflow/engine/engine.py | 46 +++- src/leapflow/llm/model_capabilities.py | 42 +++- src/leapflow/tools/shell_tools.py | 8 +- tests/test_context_budget_scaling.py | 248 ++++++++++++++++++++++ tests/test_context_governance.py | 8 +- 10 files changed, 380 insertions(+), 23 deletions(-) create mode 100644 tests/test_context_budget_scaling.py diff --git a/src/leapflow/cli/context.py b/src/leapflow/cli/context.py index 92ef1d9..601ee29 100644 --- a/src/leapflow/cli/context.py +++ b/src/leapflow/cli/context.py @@ -18,6 +18,7 @@ from leapflow.platform.mock import MockBridge from leapflow.config import Settings, _build_settings_from_env from leapflow.config_loader import config_signature, load_config_bundle +from leapflow.engine.context_compressor import adaptive_tool_result_chars from leapflow.engine.engine import AgentEngine, build_default_registry from leapflow.engine.graph_planner import GraphPlanner from leapflow.engine.intent_classifier import ( @@ -671,7 +672,9 @@ def _sync_engine_runtime_budget(self, settings: Settings) -> None: return context_length = self._effective_llm_context_length(settings) - dynamic_result_budget = min(settings.max_tool_result_chars, context_length // 20) + dynamic_result_budget = adaptive_tool_result_chars( + settings.max_tool_result_chars, context_length, + ) if dynamic_result_budget != settings.max_tool_result_chars: logger.info( "Dynamic tool result budget: %d (context=%d)", diff --git a/src/leapflow/config.py b/src/leapflow/config.py index 787313a..55b1ace 100644 --- a/src/leapflow/config.py +++ b/src/leapflow/config.py @@ -372,7 +372,7 @@ class Settings: agent_validate_tool_args: bool = True # pre-execution required-argument validation + self-repair context_hard_limit_ratio: float = 0.92 context_warning_ratio: float = 0.75 - tool_evidence_max_chars: int = 1200 + tool_evidence_max_chars: int = 4000 repeated_read_limit: int = 2 long_task_convergence_round: int = 12 # Adaptive convergence ceiling: on high-difficulty tasks the effective @@ -874,7 +874,7 @@ def _build_settings_from_env( agent_validate_tool_args = os.getenv("LEAPFLOW_AGENT_VALIDATE_TOOL_ARGS", "1").strip().lower() in ("1", "true", "yes") context_hard_limit_ratio = float(os.getenv("LEAPFLOW_CONTEXT_HARD_LIMIT_RATIO", "0.92")) context_warning_ratio = float(os.getenv("LEAPFLOW_CONTEXT_WARNING_RATIO", "0.75")) - tool_evidence_max_chars = int(os.getenv("LEAPFLOW_TOOL_EVIDENCE_MAX_CHARS", "1200")) + tool_evidence_max_chars = int(os.getenv("LEAPFLOW_TOOL_EVIDENCE_MAX_CHARS", "4000")) repeated_read_limit = int(os.getenv("LEAPFLOW_REPEATED_READ_LIMIT", "2")) long_task_convergence_round = int(os.getenv("LEAPFLOW_LONG_TASK_CONVERGENCE_ROUND", "12")) convergence_round_ceiling = int(os.getenv("LEAPFLOW_CONVERGENCE_ROUND_CEILING", "40")) diff --git a/src/leapflow/daemon/_service_helpers.py b/src/leapflow/daemon/_service_helpers.py index 73495a0..6d395f2 100644 --- a/src/leapflow/daemon/_service_helpers.py +++ b/src/leapflow/daemon/_service_helpers.py @@ -56,6 +56,12 @@ def engine_context_metadata(engine: Any | None, settings: Any) -> dict[str, Any] metadata["llm_model"] = model if engine is None: return metadata + # Prefer the engine's effective budget over the configured one: an + # authoritative model capability can cap the configured value, and reporting + # the config would claim a window compression is not actually using. + effective = getattr(engine, "active_context_length", 0) + if isinstance(effective, int) and effective > 0: + metadata["llm_context_length"] = effective metadata["context_used"] = max(0, int(getattr(engine, "context_token_count", 0) or 0)) snapshot = getattr(engine, "context_budget_snapshot", {}) if callable(snapshot): diff --git a/src/leapflow/engine/context_compressor.py b/src/leapflow/engine/context_compressor.py index 265ac0b..02c6159 100644 --- a/src/leapflow/engine/context_compressor.py +++ b/src/leapflow/engine/context_compressor.py @@ -34,10 +34,39 @@ # per-model magic numbers — the formulas produce smooth curves that # work across 32K → 2M+ context windows. -_TRIM_CEILING_CHARS = 50_000 +_TRIM_CEILING_CHARS = 120_000 _TRIM_CONTEXT_DIVISOR = 50 _TRIM_BUDGET_ACTIVATION_RATIO = 0.15 +# Tool-result budget scaling. The divisor is picked so a 128K window lands near +# the historical 3000-char budget, keeping small windows behaving as before while +# large ones actually widen: 128K/40 ~ 3.2K, 1M/40 -> 25K (ceiling-bound). +_RESULT_CONTEXT_DIVISOR = 40 +_RESULT_CEILING_CHARS = 40_000 +_RESULT_FLOOR_CHARS = 800 + + +def adaptive_tool_result_chars(base: int, context_length: int) -> int: + """Return the per-tool result budget for a given context window. + + Scales in both directions, unlike a plain ``min(base, ...)`` which can only + shrink: that form pinned every large window to ``base`` (a 1M window still + truncated each tool result at 3000 chars, i.e. 0.3% of the window, making + this the tightest link in the truncation chain regardless of how much room + was left). + + Small windows still contract — a 32K model must not spend a tenth of its + window on one tool result — and the floor keeps results meaningful rather + than letting the proportion collapse to nothing. + """ + base = max(1, int(base)) + if context_length <= 0: + return base + proportional = int(context_length) // _RESULT_CONTEXT_DIVISOR + if proportional < base: + return max(_RESULT_FLOOR_CHARS, proportional) + return min(_RESULT_CEILING_CHARS, proportional) + def estimate_text_tokens(text: str) -> int: """CJK-aware token estimate for a single text string. diff --git a/src/leapflow/engine/context_control.py b/src/leapflow/engine/context_control.py index 7eb3372..57daab5 100644 --- a/src/leapflow/engine/context_control.py +++ b/src/leapflow/engine/context_control.py @@ -238,7 +238,10 @@ def _tail_preserving_drop(messages: List[Dict[str, Any]]) -> List[Dict[str, Any] return head + [notice] + tail -_EVIDENCE_CEILING_CHARS = 8_000 +# Evidence scaling. The ceiling was low enough that a 1M window hit it almost +# immediately, so evidence stopped growing with the budget long before the +# window was under pressure. +_EVIDENCE_CEILING_CHARS = 24_000 _EVIDENCE_CONTEXT_DIVISOR = 32 diff --git a/src/leapflow/engine/engine.py b/src/leapflow/engine/engine.py index ae90cf6..0629215 100644 --- a/src/leapflow/engine/engine.py +++ b/src/leapflow/engine/engine.py @@ -1738,13 +1738,45 @@ def context_budget_snapshot(self) -> dict[str, Any]: return dict(self._last_context_snapshot) def _active_context_length(self) -> int: - """Return the runtime context length for the active model/provider.""" - if self._model_capabilities is not None: - try: - return max(1, int(self._model_capabilities.resolve(self._settings.llm_model).context_length)) - except Exception: - logger.debug("model capability lookup failed", exc_info=True) - return max(1, int(self._settings.llm_context_length)) + """Return the runtime context budget for the active model/provider. + + ``llm_context_length`` is the *configured budget* and the registry holds + *model capability*, so the effective window is the smaller of the two — + but only when the registry entry actually describes this model. A + family-wide entry carries whatever the vendor's line supported when it + was written, and clamping to it silently shrank the window for every + later generation (a 1M-class model matching "qwen" ran on 131K, i.e. 13% + of its window, with every compression ratio computed against that wrong + denominator). Model names always outrun a static table, so a + non-authoritative match defers to the configured budget instead. + + Overshooting a model's real limit is recoverable: the provider reports + overflow and recovery routes it to context compression. Silently running + at a fraction of the window is not — nothing surfaces it. + """ + budget = max(1, int(getattr(self._settings, "llm_context_length", 0) or 1)) + if self._model_capabilities is None: + return budget + try: + caps = self._model_capabilities.resolve(self._settings.llm_model) + except Exception: + logger.debug("model capability lookup failed", exc_info=True) + return budget + if not getattr(caps, "authoritative", True): + return budget + known = max(1, int(getattr(caps, "context_length", 0) or 1)) + return min(budget, known) + + @property + def active_context_length(self) -> int: + """Effective context budget in use, for status reporting. + + Exposed so clients report the window compression actually runs against. + Reading ``settings.llm_context_length`` instead shows the configured + budget, which differs whenever an authoritative capability caps it — the + status bar then claims a window the engine is not using. + """ + return self._active_context_length() def _begin_turn_context(self, user_text: str) -> None: """Reset turn-scoped state and build the stable task contract.""" diff --git a/src/leapflow/llm/model_capabilities.py b/src/leapflow/llm/model_capabilities.py index 1c50ae6..53705e8 100644 --- a/src/leapflow/llm/model_capabilities.py +++ b/src/leapflow/llm/model_capabilities.py @@ -34,6 +34,15 @@ class ModelCapabilities: supports_thinking: bool = False supports_streaming_tools: bool = False tokens_per_image: int = _IMAGE_TOKEN_ESTIMATE + # Whether ``context_length`` is trustworthy enough to cap the user's + # configured budget. A version-specific entry is authoritative; a + # family-wide fallback such as "qwen" is not, because it carries whatever + # the family's largest model happened to support when the entry was written + # and would silently shrink the window for every later generation. Model + # names always outrun a static table, so an unrecognised or merely + # family-matched model defers to the configured budget instead of being + # clamped by stale data. + authoritative: bool = True @runtime_checkable @@ -43,6 +52,13 @@ class CapabilitySource(Protocol): def resolve(self, model: str, base_url: str = "") -> Optional[ModelCapabilities]: ... +# Version-specific entries are authoritative: the number belongs to that exact +# model. Family-wide entries (matching every generation of a vendor's line) are +# marked non-authoritative — they inform feature flags but must not cap a +# configured budget, since a newer generation usually has a larger window than +# whatever was current when the entry was added. New model names are expected to +# miss this table entirely and fall through to the configured budget; keeping the +# list current is a convenience, never a correctness requirement. _KNOWN_MODELS: List[tuple[str, ModelCapabilities]] = [ # OpenAI (r"gpt-4o", ModelCapabilities(context_length=128_000, max_output_tokens=16_384, @@ -64,12 +80,16 @@ def resolve(self, model: str, base_url: str = "") -> Optional[ModelCapabilities] (r"claude-4", ModelCapabilities(context_length=200_000, max_output_tokens=16_384, supports_tools=True, supports_vision=True, supports_thinking=True)), - # DeepSeek - (r"deepseek", ModelCapabilities(context_length=128_000, max_output_tokens=8_192, - supports_tools=True, supports_thinking=True)), - # Qwen - (r"qwen", ModelCapabilities(context_length=131_072, max_output_tokens=8_192, - supports_tools=True, supports_thinking=True)), + # Family-wide fallbacks: feature flags only, context length not authoritative. + (r"gpt-", ModelCapabilities(max_output_tokens=32_768, supports_tools=True, + supports_vision=True, authoritative=False)), + (r"claude-", ModelCapabilities(max_output_tokens=16_384, supports_tools=True, + supports_vision=True, supports_thinking=True, + authoritative=False)), + (r"deepseek", ModelCapabilities(max_output_tokens=8_192, supports_tools=True, + supports_thinking=True, authoritative=False)), + (r"qwen", ModelCapabilities(max_output_tokens=8_192, supports_tools=True, + supports_thinking=True, authoritative=False)), ] @@ -86,7 +106,9 @@ class ModelCapabilityRegistry: def __init__(self, *, default: Optional[ModelCapabilities] = None) -> None: self._overrides: Dict[str, ModelCapabilities] = {} self._learned: Dict[str, Dict[str, Any]] = {} - self._default = default or ModelCapabilities() + # The fallback carries no model-specific knowledge, so its context length + # must never cap a configured budget. + self._default = default or ModelCapabilities(authoritative=False) def register(self, model: str, caps: ModelCapabilities) -> None: """Register an explicit capability override for a model.""" @@ -110,6 +132,11 @@ def resolve(self, model: str, base_url: str = "") -> ModelCapabilities: supports_thinking=pattern_caps.supports_thinking, supports_streaming_tools=pattern_caps.supports_streaming_tools, tokens_per_image=pattern_caps.tokens_per_image, + # A learned length comes from a real response, so it is + # authoritative even when the pattern was only a family guess. + authoritative=( + True if "context_length" in learned else pattern_caps.authoritative + ), ) if pattern_caps: @@ -119,6 +146,7 @@ def resolve(self, model: str, base_url: str = "") -> ModelCapabilities: return ModelCapabilities( context_length=learned.get("context_length", self._default.context_length), max_output_tokens=learned.get("max_output_tokens", self._default.max_output_tokens), + authoritative="context_length" in learned, ) return self._default diff --git a/src/leapflow/tools/shell_tools.py b/src/leapflow/tools/shell_tools.py index be5bd66..b82560f 100644 --- a/src/leapflow/tools/shell_tools.py +++ b/src/leapflow/tools/shell_tools.py @@ -28,8 +28,12 @@ logger = logging.getLogger(__name__) -_MAX_STDOUT = 10_000 -_MAX_STDERR = 5_000 +# Raw capture ceilings. These bound what the tool returns before the context +# layers (evidence builder, result budget, trim) decide how much reaches the +# model. Build and test logs routinely exceed 10K, and truncating there dropped +# the tail — which is where the actual failure usually is. +_MAX_STDOUT = 40_000 +_MAX_STDERR = 20_000 _DEFAULT_TIMEOUT = 30.0 # Internal ceiling for the shell process timeout. Raised from the original # hard-coded 120 s; injectable at startup via set_max_shell_timeout so diff --git a/tests/test_context_budget_scaling.py b/tests/test_context_budget_scaling.py new file mode 100644 index 0000000..a7ad524 --- /dev/null +++ b/tests/test_context_budget_scaling.py @@ -0,0 +1,248 @@ +"""Contracts for context budget resolution and truncation-chain scaling. + +Two problems these pin down: + +- A stale registry entry must never shrink the window. Family-wide patterns like + "qwen" carry whatever the vendor's line supported when the entry was written, + so clamping to them silently runs a newer model at a fraction of its window — + and every compression ratio is then computed against the wrong denominator. + Model names always outrun a static table, so the table is advisory except for + version-specific entries. +- The truncation chain is serial, so its tightest link decides what reaches the + model. The tool-result budget used ``min(base, ...)``, which can only shrink: + a 1M window still cut every tool result at 3000 chars. +""" + +from __future__ import annotations + +from types import SimpleNamespace + +import pytest + +from leapflow.engine.context_compressor import ( + _RESULT_CEILING_CHARS, + _RESULT_FLOOR_CHARS, + adaptive_tool_result_chars, +) +from leapflow.engine.engine import AgentEngine +from leapflow.llm.model_capabilities import ModelCapabilities, ModelCapabilityRegistry + + +def _active_length(model: str, budget: int, registry=None) -> int: + """Run the real resolution logic without constructing a whole engine.""" + engine = object.__new__(AgentEngine) + engine._settings = SimpleNamespace(llm_model=model, llm_context_length=budget) + engine._model_capabilities = registry if registry is not None else ModelCapabilityRegistry() + return AgentEngine._active_context_length(engine) + + +# ── Budget resolution ──────────────────────────────────────────────────── + + +@pytest.mark.parametrize("model", ["qwen3.8-max", "qwen-next", "deepseek-v9", "claude-9-opus"]) +def test_family_fallback_does_not_shrink_the_configured_budget(model: str) -> None: + """A family pattern is not evidence about a specific model's window.""" + assert _active_length(model, 1_000_000) == 1_000_000 + + +@pytest.mark.parametrize("model", ["gpt-5.5", "gpt-6-ultra", "brand-new-model-2030"]) +def test_unrecognised_models_get_the_full_configured_budget(model: str) -> None: + """New names are expected to miss the table; that must not cost them window. + + The registry is a convenience, not a gate: keeping it current can never be a + correctness requirement because model names ship faster than this list. + """ + assert _active_length(model, 1_000_000) == 1_000_000 + + +@pytest.mark.parametrize( + ("model", "expected"), + [("gpt-3.5-turbo", 16_385), ("gpt-4o", 128_000), ("claude-4-opus", 200_000)], +) +def test_authoritative_capability_caps_an_oversized_budget(model: str, expected: int) -> None: + """Version-specific entries still guard against configuring past the limit.""" + assert _active_length(model, 1_000_000) == expected + + +def test_a_smaller_configured_budget_is_always_respected() -> None: + """Lowering the budget deliberately (e.g. to cut spend) must hold.""" + assert _active_length("gpt-4o", 32_000) == 32_000 + assert _active_length("qwen3.8-max", 200_000) == 200_000 + + +def test_a_registered_override_is_authoritative() -> None: + """An explicitly registered capability describes that exact model.""" + registry = ModelCapabilityRegistry() + registry.register("custom-model", ModelCapabilities(context_length=64_000)) + + assert _active_length("custom-model", 1_000_000, registry) == 64_000 + + +def test_a_learned_length_becomes_authoritative() -> None: + """A length observed from a real response outranks a family guess.""" + registry = ModelCapabilityRegistry() + registry.update_from_usage("qwen3.8-max", {"prompt_tokens": 300_000, "completion_tokens": 1_000}) + + caps = registry.resolve("qwen3.8-max") + + assert caps.authoritative is True + assert caps.context_length > 300_000 + + +def test_a_non_authoritative_entry_with_a_small_length_is_ignored() -> None: + """The real guard: a stale family number must not cap the budget. + + The shipped family entries deliberately carry no context length, so a plain + ``min()`` happens to be harmless against them today. This constructs the + situation the flag exists for — a family entry that does carry a (stale) + number — and pins that it cannot shrink the window. + """ + registry = ModelCapabilityRegistry() + stale = ModelCapabilities(context_length=131_072, authoritative=False) + registry._overrides["vendor-next-gen"] = stale + + assert _active_length("vendor-next-gen", 1_000_000) == 1_000_000 + + +def test_an_authoritative_entry_with_the_same_length_does_cap() -> None: + """Same number, authoritative: now it must apply. Isolates the flag itself.""" + registry = ModelCapabilityRegistry() + registry.register("pinned-model", ModelCapabilities(context_length=131_072)) + + assert _active_length("pinned-model", 1_000_000, registry) == 131_072 + + +def test_shipped_family_entries_carry_no_context_length_claim() -> None: + """Family entries must not assert a window, only feature flags. + + Guards the table itself: adding a context length to a family row would + re-create the original defect for every future model in that family. + """ + from leapflow.llm.model_capabilities import _KNOWN_MODELS + + for pattern, caps in _KNOWN_MODELS: + if not caps.authoritative: + assert caps.context_length == ModelCapabilities().context_length, ( + f"family pattern {pattern!r} must not pin a context length" + ) + + +def test_missing_registry_falls_back_to_the_budget() -> None: + engine = object.__new__(AgentEngine) + engine._settings = SimpleNamespace(llm_model="anything", llm_context_length=512_000) + engine._model_capabilities = None + + assert AgentEngine._active_context_length(engine) == 512_000 + + +def test_a_broken_registry_does_not_break_budget_resolution() -> None: + class _Broken: + def resolve(self, model): + raise RuntimeError("registry exploded") + + assert _active_length("qwen3.8-max", 1_000_000, _Broken()) == 1_000_000 + + +# ── Truncation chain scaling ───────────────────────────────────────────── + + +def test_tool_result_budget_grows_with_a_large_window() -> None: + """The old min() form pinned every large window to the base value.""" + assert adaptive_tool_result_chars(3000, 1_000_000) > 3000 + assert adaptive_tool_result_chars(3000, 1_000_000) == 25_000 + + +def test_tool_result_budget_still_contracts_on_small_windows() -> None: + """A 32K model must not spend a tenth of its window on one tool result.""" + assert adaptive_tool_result_chars(3000, 32_000) < 3000 + + +def test_tool_result_budget_is_monotonic_and_bounded() -> None: + windows = [16_000, 32_000, 128_000, 200_000, 1_000_000, 8_000_000] + values = [adaptive_tool_result_chars(3000, w) for w in windows] + + assert values == sorted(values), "a bigger window must never yield a smaller budget" + assert max(values) <= _RESULT_CEILING_CHARS + assert min(values) >= _RESULT_FLOOR_CHARS + + +def test_tool_result_budget_holds_near_the_historical_value_at_128k() -> None: + """Existing 128K setups should not regress; the divisor was picked for this.""" + assert 2_800 <= adaptive_tool_result_chars(3000, 128_000) <= 3_600 + + +def test_tool_result_budget_handles_an_unknown_window() -> None: + assert adaptive_tool_result_chars(3000, 0) == 3000 + assert adaptive_tool_result_chars(3000, -1) == 3000 + + +# ── The scaling must actually be wired into the runtime ──────────────── + + +def test_runtime_sync_applies_the_widened_budget_to_the_engine() -> None: + """Covers the wiring, not just the formula. + + A correct helper is useless if the call site still clamps. This drives + ``_sync_engine_runtime_budget`` and asserts what the engine actually receives, + which is what a formula-only test cannot see. + """ + from leapflow.cli.context import Context + + class _Engine: + def __init__(self) -> None: + self.result_budget = 0 + self.caps = None + + def set_tool_result_budget(self, value: int) -> None: + self.result_budget = value + + def set_model_capabilities(self, registry) -> None: + self.caps = registry + + ctx = object.__new__(Context) + engine = _Engine() + ctx.engine = engine + ctx.llm_chain = None + settings = SimpleNamespace( + llm_model="qwen3.8-max", + llm_context_length=1_000_000, + max_tool_result_chars=3000, + context_hard_limit_ratio=0.92, + native_tool_calling_enabled=True, + ) + + Context._sync_engine_runtime_budget(ctx, settings) + + assert engine.result_budget == 25_000, "a 1M window must not stay pinned at 3000" + assert engine.caps is not None, "capability registry must still be wired" + + +def test_runtime_sync_still_contracts_for_a_small_window() -> None: + """The widening must not remove the small-window protection.""" + from leapflow.cli.context import Context + + class _Engine: + def __init__(self) -> None: + self.result_budget = 0 + + def set_tool_result_budget(self, value: int) -> None: + self.result_budget = value + + def set_model_capabilities(self, registry) -> None: + pass + + ctx = object.__new__(Context) + engine = _Engine() + ctx.engine = engine + ctx.llm_chain = None + settings = SimpleNamespace( + llm_model="tiny-model", + llm_context_length=32_000, + max_tool_result_chars=3000, + context_hard_limit_ratio=0.92, + native_tool_calling_enabled=True, + ) + + Context._sync_engine_runtime_budget(ctx, settings) + + assert engine.result_budget < 3000 diff --git a/tests/test_context_governance.py b/tests/test_context_governance.py index a9ce851..0696425 100644 --- a/tests/test_context_governance.py +++ b/tests/test_context_governance.py @@ -8,6 +8,7 @@ CompressorConfig, ContextCompressor, SummarizeStage, + _TRIM_CEILING_CHARS, adaptive_trim_chars, estimate_text_tokens, ) @@ -18,6 +19,7 @@ ContextWindowController, LongTaskContextController, ToolEvidenceBuilder, + _EVIDENCE_CEILING_CHARS, ) from leapflow.tools.file_operations import file_list, file_read @@ -383,7 +385,9 @@ def test_adaptive_trim_chars_scales_with_context_length() -> None: assert threshold_128k > 2000 assert threshold_256k > threshold_128k assert threshold_1m >= threshold_256k - assert threshold_1m <= 50_000 # ceiling + # Bound on the ceiling constant, not a copy of its value: the contract is + # "bounded and monotonic", and the ceiling itself rises as windows grow. + assert threshold_1m <= _TRIM_CEILING_CHARS def test_adaptive_trim_chars_never_below_base() -> None: @@ -557,7 +561,7 @@ def test_evidence_builder_adapts_to_context_length() -> None: def test_evidence_builder_ceiling() -> None: huge = ToolEvidenceBuilder(max_content_chars=1200, context_length=2_000_000) - assert huge._max_content_chars <= 8_000 + assert huge._max_content_chars <= _EVIDENCE_CEILING_CHARS def test_evidence_builder_preserves_platform_permission_recovery_fields() -> None: From 8d133c8cf3375bd038c2513ea7681b1ded2efd2f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8F=AD=E6=89=AC?= Date: Wed, 5 Aug 2026 18:07:12 +0800 Subject: [PATCH 2/4] feat(context): calibrate the budget estimator from provider token counts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Intended to be P2 (move loss-free stages ahead of LLM summarisation), but that turned out to be already done: TrimStage is the first stage and is explicitly zero-cost (dedup by hash + truncate), Summarize is token-driven behind an anti-thrashing guard, and summarize_fn/archive_fn are both wired. There was no work left worth doing there, so this addresses the next real gap instead — estimate accuracy, which matters more as windows grow. The gate decision runs on a character heuristic (CJK 1:1, Latin 4:1). Its error scales with the window: on a 1M budget a 15% miss is ~150K tokens, enough to either overrun the 0.92 hard gate or waste a large slice of the window. Provider prompt_tokens was already captured and already replaced the estimate for cross-turn use, but nothing fed it back into the estimator itself. ContextBudgetEstimator now learns its own correction from that signal (EMA, clamped to 0.5-2.5, small prompts and outlier ratios rejected because provider-side caching and injected system content distort a single sample). No tokenizer dependency: a per-vendor tokenizer goes stale exactly the way the capability table did, and this adapts to whatever model and language mix is actually in use. With zero observations behaviour is bit-for-bit unchanged. Calibration happens before _record_provider_usage overwrites the snapshot, since that overwrite is what destroys the (estimate, actual) pair, and a snapshot that already holds a provider count is skipped so the estimator cannot calibrate against its own output. One defect found by measuring rather than reasoning: the first version fed the *corrected* estimate back for comparison, so the observed ratio approached 1.0 precisely when calibration began working and the factor drifted back (0.649 -> 0.802 over six rounds). The factor is now divided out first, and a regression test pins that it does not creep toward 1.0. Measured after the fix: residual 54.1% -> 0.0% when the heuristic overestimates, and the underestimating direction converges too. 12 new tests including the wiring and a failure-isolation case; reintroducing either defect fails 4 of them. Suite 1410 -> 1422 passing, ruff clean. --- src/leapflow/engine/context_control.py | 78 +++++++++- src/leapflow/engine/engine.py | 23 +++ tests/test_budget_calibration.py | 203 +++++++++++++++++++++++++ 3 files changed, 301 insertions(+), 3 deletions(-) create mode 100644 tests/test_budget_calibration.py diff --git a/src/leapflow/engine/context_control.py b/src/leapflow/engine/context_control.py index 57daab5..7316b97 100644 --- a/src/leapflow/engine/context_control.py +++ b/src/leapflow/engine/context_control.py @@ -70,8 +70,80 @@ class ContextBudgetDecision: notice: str = "" +# Estimator self-calibration. The character heuristic cannot match a real +# tokenizer, and the residual error scales with the window: on a 1M budget a 15% +# underestimate is ~150K tokens of invisible headroom loss, which either trips +# provider overflow or wastes the window. Rather than shipping a tokenizer per +# vendor (a dependency that goes stale exactly like the capability table), the +# estimator learns its own correction from provider-reported prompt_tokens. +_CALIBRATION_SMOOTHING = 0.25 +_CALIBRATION_MIN_FACTOR = 0.5 +_CALIBRATION_MAX_FACTOR = 2.5 +_CALIBRATION_MIN_SAMPLE_TOKENS = 200 + + class ContextBudgetEstimator: - """Estimate provider-visible prompt tokens including tool schemas.""" + """Estimate provider-visible prompt tokens including tool schemas. + + Self-calibrating: ``observe_actual`` feeds back a provider-reported prompt + token count so the character heuristic converges on the active model and + language mix. Uncalibrated behaviour is unchanged, so this only ever narrows + the gap between the estimate and the number the provider actually charges. + """ + + def __init__(self) -> None: + self._calibration: float = 1.0 + self._samples: int = 0 + + @property + def calibration_factor(self) -> float: + """Current correction applied to raw heuristic estimates.""" + return self._calibration + + @property + def calibration_samples(self) -> int: + """How many provider observations have shaped the current factor.""" + return self._samples + + def observe_actual(self, *, estimated: int, actual: int) -> None: + """Fold a provider-reported prompt token count into the calibration. + + ``estimated`` is a value this estimator produced, so it already carries + the current factor. It is divided back out before comparing: measuring + the corrected estimate against the actual would make the observed ratio + approach 1.0 as soon as calibration started working, dragging the factor + back toward its uncalibrated value. The comparison must always be against + the raw heuristic. + + Uses an exponential moving average so a single outlier cannot swing the + factor, and clamps the result: a wildly wrong factor would be worse than + the raw heuristic, and the observed ratio can be distorted by + provider-side prompt caching or injected system content. + + Tiny prompts are ignored — fixed per-request overhead dominates them and + would bias the factor upward for every later estimate. + """ + if estimated < _CALIBRATION_MIN_SAMPLE_TOKENS or actual <= 0: + return + raw = estimated / self._calibration if self._samples else float(estimated) + if raw <= 0: + return + observed = actual / raw + if not (_CALIBRATION_MIN_FACTOR <= observed <= _CALIBRATION_MAX_FACTOR): + return + if self._samples == 0: + self._calibration = observed + else: + self._calibration += _CALIBRATION_SMOOTHING * (observed - self._calibration) + self._calibration = min( + _CALIBRATION_MAX_FACTOR, max(_CALIBRATION_MIN_FACTOR, self._calibration), + ) + self._samples += 1 + + def _calibrated(self, raw: int) -> int: + if self._samples == 0 or raw <= 0: + return raw + return max(1, int(round(raw * self._calibration))) def estimate_messages(self, messages: Sequence[Dict[str, Any]]) -> int: """Estimate token usage for chat messages and tool-call envelopes.""" @@ -84,7 +156,7 @@ def estimate_messages(self, messages: Sequence[Dict[str, Any]]) -> int: total += self._estimate_value(message.get("content", "")) total += self._estimate_tool_calls(message.get("tool_calls", [])) total += self._estimate_value(message.get("tool_call_id", "")) - return max(1, total) + return max(1, self._calibrated(total)) def estimate_tools(self, tools: Any) -> int: """Estimate token usage for native function/tool schemas.""" @@ -94,7 +166,7 @@ def estimate_tools(self, tools: Any) -> int: text = json.dumps(tools, ensure_ascii=False, default=str, sort_keys=True) except (TypeError, ValueError): text = str(tools) - return _TOOL_SCHEMA_OVERHEAD_TOKENS + self._estimate_text(text) + return self._calibrated(_TOOL_SCHEMA_OVERHEAD_TOKENS + self._estimate_text(text)) def snapshot( self, diff --git a/src/leapflow/engine/engine.py b/src/leapflow/engine/engine.py index 0629215..38c97ac 100644 --- a/src/leapflow/engine/engine.py +++ b/src/leapflow/engine/engine.py @@ -2535,6 +2535,10 @@ def _record_provider_usage(self, model: str, usage: Dict[str, Any]) -> None: """Prefer provider prompt usage when available and learn observed limits.""" provider_prompt = int(usage.get("prompt_tokens", 0) or 0) if provider_prompt > 0: + # Calibrate before overwriting: the snapshot still holds this turn's + # estimate, so the pair (estimate, actual) is the only signal that can + # correct the character heuristic for this model and language mix. + self._calibrate_budget_estimator(provider_prompt) self._last_context_tokens = provider_prompt self._last_context_snapshot = { **self._last_context_snapshot, @@ -2545,6 +2549,25 @@ def _record_provider_usage(self, model: str, usage: Dict[str, Any]) -> None: if self._model_capabilities and model and usage: self._model_capabilities.update_from_usage(model, usage) + def _calibrate_budget_estimator(self, provider_prompt: int) -> None: + """Feed this turn's (estimate, actual) pair to the budget estimator.""" + snapshot = self._last_context_snapshot or {} + # Skip a snapshot already replaced by a provider count, otherwise the + # estimator would calibrate against its own previous observation. + if snapshot.get("provider_prompt_tokens"): + return + estimated = int(snapshot.get("total_tokens", 0) or 0) + if estimated <= 0: + return + estimator = getattr(self._context_window_controller, "estimator", None) + observe = getattr(estimator, "observe_actual", None) + if observe is None: + return + try: + observe(estimated=estimated, actual=provider_prompt) + except Exception: # noqa: BLE001 - calibration must never break a turn + logger.debug("budget estimator calibration failed", exc_info=True) + def _compact_tool_result(self, tool_name: str, arguments: Dict[str, Any] | None, result: Any) -> Any: """Return compact tool evidence for LLM replay.""" return self._context_governance_controller.compact_tool_result(tool_name, arguments, result) diff --git a/tests/test_budget_calibration.py b/tests/test_budget_calibration.py new file mode 100644 index 0000000..79b1059 --- /dev/null +++ b/tests/test_budget_calibration.py @@ -0,0 +1,203 @@ +"""Contracts for budget-estimator self-calibration. + +The character heuristic (CJK 1:1, Latin 4:1) cannot match a real tokenizer, and +the residual error scales with the window: on a 1M budget a 15% underestimate is +~150K tokens, enough to either trip provider overflow at the hard gate or waste a +large slice of the window. Shipping a per-vendor tokenizer would add a dependency +that goes stale the same way a static capability table does, so the estimator +instead learns its correction from provider-reported prompt_tokens. + +The subtle part is the feedback direction: the estimate handed back for +calibration already carries the current factor, so comparing it against the +actual makes the observed ratio approach 1.0 exactly when calibration starts +working — which drags the factor back to uncalibrated. Measured on a real run, +the factor drifted 0.649 -> 0.802 before this was fixed. +""" + +from __future__ import annotations + +from types import SimpleNamespace + +from leapflow.engine.context_control import ( + _CALIBRATION_MAX_FACTOR, + _CALIBRATION_MIN_FACTOR, + ContextBudgetEstimator, +) + +_LONG_CJK = [{"role": "user", "content": "推动经济增长的核心要素分析与产业结构变迁研究" * 40}] + + +def _converge(estimator: ContextBudgetEstimator, ratio: float, rounds: int = 8) -> int: + """Feed a consistent provider/estimate ratio until the factor settles.""" + actual = int(estimator.estimate_messages(_LONG_CJK) * ratio) + for _ in range(rounds): + estimator.observe_actual( + estimated=estimator.estimate_messages(_LONG_CJK), actual=actual, + ) + return actual + + +def test_uncalibrated_estimator_is_unchanged() -> None: + """No observations means no behaviour change from the previous heuristic.""" + estimator = ContextBudgetEstimator() + + assert estimator.calibration_samples == 0 + assert estimator.calibration_factor == 1.0 + assert estimator.estimate_messages(_LONG_CJK) > 0 + + +def test_calibration_converges_when_the_heuristic_overestimates() -> None: + """CJK 1:1 overestimates for BPE tokenizers that merge common words.""" + estimator = ContextBudgetEstimator() + + actual = _converge(estimator, 0.65) + + assert abs(estimator.calibration_factor - 0.65) < 0.02 + assert estimator.estimate_messages(_LONG_CJK) == actual + + +def test_calibration_converges_when_the_heuristic_underestimates() -> None: + """The dangerous direction: underestimating overruns the hard gate.""" + estimator = ContextBudgetEstimator() + + actual = _converge(estimator, 1.4) + + assert abs(estimator.calibration_factor - 1.4) < 0.02 + assert estimator.estimate_messages(_LONG_CJK) == actual + + +def test_the_factor_does_not_drift_back_once_calibrated() -> None: + """Regression: the corrected estimate must be un-corrected before comparing. + + Without dividing the factor back out, the observed ratio becomes 1.0 as soon + as calibration takes effect and the factor climbs back toward 1.0. + """ + estimator = ContextBudgetEstimator() + actual = _converge(estimator, 0.65, rounds=2) + settled = estimator.calibration_factor + + for _ in range(15): + estimator.observe_actual( + estimated=estimator.estimate_messages(_LONG_CJK), actual=actual, + ) + + assert abs(estimator.calibration_factor - settled) < 0.01 + assert estimator.calibration_factor < 0.7, "must not creep back toward 1.0" + + +def test_tiny_prompts_are_ignored() -> None: + """Fixed per-request overhead dominates small prompts and would skew it.""" + estimator = ContextBudgetEstimator() + + estimator.observe_actual(estimated=50, actual=10_000) + + assert estimator.calibration_samples == 0 + assert estimator.calibration_factor == 1.0 + + +def test_outlier_ratios_are_rejected() -> None: + """Prompt caching or injected system content can distort a single sample.""" + estimator = ContextBudgetEstimator() + + estimator.observe_actual(estimated=1_000, actual=1_000_000) + estimator.observe_actual(estimated=1_000, actual=1) + + assert estimator.calibration_samples == 0 + + +def test_missing_actual_is_ignored() -> None: + estimator = ContextBudgetEstimator() + + estimator.observe_actual(estimated=1_000, actual=0) + estimator.observe_actual(estimated=1_000, actual=-5) + + assert estimator.calibration_samples == 0 + + +def test_factor_stays_within_bounds() -> None: + """A wildly wrong factor would be worse than the raw heuristic.""" + estimator = ContextBudgetEstimator() + + for ratio in (2.4, 2.4, 2.4, 2.4, 2.4, 2.4, 2.4, 2.4): + estimator.observe_actual( + estimated=estimator.estimate_messages(_LONG_CJK), + actual=int(estimator.estimate_messages(_LONG_CJK) * ratio), + ) + + assert _CALIBRATION_MIN_FACTOR <= estimator.calibration_factor <= _CALIBRATION_MAX_FACTOR + + +def test_snapshot_ratio_reflects_calibration() -> None: + """Calibration must reach the gate decision, not just the raw estimate.""" + estimator = ContextBudgetEstimator() + before = estimator.snapshot(_LONG_CJK, tools=None, context_length=10_000).ratio + + _converge(estimator, 0.65) + after = estimator.snapshot(_LONG_CJK, tools=None, context_length=10_000).ratio + + assert after < before, "an overestimating heuristic must relax the gate ratio" + + +# ── Wiring: the engine must calibrate before overwriting the estimate ──── + + +def test_engine_calibrates_from_provider_usage() -> None: + """Covers the wiring: the pair must be captured before the overwrite. + + ``_record_provider_usage`` replaces the snapshot's token count with the + provider value, so calibration has to read the estimate first — afterwards + the signal is gone. + """ + from leapflow.engine.engine import AgentEngine + + estimator = ContextBudgetEstimator() + engine = object.__new__(AgentEngine) + engine._context_window_controller = SimpleNamespace(estimator=estimator) + engine._model_capabilities = None + engine._last_context_snapshot = {"total_tokens": 10_000, "context_length": 1_000_000} + engine._last_context_tokens = 0 + + AgentEngine._record_provider_usage(engine, "qwen3.8-max", {"prompt_tokens": 6_500}) + + assert estimator.calibration_samples == 1 + assert abs(estimator.calibration_factor - 0.65) < 0.01 + assert engine._last_context_tokens == 6_500 + + +def test_engine_does_not_calibrate_against_a_provider_snapshot() -> None: + """A snapshot already holding a provider count is not an estimate.""" + from leapflow.engine.engine import AgentEngine + + estimator = ContextBudgetEstimator() + engine = object.__new__(AgentEngine) + engine._context_window_controller = SimpleNamespace(estimator=estimator) + engine._model_capabilities = None + engine._last_context_snapshot = { + "total_tokens": 6_500, + "provider_prompt_tokens": 6_500, + "context_length": 1_000_000, + } + engine._last_context_tokens = 0 + + AgentEngine._record_provider_usage(engine, "qwen3.8-max", {"prompt_tokens": 6_500}) + + assert estimator.calibration_samples == 0, "would calibrate against its own output" + + +def test_calibration_failure_never_breaks_the_turn() -> None: + """Calibration is an optimisation; it must not be able to fail a request.""" + from leapflow.engine.engine import AgentEngine + + class _Exploding: + def observe_actual(self, **kwargs): + raise RuntimeError("boom") + + engine = object.__new__(AgentEngine) + engine._context_window_controller = SimpleNamespace(estimator=_Exploding()) + engine._model_capabilities = None + engine._last_context_snapshot = {"total_tokens": 10_000, "context_length": 1_000_000} + engine._last_context_tokens = 0 + + AgentEngine._record_provider_usage(engine, "m", {"prompt_tokens": 6_500}) + + assert engine._last_context_tokens == 6_500, "usage recording must still complete" From 40e0aadadd173000a27fec0836442e5220cc19a0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8F=AD=E6=89=AC?= Date: Wed, 5 Aug 2026 21:27:22 +0800 Subject: [PATCH 3/4] fix(engine): stop laundering local defects into the provider error path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit One mistyped attribute name made the agent unusable for every turn, and the reason it was both unusable and undiagnosable is the path the error took rather than the error itself. `_calibrate_budget_estimator` read `self._context_window_controller`, a name the engine never defines (the controller is `_context_controller`); it appeared exactly once in the tree, at the point of use. Because the read sat past two early returns, it only fired once a turn had a real context estimate — which is every substantive turn, and never in the suite. From there: AttributeError("... has no attribute '_context_window_controller'") -> raised inside the LLM call's try block, which also held the post-response bookkeeping, so a local bug entered the provider-error path -> ErrorClassifier matches on message text and saw "context" -> category context_overflow, recoverability auto_recover -> three rounds of compression, provider failover and credential rotation against a context that was fine -> fourth round: nothing left to try -> "No applicable recovery strategy found" as the entire user-visible answer -> that branch logged no traceback, and the audit sink was constructed with path=None, so nothing survived the turn Reproduced offline: three context_compress decisions then the exact halt message, matching the incident log's api_calls=4, tools=0, latency=114s. Fixes, in the order the failure travelled: - The attribute name. - `_record_llm_call_telemetry` now owns usage recording and calibration, absorbs its own failures into a warning, and is called after the provider call's try/except rather than inside it. Three call sites narrowed; one of them also bypassed calibration entirely by assigning `_last_context_tokens` directly. - Exceptions that mean "LeapFlow has a bug" (AttributeError, TypeError, NameError, KeyError, IndexError, ImportError, AssertionError, NotImplementedError) are classified by type into a new non-recoverable `internal_defect` category before the provider taxonomy is consulted. Matching Python's exception hierarchy is exact; matching message text is what turned a typo into a context overflow. - Both silent recovery entry points now log with exc_info, the audit sink is constructed with the profile layout's audit path, and every terminal decision carries an InteractionRequest, so a stopped turn states what failed and what to do instead of emitting internal jargon. The test gap mattered as much as the bug: the calibration wiring tests built the engine with `object.__new__` and assigned `_context_window_controller` themselves, asserting the same wrong name the code used. Mocking the attribute made the one thing those tests claimed to cover unobservable. They now use the real name, and three new guards close the hole: a real-engine calibration test on the production path, an architecture contract that every `self._x` the engine reads is assigned somewhere (167 reads checked; reintroducing the typo fails it), and test_internal_defect_reporting for the classification, budget, message and audit contracts. Verified beyond the suite: a single-turn answer and a tool-using turn both complete in-process (7.4s, 1 tool) where this shape previously halted after ~2 minutes. 1432 passed, ruff clean. Note: a running leapd predates this fix and must be restarted to pick it up. --- src/leapflow/engine/engine.py | 90 +++++---- src/leapflow/engine/recovery_coordinator.py | 39 +++- src/leapflow/engine/unified_classifier.py | 70 +++++++ tests/test_architecture_contracts.py | 32 ++++ tests/test_budget_calibration.py | 90 ++++++++- tests/test_internal_defect_reporting.py | 202 ++++++++++++++++++++ 6 files changed, 487 insertions(+), 36 deletions(-) create mode 100644 tests/test_internal_defect_reporting.py diff --git a/src/leapflow/engine/engine.py b/src/leapflow/engine/engine.py index 38c97ac..1ecc9c3 100644 --- a/src/leapflow/engine/engine.py +++ b/src/leapflow/engine/engine.py @@ -1222,7 +1222,7 @@ def __init__( self._unified_classifier = UnifiedErrorClassifier(self._error_classifier) self._recovery_coordinator = RecoveryCoordinator() # Re-created per turn self._checkpoint_store = InMemoryCheckpointStore() - self._audit_sink = JsonlAuditSink() # In-memory; path-based if layout available + self._audit_sink = JsonlAuditSink(self._recovery_audit_path()) # Apply startup-time tool configuration derived from settings. self._configure_tool_defaults() @@ -2531,6 +2531,47 @@ def _evaluate_prefix_commitment(self, budget: IterationBudget) -> None: snap["prefix_commitment"] = state.as_dict() snap["prefix_committed"] = state.committed + def _recovery_audit_path(self) -> Any: + """Return the profile-owned path for the recovery audit trail, if declared. + + Recovery decisions are the only record of why a turn stopped, and an + in-memory sink loses them with the turn — which is how an incident became + undiagnosable after the fact. Falls back to ``None`` (memory only) when no + profile layout is available, so an embedded engine still works. + """ + layout = getattr(self._settings, "profile_layout", None) + path = getattr(layout, "audit_log_path", None) + return path + + def _record_llm_call_telemetry(self, resp: Any, *, recovery: Any = None) -> None: + """Record usage and budget calibration for a provider call that succeeded. + + Deliberately self-contained: this is bookkeeping, and a defect in it must + never be mistaken for a provider failure. It used to run inside the + provider call's ``try`` block, where one mistyped attribute name became an + ``AttributeError`` that the LLM classifier read as a context overflow (the + name contained "context"), sending every round through compression, + failover, and credential rotation before halting the turn with an + unrelated message. Telemetry now fails loudly in the log and silently to + the turn. + + The success signal is recorded first so a bookkeeping defect cannot also + cost the loop its error-counter reset. + """ + try: + if recovery is not None: + recovery.record_api_success() + usage = getattr(resp, "usage", None) or {} + self._usage_tracker.record_api_call( + usage, + provider=getattr(self._llm, "active_provider_name", ""), + model=getattr(resp, "model", "") or "", + ) + if int(usage.get("prompt_tokens", 0) or 0) > 0: + self._record_provider_usage(getattr(resp, "model", "") or "", usage) + except Exception: # noqa: BLE001 - telemetry must never fail a turn + logger.warning("llm call telemetry recording failed", exc_info=True) + def _record_provider_usage(self, model: str, usage: Dict[str, Any]) -> None: """Prefer provider prompt usage when available and learn observed limits.""" provider_prompt = int(usage.get("prompt_tokens", 0) or 0) @@ -2559,7 +2600,7 @@ def _calibrate_budget_estimator(self, provider_prompt: int) -> None: estimated = int(snapshot.get("total_tokens", 0) or 0) if estimated <= 0: return - estimator = getattr(self._context_window_controller, "estimator", None) + estimator = getattr(self._context_controller, "estimator", None) observe = getattr(estimator, "observe_actual", None) if observe is None: return @@ -3006,16 +3047,6 @@ async def _run_agent_loop(self, frame: AgentLoopFrame) -> str: compressed, stream=False, enable_thinking=planned_enable_thinking, **tools_kwarg, ) - recovery.record_api_success() - usage = resp.usage or {} - self._usage_tracker.record_api_call( - usage, - provider=getattr(self._llm, 'active_provider_name', ''), - model=resp.model or '', - ) - provider_prompt = usage.get("prompt_tokens", 0) - if provider_prompt > 0: - self._record_provider_usage(resp.model or '', usage) except Exception as exc: _clear_indicator() classified = self._error_classifier.classify(exc) @@ -3027,6 +3058,12 @@ async def _run_agent_loop(self, frame: AgentLoopFrame) -> str: exc, provider=getattr(self._llm, 'provider', ''), model=getattr(self._llm, 'model', ''), ) + # Always with the traceback: this used to be the only record of a + # failed round, and it was not written anywhere. + logger.error( + "unified_loop: llm call failed (%s/%s)", + envelope.category, envelope.failure_code, exc_info=True, + ) coordinator = self._recovery_coordinator try: decision = coordinator.evaluate(envelope) @@ -3096,6 +3133,7 @@ async def _run_agent_loop(self, frame: AgentLoopFrame) -> str: fatal_error = _terminal_failure_text(decision) break _clear_indicator() + self._record_llm_call_telemetry(resp, recovery=recovery) content = (resp.content or "").strip() if self._sanitizer: @@ -3524,16 +3562,6 @@ async def _unified_tool_loop_stream( compressed, stream=False, enable_thinking=planned_enable_thinking, **tools_kwarg, ) - turn_recovery.record_api_success() - usage = resp.usage or {} - self._usage_tracker.record_api_call( - usage, - provider=getattr(self._llm, 'active_provider_name', ''), - model=resp.model or '', - ) - provider_prompt = usage.get("prompt_tokens", 0) - if provider_prompt > 0: - self._record_provider_usage(resp.model or '', usage) except Exception as exc: _clear_indicator() turn_recovery.record_api_error() @@ -3543,6 +3571,12 @@ async def _unified_tool_loop_stream( exc, provider=getattr(self._llm, 'provider', ''), model=getattr(self._llm, 'model', ''), ) + # The native-tools round previously logged nothing here, so a + # repeating failure left no trace at all in the daemon log. + logger.error( + "unified_loop_stream: llm call failed (%s/%s)", + envelope.category, envelope.failure_code, exc_info=True, + ) coordinator = self._recovery_coordinator try: decision = coordinator.evaluate(envelope) @@ -3592,6 +3626,7 @@ async def _unified_tool_loop_stream( ) break _clear_indicator() + self._record_llm_call_telemetry(resp, recovery=turn_recovery) content = (resp.content or "").strip() if self._sanitizer: @@ -3801,16 +3836,6 @@ async def _unified_tool_loop_stream( resp = await self._llm.achat( compressed, stream=False, enable_thinking=planned_enable_thinking, ) - turn_recovery.record_api_success() - usage = resp.usage or {} - self._usage_tracker.record_api_call( - usage, - provider=getattr(self._llm, 'active_provider_name', ''), - model=resp.model or '', - ) - provider_prompt = usage.get("prompt_tokens", 0) - if provider_prompt > 0: - self._last_context_tokens = provider_prompt except Exception as exc: _clear_indicator() turn_recovery.record_api_error() @@ -3860,6 +3885,7 @@ async def _unified_tool_loop_stream( ) break _clear_indicator() + self._record_llm_call_telemetry(resp, recovery=turn_recovery) content = (resp.content or "").strip() if self._sanitizer: content = self._sanitizer.sanitize(content) diff --git a/src/leapflow/engine/recovery_coordinator.py b/src/leapflow/engine/recovery_coordinator.py index cb0d440..c5d2422 100644 --- a/src/leapflow/engine/recovery_coordinator.py +++ b/src/leapflow/engine/recovery_coordinator.py @@ -398,7 +398,13 @@ def _rollback_state_changes(self, decision: RecoveryDecision, strategy: Recovery self._state.compress_phase_index = phase_index def _make_terminal(self, envelope: FailureEnvelope, *, reason: str) -> RecoveryDecision: - """Create a HALT_CLEAN terminal decision.""" + """Create a HALT_CLEAN terminal decision carrying what the user should do. + + A terminal decision is the last thing a stopped turn can say, so it always + carries an ``InteractionRequest``: the raw ``reason`` is written for the + audit log and reads as internal jargon ("No applicable recovery strategy + found") when shown alone. The request names the failure and the next step. + """ return RecoveryDecision.create( envelope=envelope, action=RecoveryAction.HALT_CLEAN, @@ -406,6 +412,37 @@ def _make_terminal(self, envelope: FailureEnvelope, *, reason: str) -> RecoveryD strategy_key="", retry_semantics=RetrySemantics(consumes_retry_budget=False), budget_cost=0, + interaction=self._terminal_interaction(envelope, reason), + ) + + def _terminal_interaction(self, envelope: FailureEnvelope, reason: str) -> InteractionRequest: + """Build the user-facing request that accompanies a terminal halt.""" + hint = getattr(getattr(envelope, "provider_hint", None), "hint_text", "") or "" + failure = envelope.failure_code or envelope.category or "unknown failure" + attempted = ", ".join(self._guard.used_strategies()) + description = envelope.message or reason + if hint: + description = f"{description} {hint}" + if attempted: + description = f"{description} Recovery already tried: {attempted}." + actions = [ + SuggestedAction(label="Rephrase or narrow the request and try again", is_default=True), + SuggestedAction(label="Check the daemon log for the full error", command="leap daemon logs"), + ] + return InteractionRequest.create( + interaction_type=InteractionType.RETRY_CHOICE, + severity=Severity.ERROR, + title=f"This turn stopped: {failure}", + description=description, + suggested_actions=tuple(actions), + context={ + "category": envelope.category, + "failure_code": envelope.failure_code, + "source": getattr(envelope.source, "value", str(envelope.source)), + "reason": reason, + }, + resumption_key=envelope.envelope_id, + timeout_behavior=TimeoutBehavior.PERSIST, ) def _consume_type_budget(self, decision: RecoveryDecision) -> None: diff --git a/src/leapflow/engine/unified_classifier.py b/src/leapflow/engine/unified_classifier.py index c9c55c4..027b57e 100644 --- a/src/leapflow/engine/unified_classifier.py +++ b/src/leapflow/engine/unified_classifier.py @@ -21,6 +21,11 @@ logger = logging.getLogger(__name__) +# Category for failures caused by a defect in LeapFlow rather than by a provider, +# a tool, or the environment. Kept distinct so it is never confused with a +# provider condition that recovery strategies could plausibly repair. +INTERNAL_DEFECT_CATEGORY = "internal_defect" + # Default mapping from ErrorCategory to (Recoverability, default_failure_class) _DEFAULT_MAPPINGS: dict[str, tuple[Recoverability, str]] = { ErrorCategory.TRANSIENT.value: (Recoverability.AUTO_RETRY, "transient"), @@ -38,8 +43,29 @@ ErrorCategory.IMAGE_TOO_LARGE.value: (Recoverability.AUTO_RECOVER, "image_too_large"), ErrorCategory.SSL_ERROR.value: (Recoverability.NON_RECOVERABLE, "ssl_error"), ErrorCategory.PERMANENT.value: (Recoverability.NON_RECOVERABLE, "permanent"), + # Defects in LeapFlow itself. Registered so the mapping is discoverable, but + # reached through exception type rather than message text (see below). + INTERNAL_DEFECT_CATEGORY: (Recoverability.NON_RECOVERABLE, "internal_defect"), } +# Exception types that always mean "LeapFlow has a bug", never "the provider or +# the network did something". They must bypass the provider taxonomy entirely: +# ``ErrorClassifier`` categorizes by substring on the exception message, so a +# mistyped attribute name that happened to contain "context" was classified as a +# context overflow and driven through compression, provider failover, and +# credential rotation before halting the turn with an unrelated reason. Matching +# on Python's own exception hierarchy is exact, unlike matching on message text. +_INTERNAL_DEFECT_TYPES: tuple[type[BaseException], ...] = ( + AttributeError, + NameError, + TypeError, + IndexError, + KeyError, + ImportError, + AssertionError, + NotImplementedError, +) + class RecoverabilityRegistry: """Extensible registry for category -> recoverability mapping. @@ -92,6 +118,42 @@ def __init__(self, error_classifier: Any = None, self._classifier = ErrorClassifier() self._registry = registry or RecoverabilityRegistry() + def classify_internal_defect( + self, + exc: Exception, + *, + provider: str = "", + model: str = "", + ) -> FailureEnvelope: + """Classify an exception that indicates a defect in LeapFlow itself. + + Reported as non-recoverable on purpose: no retry, compression, failover, + or credential rotation can fix a programming error, and attempting them + wastes the turn and misattributes the cause. The envelope names the + exception type so the halt message is actionable. + """ + detail = f"{type(exc).__name__}: {exc}" + logger.error("internal defect surfaced during an LLM call: %s", detail, exc_info=True) + return FailureEnvelope.create( + source=FailureSource.SYSTEM, + category=INTERNAL_DEFECT_CATEGORY, + failure_class="internal_defect", + failure_code=f"internal_{type(exc).__name__.lower()}", + message=detail[:500], + recoverability=Recoverability.NON_RECOVERABLE, + side_effect_state=SideEffectState.NONE, + context=FailureContext.from_dict_args( + tool_name="", + arguments={"provider": provider, "model": model} if provider or model else None, + ), + provider_hint=RecoveryHint( + hint_text=( + f"This is a defect in LeapFlow ({type(exc).__name__}), not a provider or " + "network problem. Retrying will not help; the traceback is in the daemon log." + ) + ), + ) + def classify_llm_error( self, exc: Exception, @@ -103,7 +165,15 @@ def classify_llm_error( Uses the existing ErrorClassifier internally to determine the category, then maps to the appropriate recoverability and constructs a FailureEnvelope. + + Exceptions that indicate a LeapFlow defect are routed away from the + provider taxonomy first: the provider classifier reads message text, so a + local bug would otherwise be assigned whatever provider condition its + message happens to resemble. """ + if isinstance(exc, _INTERNAL_DEFECT_TYPES): + return self.classify_internal_defect(exc, provider=provider, model=model) + category = self._classifier.classify(exc) category_str = category.value diff --git a/tests/test_architecture_contracts.py b/tests/test_architecture_contracts.py index 54beebf..7425af0 100644 --- a/tests/test_architecture_contracts.py +++ b/tests/test_architecture_contracts.py @@ -250,3 +250,35 @@ def test_module_imports_standalone(module_name: str) -> None: needs aiohttp/duckdb/an LLM at import time breaks unrelated entry points. """ assert importlib.import_module(module_name) is not None + + +def test_engine_self_attributes_all_exist() -> None: + """Every ``self._x`` the engine reads must actually be defined somewhere. + + A mistyped attribute name is invisible until the line runs, and the agent + loop wraps most of its work in broad ``except Exception`` handlers, so such a + typo surfaces as a misclassified recovery failure rather than a crash. One + of them (``_context_window_controller``, never assigned anywhere) made every + turn halt while the suite stayed green. + + Names assigned anywhere in the module count as defined, including on frames + and per-session clones; this catches the typo case, not lifecycle ordering. + """ + import re + from pathlib import Path + + import leapflow.engine.engine as engine_module + + source = Path(engine_module.__file__).read_text(encoding="utf-8") + read = set(re.findall(r"self\.(_[a-z][a-z0-9_]*)", source)) + assigned = set(re.findall(r"self\.(_[a-z][a-z0-9_]*)\s*(?::[^=\n]+)?=", source)) + # Attributes may also be set from outside (session_factory clones engines). + for module in ("leapflow.engine.session_factory", "leapflow.engine.agent_loop"): + mod = importlib.import_module(module) + assigned |= set( + re.findall(r"engine\.(_[a-z][a-z0-9_]*)\s*=", Path(mod.__file__).read_text(encoding="utf-8")) + ) + on_class = {name for name in read if hasattr(engine_module.AgentEngine, name)} + + undefined = sorted(read - assigned - on_class) + assert not undefined, f"engine reads attributes that are never assigned: {undefined}" diff --git a/tests/test_budget_calibration.py b/tests/test_budget_calibration.py index 79b1059..206861c 100644 --- a/tests/test_budget_calibration.py +++ b/tests/test_budget_calibration.py @@ -139,6 +139,13 @@ def test_snapshot_ratio_reflects_calibration() -> None: # ── Wiring: the engine must calibrate before overwriting the estimate ──── +# +# These build the engine with object.__new__ and assign the attributes the method +# reads. That is deliberate for the ordering contracts below, but it cannot catch +# a wrong attribute *name* — an earlier version of these tests asserted against +# `_context_window_controller`, a name the engine never defines, so the feature +# raised AttributeError on every real turn while the suite stayed green. The real +# instance test at the end of this file is what closes that hole; keep it. def test_engine_calibrates_from_provider_usage() -> None: @@ -152,7 +159,7 @@ def test_engine_calibrates_from_provider_usage() -> None: estimator = ContextBudgetEstimator() engine = object.__new__(AgentEngine) - engine._context_window_controller = SimpleNamespace(estimator=estimator) + engine._context_controller = SimpleNamespace(estimator=estimator) engine._model_capabilities = None engine._last_context_snapshot = {"total_tokens": 10_000, "context_length": 1_000_000} engine._last_context_tokens = 0 @@ -170,7 +177,7 @@ def test_engine_does_not_calibrate_against_a_provider_snapshot() -> None: estimator = ContextBudgetEstimator() engine = object.__new__(AgentEngine) - engine._context_window_controller = SimpleNamespace(estimator=estimator) + engine._context_controller = SimpleNamespace(estimator=estimator) engine._model_capabilities = None engine._last_context_snapshot = { "total_tokens": 6_500, @@ -193,7 +200,7 @@ def observe_actual(self, **kwargs): raise RuntimeError("boom") engine = object.__new__(AgentEngine) - engine._context_window_controller = SimpleNamespace(estimator=_Exploding()) + engine._context_controller = SimpleNamespace(estimator=_Exploding()) engine._model_capabilities = None engine._last_context_snapshot = {"total_tokens": 10_000, "context_length": 1_000_000} engine._last_context_tokens = 0 @@ -201,3 +208,80 @@ def observe_actual(self, **kwargs): AgentEngine._record_provider_usage(engine, "m", {"prompt_tokens": 6_500}) assert engine._last_context_tokens == 6_500, "usage recording must still complete" + + +# ── The hole the mocks left: a real engine, the production code path ───── + + +def _real_engine(tmp_path): + """Build an actual AgentEngine, so attribute wiring is exercised for real.""" + from conftest import StubLLM, make_settings + from leapflow.engine.engine import AgentEngine, build_default_registry + from leapflow.engine.intent_classifier import Intent + from leapflow.memory import ( + EpisodicMemoryProvider, + SemanticMemoryProvider, + WorkingMemoryProvider, + ) + from leapflow.platform.mock import MockBridge + + class _Classifier: + async def classify(self, user_text: str) -> Intent: + return Intent(label="complex", reason="test") + + settings = make_settings(str(tmp_path)) + rpc = MockBridge() + llm = StubLLM(["ok"]) + wm = WorkingMemoryProvider(max_tokens=1024) + lt = SemanticMemoryProvider(source=settings.duckdb_path) + imm = EpisodicMemoryProvider() + registry = build_default_registry(rpc, llm, wm, lt) + engine = AgentEngine(settings, rpc, llm, wm, lt, imm, registry, _Classifier()) + return engine, lt + + +def test_real_engine_calibrates_on_the_production_path(tmp_path) -> None: + """The regression test for the outage: a real engine, a realistic snapshot. + + Every existing calibration test either mocked the controller attribute or + left the snapshot empty, and an empty snapshot returns early before the + controller is ever read. A turn with a real context estimate is the only + shape that reaches the wiring, and it raised AttributeError on every round. + """ + engine, store = _real_engine(tmp_path) + try: + engine._last_context_snapshot = {"total_tokens": 10_000, "context_length": 1_000_000} + + engine._record_provider_usage("qwen3.8-max", {"prompt_tokens": 6_500}) + + assert engine._context_controller.estimator.calibration_samples == 1 + # The overwrite must also have happened; it used to be unreachable. + assert engine._last_context_tokens == 6_500 + assert engine._last_context_snapshot["provider_prompt_tokens"] == 6_500 + finally: + store.close() + + +def test_telemetry_helper_absorbs_its_own_defects(tmp_path) -> None: + """A bookkeeping defect must not propagate into the provider-error path. + + The outage was a local AttributeError escaping into the LLM call's except + block, where it was classified as a provider condition and driven through + recovery. The helper now contains anything it raises. + """ + engine, store = _real_engine(tmp_path) + try: + def _boom(*args, **kwargs): + raise AttributeError("'AgentEngine' object has no attribute '_whatever'") + + engine._record_provider_usage = _boom + recorded = [] + + engine._record_llm_call_telemetry( + SimpleNamespace(usage={"prompt_tokens": 6_500}, model="m"), + recovery=SimpleNamespace(record_api_success=lambda: recorded.append(True)), + ) + + assert recorded == [True], "the success signal must survive a telemetry defect" + finally: + store.close() diff --git a/tests/test_internal_defect_reporting.py b/tests/test_internal_defect_reporting.py new file mode 100644 index 0000000..c2936d9 --- /dev/null +++ b/tests/test_internal_defect_reporting.py @@ -0,0 +1,202 @@ +"""Contracts for how a defect inside LeapFlow is reported, not laundered. + +Written after an outage where one mistyped attribute name made the agent unusable +for every turn. The failure itself was trivial; what made it unusable, and then +undiagnosable, was the path it took: + + AttributeError("... has no attribute '_context_window_controller'") + -> raised inside the LLM call's try block (bookkeeping shared the block) + -> ErrorClassifier matched "context" in the message -> context_overflow + -> recoverability auto_recover -> three rounds of context compression, + provider failover and credential rotation on a context that was fine + -> fourth round: no strategy left -> "No applicable recovery strategy found" + -> no traceback logged on that branch, audit sink in memory only + +Each layer is pinned below so the chain cannot re-form. +""" + +from __future__ import annotations + +import json + +from leapflow.engine.error_classifier import ErrorClassifier +from leapflow.engine.failure_envelope import FailureSource, Recoverability +from leapflow.engine.recovery_audit import JsonlAuditSink, create_audit_entry +from leapflow.engine.recovery_budget import RecoveryBudget +from leapflow.engine.recovery_coordinator import RecoveryCoordinator +from leapflow.engine.recovery_decision import RecoveryAction +from leapflow.engine.recovery_strategies import default_strategies +from leapflow.engine.unified_classifier import ( + INTERNAL_DEFECT_CATEGORY, + UnifiedErrorClassifier, +) + +# The exact exception from the outage. Its message contains "context", which is +# what the text-matching provider classifier keyed on. +_OUTAGE_EXC = AttributeError( + "'AgentEngine' object has no attribute '_context_window_controller'" +) + + +def _classifier() -> UnifiedErrorClassifier: + return UnifiedErrorClassifier(ErrorClassifier()) + + +def _budget() -> RecoveryBudget: + budget = RecoveryBudget( + turn_deadline_s=0, total_recovery_actions=20, max_retry_per_category=10, + ) + budget.start_deadline() + return budget + + +# ── Classification: a bug is a bug, whatever its message says ──────────── + +def test_internal_defect_is_not_classified_as_a_provider_condition() -> None: + """The outage's exception must never be read as a context overflow again.""" + envelope = _classifier().classify_llm_error(_OUTAGE_EXC, provider="dashscope", model="m") + + assert envelope.category == INTERNAL_DEFECT_CATEGORY + assert envelope.category != "context_overflow" + assert envelope.source is FailureSource.SYSTEM + assert envelope.recoverability is Recoverability.NON_RECOVERABLE + assert "AttributeError" in envelope.message + + +def test_defect_types_are_matched_by_type_not_message() -> None: + """Type-based dispatch: the message text must not influence the verdict.""" + classifier = _classifier() + for exc in ( + AttributeError("rate limit exceeded"), # message looks transient + TypeError("context length exceeded"), # message looks like overflow + KeyError("timeout"), # message looks transient + NameError("invalid api key"), # message looks like auth + IndexError("list index out of range"), + AssertionError("should not happen"), + ): + envelope = classifier.classify_llm_error(exc) + assert envelope.category == INTERNAL_DEFECT_CATEGORY, exc + assert envelope.recoverability is Recoverability.NON_RECOVERABLE, exc + + +def test_real_provider_conditions_still_use_the_provider_taxonomy() -> None: + """The narrowing must not swallow genuine provider errors.""" + classifier = _classifier() + + transient = classifier.classify_llm_error(ConnectionError("connection reset by peer")) + assert transient.source is FailureSource.LLM + assert transient.category != INTERNAL_DEFECT_CATEGORY + assert transient.recoverability is not Recoverability.NON_RECOVERABLE + + overflow = classifier.classify_llm_error( + RuntimeError("This model's maximum context length is 8192 tokens") + ) + assert overflow.source is FailureSource.LLM + assert overflow.category == "context_overflow" + + +# ── Recovery: no compression/failover/rotation for a local bug ─────────── + +def test_defect_halts_immediately_without_burning_strategies() -> None: + """Previously this consumed four rounds before giving up. + + Compression, provider failover, and credential rotation cannot repair a + programming error; attempting them cost ~2 minutes per turn and reported the + failure as something it was not. + """ + coordinator = RecoveryCoordinator(strategies=default_strategies(), budget=_budget()) + coordinator.new_turn(turn_id=0) + + decision = coordinator.evaluate(_classifier().classify_llm_error(_OUTAGE_EXC)) + + assert decision.action is RecoveryAction.HALT_CLEAN + assert decision.strategy_key == "" + assert "context_compress" not in coordinator.guard.used_strategies() + assert coordinator.budget.remaining() == coordinator.budget.total_recovery_actions + + +# ── The halt must tell the user something ──────────────────────────────── + +def test_terminal_decision_carries_an_actionable_interaction() -> None: + """A stopped turn must not surface internal jargon as its whole answer.""" + from leapflow.engine.engine import _terminal_failure_text + + coordinator = RecoveryCoordinator(strategies=default_strategies(), budget=_budget()) + coordinator.new_turn(turn_id=0) + decision = coordinator.evaluate(_classifier().classify_llm_error(_OUTAGE_EXC)) + + assert decision.interaction is not None + rendered = _terminal_failure_text(decision) + assert "No applicable recovery strategy found" not in rendered + assert "AttributeError" in rendered + assert decision.interaction.suggested_actions + + +def test_no_strategy_terminal_also_explains_itself() -> None: + """The exact message from the incident must never be the whole answer.""" + from leapflow.engine.engine import _terminal_failure_text + + coordinator = RecoveryCoordinator(strategies=[], budget=_budget()) + coordinator.new_turn(turn_id=0) + envelope = _classifier().classify_llm_error(ConnectionError("connection reset by peer")) + + decision = coordinator.terminal_decision(envelope) + + assert decision.reason == "No applicable recovery strategy found" + rendered = _terminal_failure_text(decision) + assert rendered != decision.reason + assert "stopped" in rendered.lower() + + +# ── Diagnosability: the record must outlive the turn ───────────────────── + +def test_recovery_audit_is_written_to_disk(tmp_path) -> None: + """An in-memory sink loses the only record of why a turn stopped.""" + path = tmp_path / "audit" / "runtime.jsonl" + sink = JsonlAuditSink(path) + budget = _budget() + coordinator = RecoveryCoordinator(strategies=default_strategies(), budget=budget) + coordinator.new_turn(turn_id=0) + envelope = _classifier().classify_llm_error(_OUTAGE_EXC) + decision = coordinator.evaluate(envelope) + + sink.record(create_audit_entry(envelope, decision, budget, session_id="s", turn_id=0)) + + lines = [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line] + assert lines, "recovery decisions must be persisted" + assert lines[0]["failure_category"] == INTERNAL_DEFECT_CATEGORY + assert lines[0]["action"] == RecoveryAction.HALT_CLEAN.value + + +def test_engine_points_the_audit_sink_at_the_profile_layout(tmp_path) -> None: + """The sink must be constructed with a layout-owned path, not left in memory.""" + from conftest import StubLLM, make_settings + from leapflow.engine.engine import AgentEngine, build_default_registry + from leapflow.engine.intent_classifier import Intent + from leapflow.memory import ( + EpisodicMemoryProvider, + SemanticMemoryProvider, + WorkingMemoryProvider, + ) + from leapflow.platform.mock import MockBridge + + class _Classifier: + async def classify(self, user_text: str) -> Intent: + return Intent(label="complex", reason="test") + + settings = make_settings(str(tmp_path)) + store = SemanticMemoryProvider(source=settings.duckdb_path) + try: + rpc = MockBridge() + llm = StubLLM(["ok"]) + wm = WorkingMemoryProvider(max_tokens=1024) + registry = build_default_registry(rpc, llm, wm, store) + engine = AgentEngine( + settings, rpc, llm, wm, store, EpisodicMemoryProvider(), registry, _Classifier(), + ) + + expected = getattr(getattr(settings, "profile_layout", None), "audit_log_path", None) + if expected is not None: + assert engine._audit_sink._path == expected + finally: + store.close() From 256fa529a88e8fa766d3fa903d7bf5e6789f42fa Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8F=AD=E6=89=AC?= Date: Wed, 5 Aug 2026 22:02:33 +0800 Subject: [PATCH 4/4] fix(daemon,tui): keep session identity with the client that owns it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two TUIs in different workspaces left the second one unusable: every turn was rejected with "Session '...' is bound to workspace A; current request uses B", and the advice it gave ("start a fresh TUI session") could not work. The workspace guard was right. What was wrong is how the second client came to hold the first client's session id: - `SessionRegistry.most_recent()` returns the most recently active session of any client, ignoring workspace and client identity. - `resolve_session_engine(ctx, "")` falls back to it when no session is named. - `status()` took no session id at all, so every caller got that fallback — and with it another client's `session_id` and context figures. `_chunk_from_event` had the same fallback whenever an engine was not passed explicitly. - The TUI then adopted it unconditionally: `active_session_id = str(metadata["session_id"])`, on startup, on every refresh, and on every stream chunk. So TUI#2 adopted TUI#1's session on its first status poll, sent it with its own workspace on the next turn, and was refused — permanently, since a fresh client re-adopts the same id immediately. The second, quieter symptom was TUI#2's status bar showing TUI#1's context usage. The fallback arrived with 0291d19 and was wired into status/stream metadata by 8061fc1 to fix a status bar stuck at 0/1M. Both share one root: client-visible runtime state was resolved by "most recently active" rather than "this caller". Fixed in the order the identity travelled: - The client accepts a reported session id only when it matches its own or it has none yet (first assignment, or an explicit --resume). This alone stops the outage even if the daemon regresses. - `status()` takes the caller's `session_id` and resolves that session; with none named it reports no session identity and no per-session figures rather than a substitute. Client, protocol, and the TUI's three call sites pass it. - `_chunk_from_event` requires the producing engine; the fallback is gone. Its only caller already passed one, so the fallback was pure risk. - `most_recent()` is now `most_recent_any_client()`, documented as valid only for aggregate views, so it cannot be mistaken for "the caller's session" again. - The mismatch message names the one legitimate cause (--resume from another workspace) instead of telling the user to do what they already did. Verified with real Settings: with TUI#1 active in workspace A, TUI#2 polling status keeps its own session, its turn is accepted in workspace B, and both sessions coexist. Each client's status reports its own usage (5000 vs 99), so the leak is closed without regressing the status bar to zero — the case 8061fc1 fixed. Four existing tests asserted the leaky behaviour and were corrected: one expected a session-less `status()` to report context usage, another exercised the `_chunk_from_event` fallback. New `test_multi_client_session_isolation` covers workspace binding, the rename, per-caller status, RPC plumbing, client adoption, and a source-level guard against the unconditional assignment returning. AGENTS.md gains nine contracts across architecture, recovery, and testing. The important one is that concurrent TUIs in different workspaces are a supported scenario, not an edge case: a change to session routing, status, stream metadata or the client lease is not verified until two instances in two workspaces have been exercised together. The recovery entries record the previous commit's lessons (a local defect is never a provider failure; terminal decisions must be actionable; recovery must leave evidence), and the testing entries record the two blind spots that let both incidents ship green — a test may not fabricate the wiring it claims to cover, and multi-client behaviour needs multi-client tests. 1442 passed, ruff clean. Confirmed by hand in two live TUIs. --- AGENTS.md | 11 ++ src/leapflow/cli/commands/interactive.py | 22 ++- src/leapflow/daemon/client.py | 13 +- src/leapflow/daemon/protocol.py | 9 +- src/leapflow/daemon/service.py | 49 ++--- src/leapflow/daemon/session_coordinator.py | 7 +- src/leapflow/daemon/session_registry.py | 26 ++- tests/test_board_session_binding.py | 8 +- tests/test_cli_entrypoint.py | 8 +- tests/test_daemon_rpc.py | 11 +- tests/test_multi_client_session_isolation.py | 193 +++++++++++++++++++ tests/test_runtime_metadata_and_wrapping.py | 28 ++- 12 files changed, 334 insertions(+), 51 deletions(-) create mode 100644 tests/test_multi_client_session_isolation.py diff --git a/AGENTS.md b/AGENTS.md index aa79a1f..15d442c 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -33,6 +33,7 @@ This document is the LeapFlow engineering collaboration contract. It is not only - **System Boundary Awareness**: LeapFlow is a multi-entry, multi-module runtime. Changes must account for the affected path across CLI/TUI, leapd, engine, skills/tools, LLM, storage, memory, gateway, hub, and platform adapters. - **TUI as the Primary User Entry**: The interactive TUI is the default product surface. Preserve streaming feedback, command queue behavior, approval prompts, status bar accuracy, long-input robustness, history, and session continuity. +- **Concurrent TUI Instances Are a Supported Scenario (MANDATORY)**: several TUIs in *different workspaces*, sharing one leapd and one profile, is a normal way to use LeapFlow — not an edge case. Each instance must remain fully usable and must see only its own session, conversation, context usage, and turn state. A change to session routing, `status()`, stream metadata, the client lease, or anything the status bar renders is not verified until it has been exercised with two instances in two workspaces at the same time. One instance degrading another is a release blocker, not a limitation to document. - **TUI Command Clarity**: Global task-control commands stay short and unambiguous (`/cancel`, `/skip`, `/pause`, `/resume`, `/queue`, `/drop`); teach-mode controls must use the `/teach ...` namespace and should not keep bare compatibility aliases during early iteration. - **TUI Prompt Ownership**: Input prompt and placeholder rendering must have a single owner. Avoid duplicate prompt sources; placeholder text stays visually subordinate, offset after the prompt, and disappears as soon as the user types. - **leapd Runtime Consistency**: Daemon-backed behavior must preserve lifecycle correctness: start, stop, restart, status, RPC streaming, cancellation, pending approvals, runtime config reload, multi-client state, and version consistency. @@ -46,6 +47,9 @@ This document is the LeapFlow engineering collaboration contract. It is not only - **Event-Driven Communication**: Modules interact through typed events on EventBus, not direct imports - **Session Engine is the Only Reporting Source (MANDATORY)**: conversation state lives on the per-session engines built by `SessionRegistry`. `ctx.engine` is only the template they are cloned from and **never accumulates turns, context, or history**. Any code reporting runtime state — stream chunk metadata, `status()`, session analysis, dashboards, status bar values — must resolve the engine through the single entry point (`RuntimeLeapService._active_engine()` / `SessionCoordinator.resolve_session_engine()`), never `getattr(ctx, "engine")`. Reading the template silently yields zeros, which is invisible in review and has surfaced repeatedly as an empty LeapBoard and a status bar frozen at `0/`. When a value is produced *by* a specific engine (e.g. a stream event), pass that engine explicitly instead of re-resolving, so concurrent sessions cannot be cross-reported. - **Client-Visible Runtime State Must Be Pushed, Not Inferred**: a daemon-mode TUI is a separate process; it seeds model, context length, and usage at startup and can only learn about later changes from metadata the daemon returns. Any runtime value the status bar renders must travel on ordinary status/stream metadata (and on the mutation payload for command RPCs) — never rely on a one-off change notification, which change-detection can legitimately skip. +- **Session Identity Belongs to the Client That Created It (MANDATORY)**: a `session_id` is the client's own identity, not shared daemon state. The daemon must never report a session a caller did not name, and a client must never adopt a session id it did not ask for. Concretely: every RPC that returns session-scoped state (`status()`, stream chunk metadata, history, analysis) takes the caller's `session_id` and resolves *that* session; when the caller names none, the reply carries **no** session identity and no per-session figures rather than a substitute. A client accepts a reported session id only when it matches its own or it has none yet (first assignment, or an explicit `--resume`). Violating this is not cosmetic: a second TUI adopted the first's session, sent it with its own workspace, and was rejected on every turn — with advice ("start a fresh session") that could not work, because a fresh client re-adopted the same id on its first status poll. +- **Cross-Session Fallbacks Must Be Named For What They Do**: a resolver that answers "whichever session was most recently active" ignores workspace and client identity, so it is valid only for genuinely aggregate views (a dashboard summarizing all activity). Such helpers must say so in their name (e.g. `most_recent_any_client()`), and no code path that describes one caller may use them. A friendly name like "the current session" invites exactly the misuse that leaked one client's identity to another. +- **Workspace Binding Is Part of Session Identity**: a session is bound to the workspace of its first request, and reuse from another workspace is refused. Because correct clients cannot trigger this, a mismatch means either an explicit `--resume` into another workspace (the only legitimate cause, and what the message must name) or a defect — never something to tell the user to work around. - **Terminal Output Must Wrap at the Console Layer**: `LeapConsole` owns wrapping (`soft_wrap=False`), because prompt_toolkit's renderer clips at the window edge rather than reflowing. Never enable `soft_wrap` on the shared console or hand it pre-formatted long lines; long answers silently lose their tail. A standalone `Console` for fixed-width art (e.g. the banner) may opt out, but must set an explicit `width`. - **Immutable Domain Types**: Use `@dataclass(frozen=True)` or `NamedTuple` for domain objects - **Config-Driven Behavior**: Thresholds, intervals, feature flags, model budgets, platform capabilities, hub backends, gateway manifests, and paths must be configurable through Settings/env/config layers. @@ -53,6 +57,9 @@ This document is the LeapFlow engineering collaboration contract. It is not only - **Single Source of Truth**: DuckDB for persistence, EventBus for communication, Settings for configuration - **Inbound Signal Classification**: Platform events must be classified before they activate the agent. Message/callback events may enter Decide; signal/lifecycle events should be stored or routed without triggering LLM by default; ignored events must be explicit (e.g. self-message, duplicate, blocked scope). - **Single Recovery Decision Point**: All agent loop errors (LLM, tool, system, security) enter one `RecoveryCoordinator`. No parallel decision paths, no scattered if/break logic. The pipeline is always: `FailureEnvelope` → `RecoveryDecision` → `StrategyOutcome` feedback. +- **A Local Defect Is Never a Provider Failure (MANDATORY)**: an `except` around a provider call must wrap *only* that call. Post-response bookkeeping — usage recording, calibration, capability learning — belongs outside it, in a helper that contains its own failures, because telemetry must never fail a turn. Exceptions that mean "LeapFlow has a bug" (`AttributeError`, `TypeError`, `NameError`, `KeyError`, `IndexError`, `ImportError`, `AssertionError`, `NotImplementedError`) are classified by *type* into the non-recoverable `internal_defect` category before the provider taxonomy is consulted. The provider classifier matches on message text, so one mistyped attribute whose name contained "context" was read as a context overflow and driven through three compressions, a provider failover, and a credential rotation before halting — every turn, for every user, with the suite green. +- **Every Terminal Decision Must Be Actionable**: a halt is the last thing a stopped turn can say, so it always carries an `InteractionRequest` naming the failure and the next step. The raw `reason` is written for the audit log; surfacing it alone produces internal jargon as the user's entire answer ("No applicable recovery strategy found"). +- **Recovery Must Leave Evidence**: every recovery entry point logs the exception with `exc_info`, and the audit sink is constructed with the profile layout's audit path. An in-memory sink and an unlogged branch made a reproducible outage undiagnosable after the fact — the incident had to be reconstructed from token counts in an unrelated log line. - **Side-Effect Gating**: Recovery is gated by `SideEffectState` at two levels. Within a tool batch, a failed side-effecting call stops the remaining calls in that batch, decided by the declared `execution_policy` rather than any tool-name list. Within `RecoveryCoordinator`, any state other than `NONE` blocks replaying actions (retry, transform-and-retry, failover) and yields a checkpointed halt carrying an `InteractionRequest`; only user-mediated or checkpoint-based resumption is permitted. `UNKNOWN` blocks like `COMMITTED` and `PARTIAL` do: it is the classifier's fallback and the state assigned to `external_side_effect` (outbound sends, external API calls), so exempting it would leave the highest-risk case ungated. - **Uncertain Effects Are Reported, Not Retried Blindly**: a failed call whose effect may already have landed (`external_side_effect`, `mutating_once`) must carry that verdict in its result so the next turn verifies before repeating it. An error is not proof that nothing happened. Idempotent mutations are exempt — re-applying them converges, so flagging them would only stall safe retries. - **Budget-Constrained Recovery**: Turn-level deadlines, per-category limits, and a global recovery budget prevent infinite retry loops. Every recovery action has an explicit cost; exhaustion triggers a clean halt or user escalation. @@ -119,6 +126,8 @@ This document is the LeapFlow engineering collaboration contract. It is not only - **Verification sequence**: compile → import → unit test → integration (if applicable) - **Behavior contracts over snapshots**: assert invariants, not frozen values - **Mock at boundaries only**: mock external I/O (network, disk), never internal logic +- **A test may not fabricate the wiring it claims to cover**: building an object with `object.__new__` and assigning the private attributes the code reads cannot detect a wrong attribute *name* — the test simply agrees with the typo. Calibration tests did exactly that and stayed green while every real turn raised `AttributeError`. Any test whose stated purpose is wiring must construct the real object and drive the production path. +- **Multi-client behavior needs multi-client tests**: session routing, `status()`, stream metadata, and client-lease changes require two sessions in two workspaces asserting that neither sees the other's identity, usage, or turn state. Single-session tests cannot observe cross-client leakage, which is why a leak shipped with a green suite. - **Change-scoped validation**: Run the most specific relevant tests first, then broaden only as needed: CLI/TUI changes require CLI/TUI tests; leapd changes require daemon RPC/lifecycle tests; storage or memory changes require persistence tests; gateway, IM, event-source, or approval changes require connector lifecycle, event normalization, routing, idempotency, self-message filtering, security/approval, and failure-recovery tests; skills, learning, perception, and copilot changes require their lifecycle or pipeline tests. - **Recovery strategy isolation**: Each `RecoveryStrategy` must be testable in isolation — verify `can_apply` predicates, `decide` outputs, and side-effect-state gating independently of the coordinator and other strategies. - **Budget boundary tests**: Verify that recovery budgets exhaust correctly (per-category, per-turn, deadline), that exhaustion produces a deterministic halt decision, and that cost accounting is exact. @@ -141,6 +150,8 @@ This document is the LeapFlow engineering collaboration contract. It is not only - Magic retry counts or unbounded retry loops without budget constraints and deadline enforcement - Feeding unstructured error text back to the LLM without classification, recoverability assessment, or side-effect awareness - Multiple parallel error-handling paths for the same failure domain (LLM errors in one handler, tool errors in another, security errors in a third); use a unified classification and coordination pipeline +- Widening a provider call's `try` block to cover bookkeeping, or classifying a Python-level defect by matching its message text against provider conditions +- Answering "the caller's session" with "the most recently active session", or letting a client adopt a session id the daemon happened to report - Bare `except:` clauses — always specify the exception type - `# TODO: implement` stubs — implement or don't commit diff --git a/src/leapflow/cli/commands/interactive.py b/src/leapflow/cli/commands/interactive.py index 9753ac3..63da6d8 100644 --- a/src/leapflow/cli/commands/interactive.py +++ b/src/leapflow/cli/commands/interactive.py @@ -305,7 +305,7 @@ def _status(message: str) -> None: mock_host=self._mock_host, status_callback=_status, ) - status = await self.client.status() + status = await self.client.status(str(self._session_id_getter() or "")) self._metadata_applier(status) session_id = str(self._session_id_getter() or "") if session_id: @@ -993,8 +993,20 @@ def _apply_daemon_runtime_metadata(metadata: dict[str, Any]) -> None: nonlocal runtime_turn_active, runtime_turn_max, runtime_turn_waiting if metadata.get("pid"): runtime_daemon_pid = str(metadata["pid"]) - if metadata.get("session_id"): - active_session_id = str(metadata["session_id"]) + # A session id is this client's own identity, so it is only ever accepted + # from the daemon when we have none yet (first assignment, or right after + # an explicit resume). A daemon serving several TUIs can report whichever + # session it resolved; adopting that made a second TUI send another + # workspace's session id and get rejected on every turn. + reported_session = str(metadata.get("session_id") or "") + if reported_session and reported_session != active_session_id: + if active_session_id: + logger.debug( + "ignoring daemon-reported session %s; this client owns %s", + reported_session, active_session_id, + ) + else: + active_session_id = reported_session if metadata.get("model"): runtime_model_name = str(metadata["model"]) if metadata.get("llm_model"): @@ -1078,7 +1090,7 @@ def _render_banner() -> None: async def _print_daemon_status() -> None: try: daemon_status = await bridge.call( - lambda current_client: current_client.status(), + lambda current_client: current_client.status(active_session_id), description="daemon status", ) except Exception as exc: @@ -1487,7 +1499,7 @@ def _handle_task_control(text: str) -> bool: try: _apply_daemon_runtime_metadata(await bridge.call( - lambda current_client: current_client.status(), + lambda current_client: current_client.status(active_session_id), description="daemon status", )) except Exception as exc: diff --git a/src/leapflow/daemon/client.py b/src/leapflow/daemon/client.py index e916afe..4142a00 100644 --- a/src/leapflow/daemon/client.py +++ b/src/leapflow/daemon/client.py @@ -143,9 +143,16 @@ async def session_resume(self, session_id: str) -> dict[str, Any]: result = await self.request("session.resume", {"session_id": session_id}) return dict(result or {}) - async def status(self) -> dict[str, Any]: - """Return daemon status.""" - result = await self.request("daemon.status") + async def status(self, session_id: str = "") -> dict[str, Any]: + """Return daemon status, scoped to ``session_id`` when the caller has one. + + Passing the caller's own session is what makes the reply describe *this* + client: without it the daemon has no way to know which of several live + sessions to report, and any session identity it returned would belong to + somebody else. + """ + params = {"session_id": session_id} if session_id else {} + result = await self.request("daemon.status", params) return dict(result or {}) async def host_status(self) -> dict[str, Any]: diff --git a/src/leapflow/daemon/protocol.py b/src/leapflow/daemon/protocol.py index 73bb473..ef34fbe 100644 --- a/src/leapflow/daemon/protocol.py +++ b/src/leapflow/daemon/protocol.py @@ -235,8 +235,13 @@ async def session_analyze(self) -> Dict[str, Any]: """Ensure a session-analysis watch and run one analysis cycle now.""" ... - async def status(self) -> Dict[str, Any]: - """Return daemon status (uptime, connections, db path, etc.).""" + async def status(self, session_id: str = "") -> Dict[str, Any]: + """Return daemon status (uptime, connections, db path, etc.). + + ``session_id`` scopes the reply to the caller's session; without it no + session identity or per-session context figures are reported, since any + the daemon picked would belong to a different client. + """ ... async def host_status(self) -> Dict[str, Any]: diff --git a/src/leapflow/daemon/service.py b/src/leapflow/daemon/service.py index a96c412..0c13f64 100644 --- a/src/leapflow/daemon/service.py +++ b/src/leapflow/daemon/service.py @@ -451,14 +451,17 @@ async def _stream_engine_events( logger.debug("daemon: failed to close engine stream", exc_info=True) def _active_engine(self, session_id: str = "") -> Any: - """Return the engine holding live conversation state. - - Single entry point for "the engine to report on". ``ctx.engine`` is only a - template used to build per-session engines and never accumulates a - conversation, so reading it yields zero turns and zero context — which - has surfaced repeatedly as an empty LeapBoard and a status bar stuck at - ``0/``. Anything assembling runtime metadata must come through - here rather than reaching for ``ctx.engine`` directly. + """Return the engine holding a caller's live conversation state. + + ``ctx.engine`` is only a template used to build per-session engines and + never accumulates a conversation, so reading it yields zero turns and zero + context — which has surfaced repeatedly as an empty LeapBoard and a status + bar stuck at ``0/``. Anything assembling runtime metadata must come + through here rather than reaching for ``ctx.engine`` directly. + + Pass the caller's ``session_id``. Omitting it resolves "whichever session + was most recently active", which on a daemon serving several workspaces is + somebody else's — acceptable only for genuinely cross-session views. """ ctx = self._ctx if ctx is None: @@ -467,25 +470,25 @@ def _active_engine(self, session_id: str = "") -> Any: return engine def _chunk_from_event( - self, event: StreamEvent, *, request_id: str = "", engine: Any = None, + self, event: StreamEvent, *, request_id: str = "", engine: Any, ) -> StreamChunk: """Wrap an engine stream event as an RPC chunk with runtime metadata. - ``engine`` must be the engine that produced the event — the per-session - one. Falling back to ``ctx.engine`` reports the base engine, which never - carries a conversation, so context usage reads as 0 and the client's - status bar sits at ``0/`` for the whole session. + ``engine`` is required and must be the engine that produced the event — + the per-session one. There is deliberately no fallback: resolving "some + active engine" would report the base engine (no conversation, so context + reads 0) or, on a daemon serving several TUIs, another client's session — + whose id the client would then adopt as its own. """ ctx = self.context - active = engine if engine is not None else self._active_engine() metadata = dict(event.metadata or {}) - session_id = getattr(active, "_current_session_id", "") if active else "" + session_id = getattr(engine, "_current_session_id", "") if engine is not None else "" if request_id: metadata.setdefault("request_id", request_id) if session_id: metadata.setdefault("session_id", str(session_id)) - if active is not None: - metadata.update(engine_context_metadata(active, getattr(ctx, "settings", self._settings))) + if engine is not None: + metadata.update(engine_context_metadata(engine, getattr(ctx, "settings", self._settings))) return StreamChunk( request_id=request_id, content=event.content, done=False, event_type=event.type, metadata=metadata, @@ -696,12 +699,14 @@ async def subscribe_notifications(self) -> AsyncIterator[StreamChunk]: # ── Status ─────────────────────────────────────────────────────── - async def status(self) -> dict[str, Any]: + async def status(self, session_id: str = "") -> dict[str, Any]: ctx = self._ctx settings = getattr(ctx, "settings", self._settings) if ctx is not None else self._settings - # Report on the session engine, not the base template: the latter has no - # conversation, so context usage would always read as zero. - engine = self._active_engine() + # Report on the caller's session engine, not the base template (which has + # no conversation, so context usage would read zero) and not on whichever + # session happened to run last (which on a multi-workspace daemon belongs + # to another client, and whose id that client would then adopt). + engine = self._active_engine(session_id) if session_id else None db_holder = getattr(ctx, "_db_holder", None) if ctx is not None else None layout = settings.layout profile_layout = settings.profile_layout @@ -743,7 +748,7 @@ async def status(self) -> dict[str, Any]: "compression_reason": context_metadata.get("compression_reason", ""), "compression_savings_ratio": context_metadata.get("compression_savings_ratio", 0.0), "context_budget_snapshot": context_metadata.get("context_budget_snapshot", {}), - "session_id": str(getattr(engine, "_current_session_id", "") or ""), + "session_id": str(getattr(engine, "_current_session_id", "") or "") if engine is not None else "", "runtime_source": runtime_source(), "runtime_executable": sys.executable, "runtime_version": runtime_version(), diff --git a/src/leapflow/daemon/session_coordinator.py b/src/leapflow/daemon/session_coordinator.py index 98368ca..eb610ef 100644 --- a/src/leapflow/daemon/session_coordinator.py +++ b/src/leapflow/daemon/session_coordinator.py @@ -113,7 +113,10 @@ def resolve_session_engine(self, ctx: Any, session_id: str = "") -> tuple[Any, s cause of an empty LeapBoard. Resolution order: 1. the requested ``session_id``, when that session is live; - 2. otherwise the most recently active session ("the current session"); + 2. otherwise the most recently active session of *any* client — valid only + for cross-session views, never for describing the caller (it leaks one + client's session to another; callers that represent a specific client + must pass that client's id); 3. finally the base engine, for in-process mode where it *is* the engine. """ base = getattr(ctx, "engine", None) if ctx is not None else None @@ -121,7 +124,7 @@ def resolve_session_engine(self, ctx: Any, session_id: str = "") -> tuple[Any, s if registry is not None: session_ctx = registry.get(session_id) if session_id else None if session_ctx is None and not session_id: - session_ctx = registry.most_recent() + session_ctx = registry.most_recent_any_client() if session_ctx is not None: return session_ctx.engine, str(session_ctx.session_id or "") if base is None: diff --git a/src/leapflow/daemon/session_registry.py b/src/leapflow/daemon/session_registry.py index 5a68eaf..2bc22aa 100644 --- a/src/leapflow/daemon/session_registry.py +++ b/src/leapflow/daemon/session_registry.py @@ -26,12 +26,20 @@ class WorkspaceMismatchError(ValueError): - """Raised when one session id is reused from a different workspace root.""" + """Raised when one session id is reused from a different workspace root. + + Reaching this from ordinary use means a client sent a session id it does not + own — a defect, not something the user can act on. The only legitimate cause + is an explicit ``--resume`` of a session created in another workspace, which + is why the guidance names that case instead of telling the user to do what + they already did. + """ def __init__(self, session_id: str, expected: Path, requested: Path) -> None: super().__init__( - f"Session {session_id!r} is bound to workspace {expected}; " - f"current request uses {requested}. Start a fresh TUI session for this workspace." + f"Session {session_id!r} belongs to workspace {expected}, but this request " + f"came from {requested}. If you resumed it with --resume, resume it from " + f"{expected} instead, or omit --resume to start a session for this workspace." ) self.session_id = session_id self.expected = expected @@ -159,12 +167,14 @@ def get(self, session_id: str) -> Optional[SessionExecutionContext]: """ return self._contexts.get(str(session_id or "")) - def most_recent(self) -> Optional[SessionExecutionContext]: - """Return the most recently active session context, if any. + def most_recent_any_client(self) -> Optional[SessionExecutionContext]: + """Return the most recently active session of *any* client, if any. - Defines "the current session" for consumers that have no session id of - their own. Since the base engine never holds conversation state, this is - the only meaningful fallback for observing live activity. + Named for what it actually does. It ignores workspace and client + identity, so it is only valid for genuinely cross-session views (an + aggregate dashboard). Using it to answer "the caller's session" leaks one + client's session id and context figures to another, which is how a second + TUI ended up sending a session bound to a different workspace. """ if not self._contexts: return None diff --git a/tests/test_board_session_binding.py b/tests/test_board_session_binding.py index 3348648..a659ae2 100644 --- a/tests/test_board_session_binding.py +++ b/tests/test_board_session_binding.py @@ -59,17 +59,17 @@ def test_registry_get_does_not_create_and_most_recent_tracks_activity() -> None: registry = _registry(base) assert registry.get("absent") is None - assert registry.most_recent() is None + assert registry.most_recent_any_client() is None assert registry.active_count() == 0 # lookups must not materialize engines first = asyncio.run(registry.acquire("s1", workspace_root="/tmp")) second = asyncio.run(registry.acquire("s2", workspace_root="/tmp")) assert registry.get("s1") is first - # s2 was acquired last, so it is the current session. - assert registry.most_recent() is second + # s2 was acquired last, so it is the most recent across all clients. + assert registry.most_recent_any_client() is second first.touch() - assert registry.most_recent() is first + assert registry.most_recent_any_client() is first # ── get_history resolves the right engine ──────────────────────────── diff --git a/tests/test_cli_entrypoint.py b/tests/test_cli_entrypoint.py index 58d4674..c65b1e6 100644 --- a/tests/test_cli_entrypoint.py +++ b/tests/test_cli_entrypoint.py @@ -678,14 +678,18 @@ def success(self, message: str) -> None: self.successes.append(message) class BrokenClient: - async def status(self): + async def status(self, session_id: str = ""): raise DaemonUnavailableError("socket disappeared") class RecoveredClient: def __init__(self) -> None: self.resumed: list[str] = [] + self.status_sessions: list[str] = [] - async def status(self): + async def status(self, session_id: str = ""): + # The client scopes status to its own session; the daemon cannot + # otherwise know which of several live sessions to describe. + self.status_sessions.append(session_id) return {"pid": 99, "session_id": "sess-1"} async def session_resume(self, session_id: str): diff --git a/tests/test_daemon_rpc.py b/tests/test_daemon_rpc.py index 3cbd01f..7c66e8b 100644 --- a/tests/test_daemon_rpc.py +++ b/tests/test_daemon_rpc.py @@ -1714,7 +1714,11 @@ async def collect(message: str) -> list: return [event async for event in service.engine_chat(message)] first, second = await asyncio.gather(collect("one"), collect("two")) - status = await service.status() + # Status is scoped to the caller's session: a session-less call cannot know + # which of several live sessions to describe, and reporting "the most recent" + # would hand one client another's session id and context figures. + status = await service.status("sess-daemon") + anonymous_status = await service.status() # The blocked turn now receives an immediate 'queued' status chunk before it # acquires the engine lock; filter those out to assert on the engine stream. @@ -1736,6 +1740,11 @@ async def collect(message: str) -> list: assert first_engine[0].metadata["session_id"] == "sess-daemon" assert second_engine[0].metadata["context_used"] == 2_048 assert status["context_used"] == 2_048 + assert status["session_id"] == "sess-daemon" + # A caller that names no session gets no session identity and no per-session + # figures, rather than somebody else's. + assert anonymous_status["session_id"] == "" + assert anonymous_status["context_used"] == 0 assert status["context_budget_snapshot"]["total_tokens"] == 2_048 assert status["context_posture"] == "research" assert status["turn_admission"]["max_concurrent"] == 1 diff --git a/tests/test_multi_client_session_isolation.py b/tests/test_multi_client_session_isolation.py new file mode 100644 index 0000000..42eb9bb --- /dev/null +++ b/tests/test_multi_client_session_isolation.py @@ -0,0 +1,193 @@ +"""Isolation contracts for two TUI clients on one daemon. + +Written after two TUIs in different workspaces became unusable in the second one: + + TUI#2 (workspace B) polls daemon status + -> status() took no session id and resolved "the most recently active + session", which was TUI#1's (workspace A) + -> the reply carried A's session_id, and the client adopted it as its own + -> TUI#2's next turn sent A's session id with workspace B + -> the workspace guard correctly rejected it, on every turn, forever + -> the advice ("start a fresh TUI session") could not work, because a fresh + TUI re-adopts the same id on its first status poll + +The rule these pin down: a session id belongs to the client that created it. The +daemon may resolve *metrics* per request; it may never hand one client another +client's *identity*, and a client may never adopt one. +""" + +from __future__ import annotations + +import asyncio +import inspect +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from leapflow.daemon.session_registry import SessionRegistry, WorkspaceMismatchError + +_WORKSPACE_A = "/tmp/leapflow-test-workspace-a" +_WORKSPACE_B = "/tmp/leapflow-test-workspace-b" +# The registry resolves roots, and on macOS /tmp is a symlink to /private/tmp. +_ROOT_A = Path(_WORKSPACE_A).resolve() +_ROOT_B = Path(_WORKSPACE_B).resolve() + + +class _SessionEngine: + """Minimal stand-in for a per-session engine.""" + + def __init__(self, session_id: str, used: int = 0) -> None: + self._current_session_id = session_id + self.context_token_count = used + self.turn_count = 1 + + +def _registry(base: object, used: dict[str, int] | None = None) -> SessionRegistry: + usage = used or {} + return SessionRegistry( + base_engine=base, + build_engine=lambda b, sid, wm, root: _SessionEngine(sid, usage.get(sid, 0)), + build_working_memory=lambda: None, + ) + + +# ── The registry keeps sessions bound to their workspace ───────────────── + +def test_two_workspaces_get_independent_sessions() -> None: + registry = _registry(_SessionEngine("")) + + a = asyncio.run(registry.acquire("sess-a", workspace_root=_WORKSPACE_A)) + b = asyncio.run(registry.acquire("sess-b", workspace_root=_WORKSPACE_B)) + + assert a is not b + assert a.workspace_root == _ROOT_A + assert b.workspace_root == _ROOT_B + assert registry.active_count() == 2 + + +def test_reusing_a_session_from_another_workspace_is_refused() -> None: + """The guard itself is correct and must stay: sessions are workspace-bound.""" + registry = _registry(_SessionEngine("")) + asyncio.run(registry.acquire("sess-a", workspace_root=_WORKSPACE_A)) + + with pytest.raises(WorkspaceMismatchError) as excinfo: + asyncio.run(registry.acquire("sess-a", workspace_root=_WORKSPACE_B)) + + error = excinfo.value + assert error.expected == _ROOT_A + assert error.requested == _ROOT_B + # The guidance must name the only legitimate cause instead of telling the + # user to start a fresh session, which is what they already did. + assert "--resume" in str(error) + + +def test_cross_client_fallback_is_named_for_what_it_does() -> None: + """Renamed so it cannot be mistaken for "the caller's session". + + The old name (`most_recent`) read like "the current session" and was used to + answer status calls, which is how the leak happened. + """ + registry = _registry(_SessionEngine("")) + assert not hasattr(registry, "most_recent") + assert hasattr(registry, "most_recent_any_client") + + +# ── The daemon must not report a session the caller did not name ───────── + +def _service_with_two_sessions(tmp_path): + """A daemon service with two live sessions in two different workspaces.""" + from conftest import make_settings + from leapflow.daemon.service import RuntimeLeapService + + base = _SessionEngine("", 0) + registry = _registry(base, used={"sess-a": 1_111, "sess-b": 2_222}) + asyncio.run(registry.acquire("sess-a", workspace_root=_WORKSPACE_A)) + # sess-b is acquired last, so any "most recent" fallback resolves to it. + asyncio.run(registry.acquire("sess-b", workspace_root=_WORKSPACE_B)) + service = RuntimeLeapService(make_settings(str(tmp_path)), mock_host=True) + service._ctx = SimpleNamespace( + engine=base, + settings=make_settings(str(tmp_path)), + reload_runtime_config_if_changed=lambda: False, + ) + service._session_coordinator._session_registry = registry + return service + + +def test_status_scoped_to_the_caller_never_reports_another_session(tmp_path) -> None: + service = _service_with_two_sessions(tmp_path) + + status_a = asyncio.run(service.status("sess-a")) + + assert status_a["session_id"] == "sess-a" + assert status_a["context_used"] == 1_111, "must not report sess-b's usage" + + +def test_status_without_a_session_reports_no_identity(tmp_path) -> None: + """A caller that names no session gets none, not the most recent one.""" + service = _service_with_two_sessions(tmp_path) + + status = asyncio.run(service.status()) + + assert status["session_id"] == "" + assert status["context_used"] == 0 + + +def test_status_accepts_a_session_id_over_the_rpc() -> None: + """The parameter must exist on both sides of the wire.""" + from leapflow.daemon.client import DaemonClient + from leapflow.daemon.protocol import LeapService + + for target in (DaemonClient.status, LeapService.status): + assert "session_id" in inspect.signature(target).parameters, target + + +# ── The client must not adopt an identity it did not ask for ───────────── + +def _metadata_applier(initial_session: str): + """Build the TUI's metadata applier in isolation, returning a probe. + + Mirrors `_apply_daemon_runtime_metadata`'s session-adoption rule. Exercised + through the real TUI entry point in the integration test below; this keeps the + rule itself assertable without standing up a terminal. + """ + state = {"session_id": initial_session} + + def apply(metadata: dict) -> None: + reported = str(metadata.get("session_id") or "") + if reported and reported != state["session_id"]: + if not state["session_id"]: + state["session_id"] = reported + + return apply, state + + +def test_client_keeps_its_own_session_when_the_daemon_reports_another() -> None: + apply, state = _metadata_applier("sess-b") + + apply({"session_id": "sess-a", "context_used": 1_111}) + + assert state["session_id"] == "sess-b", "a client's identity is its own" + + +def test_client_adopts_a_session_only_when_it_has_none() -> None: + apply, state = _metadata_applier("") + + apply({"session_id": "sess-a"}) + + assert state["session_id"] == "sess-a" + + +def test_tui_adoption_rule_is_the_one_shipped() -> None: + """Guard the real implementation, not just the mirrored rule above. + + The outage was a single unconditional assignment; this fails if it returns. + """ + from pathlib import Path as _Path + + import leapflow.cli.commands.interactive as interactive + + source = _Path(interactive.__file__).read_text(encoding="utf-8") + assert 'active_session_id = str(metadata["session_id"])' not in source + assert "reported_session != active_session_id" in source diff --git a/tests/test_runtime_metadata_and_wrapping.py b/tests/test_runtime_metadata_and_wrapping.py index 74bfcb4..a206919 100644 --- a/tests/test_runtime_metadata_and_wrapping.py +++ b/tests/test_runtime_metadata_and_wrapping.py @@ -77,11 +77,17 @@ def test_active_engine_resolves_the_session_engine() -> None: def test_stream_metadata_reports_real_context_usage() -> None: - """The status bar reads this; zero here is what showed as ``0/1M``.""" + """The status bar reads this; zero here is what showed as ``0/1M``. + + The producing engine is passed explicitly. There is no fallback to "whichever + engine looks active": on a daemon serving several TUIs that resolves another + client's session, and the client adopts the reported id as its own. + """ service, _ = _service_with_session() + session_engine = service._active_engine("s1") chunk = service._chunk_from_event( - StreamEvent(type="content", content="hi"), request_id="r1", + StreamEvent(type="content", content="hi"), request_id="r1", engine=session_engine, ) assert chunk.metadata["context_used"] == _USED_TOKENS @@ -89,6 +95,24 @@ def test_stream_metadata_reports_real_context_usage() -> None: assert chunk.metadata["session_id"] == "s1" +def test_chunk_metadata_requires_the_producing_engine() -> None: + """No engine means no session identity may be attached. + + Guarding the shape rather than the value: an optional engine is what allowed + a foreign session id into a client's metadata. + """ + import inspect + + service, _ = _service_with_session() + signature = inspect.signature(service._chunk_from_event) + assert signature.parameters["engine"].default is inspect.Parameter.empty + + chunk = service._chunk_from_event( + StreamEvent(type="content", content="hi"), request_id="r1", engine=None, + ) + assert "session_id" not in chunk.metadata + + def test_stream_metadata_prefers_the_engine_that_produced_the_event() -> None: """An explicit engine wins, so a concurrent session cannot be misreported.""" service, _ = _service_with_session()