[Feature and Refactor] Refactor the core-loop and TUI

.env.example

@@ -186,6 +186,12 @@ LEAPFLOW_RECORDING_MODE=video
 # LEAPFLOW_COMPRESS_THRESHOLD=16
 # LEAPFLOW_COMPRESS_KEEP_TAIL=4
 # LEAPFLOW_MAX_TOOL_OUTPUT_CHARS=2000
+# LEAPFLOW_MAX_TOOL_RESULT_CHARS=3000
+# LEAPFLOW_CONTEXT_HARD_LIMIT_RATIO=0.92
+# LEAPFLOW_CONTEXT_WARNING_RATIO=0.75
+# LEAPFLOW_TOOL_EVIDENCE_MAX_CHARS=1200
+# LEAPFLOW_REPEATED_READ_LIMIT=2
+# LEAPFLOW_LONG_TASK_CONVERGENCE_ROUND=12
 # LEAPFLOW_ERROR_TRANSIENT_MAX_RETRIES=3
 # LEAPFLOW_ERROR_RATE_LIMIT_BASE_DELAY=5.0
 

AGENTS.md

@@ -1,5 +1,7 @@
 # AGENTS.md
 
+This document is the LeapFlow engineering collaboration contract. It is not only a style guide: it defines the design, runtime, UX, safety, and verification rules that every implementation change must follow.
+
 ## Design Philosophy
 
 1. **Signal-Driven Intelligence** — All agent intelligence derives from observing real-world signals, not from hardcoded rules. If a behavior cannot be learned from signals, it is not in scope.
@@ -12,57 +14,83 @@
 
 5. **LLM-Native Design** — Design for LLM reasoning first. Protocols over classes. Declarative over imperative. Context over configuration.
 
+6. **User-Centric Reliability** — User experience is part of correctness. Every change must keep common paths easy, predictable, recoverable, and must not degrade adjacent workflows.
+
 ## Code Quality Requirements
 
-- SOLID principles are non-negotiable
+- SOLID principles are non-negotiable; implementations must be cohesive, well-factored, and easy to reason about
 - Occam's Razor: maximize elegance and efficiency, reject unnecessary complexity
-- Design for generalization and universality
+- 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
+- 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
 - No bare except — always specify exception types
 
 ## Architecture Principles
 
+- **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.
+- **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.
+- **Progressive Context Disclosure (PCD)**: Keep one unified execution loop, but never default every turn to full disclosure. Each LLM call must use the smallest sufficient PromptAssemblyPlan for tools, memory, history, reasoning, streaming, and risk; upgrade progressively only when observable signals require it.
 - **Dependency Inversion**: Core logic depends on Protocol abstractions, never on concrete implementations
 - **Protocol over ABC**: Use `typing.Protocol` with `runtime_checkable` for all extension points
 - **Event-Driven Communication**: Modules interact through typed events on EventBus, not direct imports
 - **Immutable Domain Types**: Use `@dataclass(frozen=True)` or `NamedTuple` for domain objects
-- **Config-Driven Behavior**: Thresholds, intervals, feature flags — all configurable via env vars
-- **Graceful Degradation**: Every optional component (LLM, Hub, OS Host) can be absent without crash
+- **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.
+- **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
 
 ## Implementation Guidelines
 
 - Define the Protocol first — the contract is the design
 - Implement against the Protocol, never against another implementation
+- Consider affected user journeys before changing shared flows; do not introduce regressions, broken links, or worse experiences in adjacent paths
+- Keep common paths transparent: long-running work must stream progress, surface recoverable errors clearly, and avoid silent stalls.
+- For context assembly, prefer manifest-driven progressive disclosure over shortcuts or intent-handler sprawl: expose compact capability indexes, selected schemas, and targeted memory only when the current plan needs them.
+- Preserve security and audit paths: dangerous actions, file writes, outbound messages, credentials, and path access must flow through the existing policy, approval, redaction, and audit mechanisms.
+- Maintain backward-compatible migrations for persistent state, configuration, skills, trajectories, sessions, and profile data.
 - Write unit tests before or alongside the implementation
 - Integrate via EventBus events, not direct function calls between modules
 - 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
 
+## Review Requirements
+
+- **Deep review for large changes**: When a change substantially affects architecture, runtime behavior, user flows, persistence, safety, or multiple modules, perform an additional deep review before considering the work complete.
+- **Design goal check**: Verify that the implementation actually achieves the intended design goal and is not just a local patch.
+- **Optimality check**: Evaluate whether the solution is the simplest robust design, avoids unnecessary abstractions, and fits the existing architecture.
+- **Regression impact check**: Inspect affected modules and user journeys for logic bugs, degraded UX, broken compatibility, slower feedback, weaker diagnostics, or worse failure recovery.
+- **SOLID and extensibility check**: Look for responsibility leaks, tight coupling, hardcoded paths/thresholds/rules, magic strings, and choices that reduce generalization or future extension.
+- **Fix what the review finds**: If the review identifies correctness, design, UX, SOLID, hardcoding, or extensibility issues, fix and simplify them directly rather than only reporting them.
+
 ## Testing Philosophy
 
-- **Unit tests must be hermetic**: no network, no OS Host, no LLM calls
+- **Unit tests must be hermetic**: no network, no LLM calls
 - **py_compile all modified files**: syntax errors caught before test run
 - **Import chain verification**: every new module must be importable standalone
 - **Existing tests must not regress**: all tests must pass after every change
+- **User-facing flows must not regress**: preserve or improve usability, feedback clarity, and failure recovery for impacted paths
 - **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, OS Host), never internal logic
+- **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 or approval changes require security/approval tests; skills, learning, perception, and copilot changes require their lifecycle or pipeline tests.
 
 ## What to Avoid
 
 - Over-engineering: if you need 3+ files for a simple feature, rethink
 - Premature optimization: measure first, optimize only bottlenecks
-- God objects: no class should exceed 300 lines
+- God objects: no class should exceed 500 lines; approaching this limit requires checking whether policy, state, rendering, protocol, storage, or adapter responsibilities should be split.
 - Magic strings: use constants or enums
 - Blocking the event loop: all IO must be async or `run_in_executor`
 - Hardcoded paths, URLs, thresholds without config escape hatch
 - Chinese comments in source code (English only)
 - Speculative infrastructure: no hooks or extension points without a concrete consumer
+- Shortcut-style natural-language fitting and large intent-handler taxonomies; use stable runtime gates plus capability manifests instead.
 - Bare `except:` clauses — always specify the exception type
 - `# TODO: implement` stubs — implement or don't commit
 

src/leapflow/cli/approval_view.py

@@ -0,0 +1,229 @@
+"""Terminal approval view helpers for LeapFlow CLI/TUI surfaces."""
+from __future__ import annotations
+
+import asyncio
+import sys
+import textwrap
+import time
+from dataclasses import dataclass
+
+from leapflow.security.approval import ApprovalDecision, ApprovalRequest
+from leapflow.security.redact import redact_sensitive_text
+
+
+@dataclass(frozen=True)
+class ApprovalChoice:
+    """One selectable approval choice."""
+
+    key: str
+    label: str
+    decision: ApprovalDecision | None = None
+
+
+_CHOICE_LABELS = {
+    "allow_once": "Allow once",
+    "allow_session": "Allow for this session",
+    "allow_always": "Add to permanent allowlist",
+    "deny": "Deny",
+    "deny_always": "Deny for this session",
+    "show_details": "Show full details",
+}
+
+_CHOICE_DECISIONS = {
+    "allow_once": ApprovalDecision.ALLOW_ONCE,
+    "allow_session": ApprovalDecision.ALLOW_SESSION,
+    "allow_always": ApprovalDecision.ALLOW_ALWAYS,
+    "deny": ApprovalDecision.DENY,
+    "deny_always": ApprovalDecision.DENY_ALWAYS,
+}
+
+
+async def prompt_approval(request: ApprovalRequest) -> ApprovalDecision:
+    """Render an approval prompt and return a user decision."""
+    if not sys.stdin.isatty():
+        return ApprovalDecision.DENY
+
+    choices = build_approval_choices(request)
+    show_details = False
+    while True:
+        if _is_expired(request):
+            return ApprovalDecision.DENY
+        _render(request, choices, show_details=show_details)
+        try:
+            answer = await asyncio.wait_for(
+                asyncio.get_running_loop().run_in_executor(
+                    None, lambda: input("Select approval choice: ").strip().lower(),
+                ),
+                timeout=remaining_seconds(request),
+            )
+        except TimeoutError:
+            return ApprovalDecision.DENY
+        except (EOFError, KeyboardInterrupt):
+            return ApprovalDecision.DENY
+
+        selected = resolve_approval_choice(answer, choices)
+        if selected is None:
+            return ApprovalDecision.DENY
+        if selected.key == "show_details":
+            show_details = True
+            continue
+        return selected.decision or ApprovalDecision.DENY
+
+
+def build_approval_choices(request: ApprovalRequest) -> list[ApprovalChoice]:
+    """Build selectable approval choices for a request."""
+    keys = list(request.choices or ("allow_once", "allow_session", "deny"))
+    if "deny" not in keys:
+        keys.append("deny")
+    choices = []
+    for key in keys:
+        choices.append(ApprovalChoice(
+            key=key,
+            label=_CHOICE_LABELS.get(key, key.replace("_", " ").title()),
+            decision=_CHOICE_DECISIONS.get(key),
+        ))
+    return choices
+
+
+def resolve_approval_choice(answer: str, choices: list[ApprovalChoice]) -> ApprovalChoice | None:
+    """Resolve a typed approval answer to a selectable choice."""
+    if not answer:
+        return next((choice for choice in choices if choice.key == "deny"), None)
+    aliases = {
+        "y": "allow_once",
+        "yes": "allow_once",
+        "o": "allow_once",
+        "once": "allow_once",
+        "s": "allow_session",
+        "session": "allow_session",
+        "a": "allow_always",
+        "always": "allow_always",
+        "n": "deny",
+        "no": "deny",
+        "d": "deny",
+        "deny": "deny",
+        "v": "show_details",
+        "view": "show_details",
+        "full": "show_details",
+    }
+    key = aliases.get(answer, answer)
+    if answer.isdigit():
+        idx = int(answer) - 1
+        if 0 <= idx < len(choices):
+            return choices[idx]
+    return next((choice for choice in choices if choice.key == key), None)
+
+
+def _render(request: ApprovalRequest, choices: list[ApprovalChoice], *, show_details: bool) -> None:
+    title = str(request.display.get("title") or title_for_approval(request))
+    summary = str(request.display.get("summary") or request.category)
+    reason = str(request.display.get("reason") or risk_reason(request))
+    detail = redact_sensitive_text(request.detail, force=True)
+    if not show_details:
+        detail = truncate_detail(detail)
+
+    try:
+        from rich.console import Console
+        from rich.panel import Panel
+        from rich.text import Text
+
+        console = Console(stderr=True, highlight=False)
+        body = Text()
+        body.append(f"{summary}\n\n", style="bold")
+        body.append("Action detail:\n", style="dim")
+        body.append(_indent(detail) + "\n\n", style="yellow")
+        if reason:
+            body.append("Why approval is needed:\n", style="dim")
+            for line in textwrap.wrap(reason, width=72) or [reason]:
+                body.append(f"- {line}\n", style="dim")
+            body.append("\n")
+        remaining = remaining_seconds(request)
+        if remaining is not None:
+            body.append(f"Defaults to Deny in {int(remaining)}s.\n\n", style="dim")
+        for idx, choice in enumerate(choices, start=1):
+            body.append(f"  {idx}. {choice.label}\n", style="bold" if choice.key == request.default_choice else "")
+        console.print(Panel(
+            body,
+            title=f"[bold yellow]⚠ {title}[/]",
+            border_style="yellow",
+            padding=(0, 1),
+        ))
+    except ImportError:
+        sys.stderr.write(f"⚠ {title}\n\n{summary}\n\n{detail}\n\n")
+        if reason:
+            sys.stderr.write(f"Why approval is needed: {reason}\n\n")
+        remaining = remaining_seconds(request)
+        if remaining is not None:
+            sys.stderr.write(f"Defaults to Deny in {int(remaining)}s.\n\n")
+        for idx, choice in enumerate(choices, start=1):
+            sys.stderr.write(f"  {idx}. {choice.label}\n")
+        sys.stderr.flush()
+
+
+def title_for_approval(request: ApprovalRequest) -> str:
+    """Return the display title for an approval request."""
+    if request.risk is not None:
+        if request.risk.level.value == "high":
+            return "High Risk Action"
+        if request.risk.level.value == "critical":
+            return "Critical Action"
+    return "Action Approval"
+
+
+def risk_reason(request: ApprovalRequest) -> str:
+    """Return the human-readable risk reason for an approval request."""
+    if request.risk is None:
+        return ""
+    if request.risk.explanation:
+        return request.risk.explanation
+    return ", ".join(request.risk.reasons)
+
+
+def remaining_seconds(request: ApprovalRequest) -> float | None:
+    """Return seconds before approval expiry, if the request has a deadline."""
+    if request.expires_at is None:
+        return None
+    return max(0.0, float(request.expires_at) - time.time())
+
+
+def _is_expired(request: ApprovalRequest) -> bool:
+    remaining = remaining_seconds(request)
+    return remaining is not None and remaining <= 0.0
+
+
+def truncate_detail(text: str, *, max_lines: int = 6, width: int = 88) -> str:
+    """Truncate approval detail for compact rendering."""
+    wrapped: list[str] = []
+    for line in text.splitlines() or [text]:
+        wrapped.extend(textwrap.wrap(line, width=width, replace_whitespace=False) or [""])
+    if len(wrapped) <= max_lines:
+        return "\n".join(wrapped)
+    return "\n".join(wrapped[: max_lines - 1] + ["… (choose Show full details)"])
+
+
+def _build_choices(request: ApprovalRequest) -> list[ApprovalChoice]:
+    return build_approval_choices(request)
+
+
+def _resolve_choice(answer: str, choices: list[ApprovalChoice]) -> ApprovalChoice | None:
+    return resolve_approval_choice(answer, choices)
+
+
+def _title_for(request: ApprovalRequest) -> str:
+    return title_for_approval(request)
+
+
+def _risk_reason(request: ApprovalRequest) -> str:
+    return risk_reason(request)
+
+
+def _remaining_seconds(request: ApprovalRequest) -> float | None:
+    return remaining_seconds(request)
+
+
+def _truncate_detail(text: str, *, max_lines: int = 6, width: int = 88) -> str:
+    return truncate_detail(text, max_lines=max_lines, width=width)
+
+
+def _indent(text: str) -> str:
+    return "\n".join(f"  {line}" for line in text.splitlines())

src/leapflow/cli/banner.py

@@ -74,17 +74,24 @@ def _categorize_skills(
 _MAX_PANEL_WIDTH = 132
 _NARROW_WIDTH = 70
 _MEDIUM_WIDTH = 100
+_BANNER_ACCENT = "#FFBF00"
+_BANNER_ACCENT_DIM = "#B8860B"
+_BANNER_TEXT = "#FFF8DC"
+_BANNER_MUTED = "#8B8682"
+_BANNER_BORDER = "#CD7F32"
+_BANNER_TITLE = "bold #FFD700"
+_BANNER_SUCCESS = "#87D687"
 
 
 class _BannerPalette:
-    def __init__(self, theme: Theme | ResolvedTheme) -> None:
-        self.accent = theme.accent
-        self.accent_dim = theme.accent_dim
-        self.text = theme.text
-        self.muted = theme.text_muted
-        self.border = theme.border
-        self.title = theme.panel_title
-        self.success = theme.success
+    def __init__(self, _theme: Theme | ResolvedTheme) -> None:
+        self.accent = _BANNER_ACCENT
+        self.accent_dim = _BANNER_ACCENT_DIM
+        self.text = _BANNER_TEXT
+        self.muted = _BANNER_MUTED
+        self.border = _BANNER_BORDER
+        self.title = _BANNER_TITLE
+        self.success = _BANNER_SUCCESS
 
 
 def _trim(text: str, limit: int) -> str: