feat(core): Implement Budget-Aware Recovery Subsystem, Tool Idempotency, and Core Stability Fixes

AGENTS.md

@@ -23,6 +23,7 @@ This document is the LeapFlow engineering collaboration contract. It is not only
 - Design for generalization and universality; prefer reusable domain concepts over one-off special cases
 - Easy to extend, avoid hardcoding and hard rules
 - Industrial-grade robustness: every external call has timeout, retry, and fallback
+- Structured error propagation: failures must flow as typed envelopes (`FailureEnvelope`) with source, category, recoverability, and side-effect state — never as ad-hoc dicts, bare strings, or unclassified exceptions
 - User experience is a first-class quality bar: optimize for clarity, ease of use, fast feedback, and graceful recovery
 - All comments and docstrings in English
 - Type annotations on all public APIs
@@ -48,6 +49,10 @@ This document is the LeapFlow engineering collaboration contract. It is not only
 - **Graceful Degradation**: Every optional component (LLM, Hub) can be absent without crash
 - **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.
+- **Side-Effect Gating**: Recovery actions are gated by `SideEffectState`. Committed or partial side effects block automatic retry; only user-mediated or checkpoint-based resumption is permitted after state mutation.
+- **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.
+- **Recovery Strategy as Protocol**: Recovery strategies implement a `RecoveryStrategy` Protocol (`can_apply` + `decide`), registered by priority, composable, and extensible without modifying the coordinator.
 
 ## Path Tree, Configuration, and Secrets Rules
 
@@ -81,6 +86,9 @@ This document is the LeapFlow engineering collaboration contract. It is not only
 - Every module must be importable standalone without side effects
 - No placeholder stubs — implement fully or do not add the code
 - ANSI output must check `sys.stdout.isatty()` before emitting escape codes
+- For error recovery, route all failures through the `RecoveryCoordinator` — classify into a `FailureEnvelope`, receive a `RecoveryDecision` with an explainable `reason` and `strategy_key`, then feed the outcome back. Never handle errors with ad-hoc if/break in the loop body.
+- Recovery strategies are standalone Protocol implementations with `can_apply()` + `decide()`. Add new strategies by registration, never by modifying the coordinator's decision logic.
+- When automatic recovery exhausts its budget or encounters non-recoverable failures, emit a structured `InteractionRequest` (typed action, severity, suggested actions, timeout behavior, resumption key) — not raw text appended to conversation.
 
 ## Review Requirements
 
@@ -103,6 +111,8 @@ This document is the LeapFlow engineering collaboration contract. It is not only
 - **Behavior contracts over snapshots**: assert invariants, not frozen values
 - **Mock at boundaries only**: mock external I/O (network, disk), never internal logic
 - **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.
 
 ## What to Avoid
 
@@ -118,6 +128,10 @@ This document is the LeapFlow engineering collaboration contract. It is not only
 - Activating IM agents on all inbound messages by default, skipping self-message filtering, or allowing cross-chat/proactive sends before Progressive Trust and approval policies explicitly permit them.
 - Putting third-party app business code into platform core: do not add vendor scopes, lark-cli/SDK JSON parsing, auth command construction, console-specific recovery instructions, or resource-specific branching to gateway-wide modules such as action registries, capability ledgers, approval gates, or engine recovery paths.
 - Shortcut-style natural-language fitting and large intent-handler taxonomies; use stable runtime gates plus capability manifests instead. Rule-based keyword/action-verb/alias matching is prohibited by default and requires explicit human second confirmation before implementation when unavoidable.
+- Scattered if/break/continue recovery decisions inside the agent loop body; all error handling enters the `RecoveryCoordinator` as a `FailureEnvelope`
+- 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
 - Bare `except:` clauses — always specify the exception type
 - `# TODO: implement` stubs — implement or don't commit
 

src/leapflow/cli/tui_app/app.py

@@ -54,7 +54,7 @@
 from prompt_toolkit.widgets import TextArea
 
 from leapflow.cli.tui_app.approval_modal import ApprovalModal, request_is_expired
-from leapflow.cli.tui_app.command import TuiCommand, TuiCommandStatus
+from leapflow.cli.tui_app.command import TuiCommand, TuiCommandStatus, command_key
 from leapflow.cli.tui_app.input import build_completer
 from leapflow.cli.tui_app.paste import (
     PASTE_FRAGMENT_WINDOW_S,
@@ -159,6 +159,9 @@ def get_nowait(self) -> TuiCommand:
     def qsize(self) -> int:
         return len(self._items)
 
+    def contains_key(self, key: str) -> bool:
+        return any(command.command_key == key for command in self._items)
+
     def snapshot(self) -> list[TuiCommand]:
         return list(self._items)
 
@@ -283,8 +286,15 @@ def submit_text(self, text: str) -> TuiCommand:
             raise ValueError("Cannot submit an empty TUI command")
         if self._dispatch_control_text(normalized):
             return TuiCommand.create(command_id=0, text=normalized).mark_done()
+        key = command_key(normalized)
         command = TuiCommand.create(command_id=self._next_command_id, text=normalized)
         self._next_command_id += 1
+        if self._is_duplicate_command_key(key):
+            skipped = command.mark_skipped("duplicate queued/running command skipped")
+            self._console.command_card(skipped)
+            self._sync_task_counts()
+            self._invalidate()
+            return skipped
         self._pending_input.put_nowait(command)
         self._sync_task_counts()
         if self._active_command is not None or self._pending_input.qsize() > 1 or self._queue_paused:
@@ -422,6 +432,10 @@ def _terminal_command(
             return command.mark_blocked(reason)
         return command.mark_failed(reason)
 
+    def _is_duplicate_command_key(self, key: str) -> bool:
+        active = self._active_command
+        return bool((active is not None and active.command_key == key) or self._pending_input.contains_key(key))
+
     def _dispatch_control_text(self, text: str) -> bool:
         handler = self._on_control
         if handler is None:

src/leapflow/cli/tui_app/command.py

@@ -7,6 +7,7 @@
 
 from __future__ import annotations
 
+import hashlib
 import time
 from dataclasses import dataclass, replace
 from enum import Enum
@@ -25,6 +26,11 @@ def _truncate(text: str, limit: int) -> str:
     return text if len(text) <= limit else text[: limit - 1] + "…"
 
 
+def command_key(text: str) -> str:
+    """Return a stable identity key for equivalent submitted commands."""
+    return hashlib.sha256(_single_line(text).encode("utf-8")).hexdigest()
+
+
 class TuiCommandStatus(str, Enum):
     """Lifecycle states for a TUI-submitted command."""
 
@@ -45,6 +51,7 @@ class TuiCommand:
     text: str
     status: TuiCommandStatus
     created_at: float
+    command_key: str
     started_at: float = 0.0
     finished_at: float = 0.0
     error: str = ""
@@ -57,6 +64,7 @@ def create(cls, *, command_id: int, text: str) -> "TuiCommand":
             text=text,
             status=TuiCommandStatus.QUEUED,
             created_at=time.monotonic(),
+            command_key=command_key(text),
         )
 
     @property

src/leapflow/cli/tui_app/stream.py

@@ -269,6 +269,8 @@ def _tool_icon(name: str, *, ok: bool = True) -> str:
 def _tool_result_detail(metadata: dict[str, Any] | None) -> str:
     if not metadata:
         return ""
+    if metadata.get("already_executed") or metadata.get("duplicate_suppressed") or metadata.get("execution_skipped"):
+        return ""
     if metadata.get("ok") is False:
         exit_code = metadata.get("exit_code")
         prefix = f"exit={exit_code} " if exit_code is not None else ""
@@ -412,6 +414,11 @@ def tool_finished(
         tool_name = _metadata_text(metadata, "normalized_tool_name") or name or self._active_tool
         original_tool_name = _metadata_text(metadata, "original_tool_name")
         alias_detail = original_tool_name if original_tool_name and original_tool_name != tool_name else ""
+        if tool_name and metadata.get("ui_hidden"):
+            self._active_tool = ""
+            self._active_tool_detail = ""
+            self._tool_start_time = 0.0
+            return
         if tool_name and self._tool_start_time > 0:
             duration = time.monotonic() - self._tool_start_time
             self._tool_history.append((tool_name, duration))

src/leapflow/config.py

@@ -349,6 +349,8 @@ class Settings:
     # ── Stream & Tool Robustness ──
     stale_stream_timeout_s: float = 180.0  # Idle timeout for streaming responses
     default_tool_timeout_s: float = 120.0  # Default per-tool execution timeout
+    daemon_request_ledger_ttl_s: float = 600.0  # Replay cache retention for completed engine requests
+    daemon_request_ledger_max_entries: int = 128  # Maximum completed engine requests kept for replay
     circuit_breaker_threshold: int = 5  # Consecutive failures before circuit opens
     circuit_breaker_cooldown_s: float = 60.0  # Circuit breaker cooldown period
 
@@ -756,6 +758,8 @@ def _build_settings_from_env(
     # Stream & Tool Robustness
     stale_stream_timeout_s = float(os.getenv("LEAPFLOW_STALE_STREAM_TIMEOUT_S", "180.0"))
     default_tool_timeout_s = float(os.getenv("LEAPFLOW_DEFAULT_TOOL_TIMEOUT_S", "120.0"))
+    daemon_request_ledger_ttl_s = float(os.getenv("LEAPFLOW_DAEMON_REQUEST_LEDGER_TTL_S", "600.0"))
+    daemon_request_ledger_max_entries = int(os.getenv("LEAPFLOW_DAEMON_REQUEST_LEDGER_MAX_ENTRIES", "128"))
     circuit_breaker_threshold = int(os.getenv("LEAPFLOW_CIRCUIT_BREAKER_THRESHOLD", "5"))
     circuit_breaker_cooldown_s = float(os.getenv("LEAPFLOW_CIRCUIT_BREAKER_COOLDOWN_S", "60.0"))
 
@@ -1018,6 +1022,8 @@ def _build_settings_from_env(
         # Stream & Tool Robustness
         stale_stream_timeout_s=stale_stream_timeout_s,
         default_tool_timeout_s=default_tool_timeout_s,
+        daemon_request_ledger_ttl_s=daemon_request_ledger_ttl_s,
+        daemon_request_ledger_max_entries=daemon_request_ledger_max_entries,
         circuit_breaker_threshold=circuit_breaker_threshold,
         circuit_breaker_cooldown_s=circuit_breaker_cooldown_s,
         # Signal Fusion

src/leapflow/daemon/server.py

@@ -156,6 +156,8 @@ async def _dispatch(self, request: RpcRequest, writer: asyncio.StreamWriter) ->
 
         method = getattr(self._service, attr)
         params = dict(request.params or {})
+        if request.method == "engine.chat":
+            params.setdefault("request_id", request.id)
         if request.method in ("engine.chat", "events.subscribe"):
             await self._dispatch_stream(request, method, params, writer)
             return