From a2d3cb7726bb1e861cbef56c6a18b72a9f49c06d Mon Sep 17 00:00:00 2001 From: fanqiNO1 <1848839264@qq.com> Date: Thu, 6 Aug 2026 20:50:23 +0800 Subject: [PATCH 1/4] fix leapd launch on windows --- src/leapflow/daemon/_transport.py | 10 ++++++++++ src/leapflow/daemon/lifecycle.py | 11 ++++++----- src/leapflow/storage/duckdb_connect.py | 13 ++++++++++++- tests/test_daemon_rpc.py | 16 ++++++++++------ tests/test_process_group.py | 1 + 5 files changed, 39 insertions(+), 12 deletions(-) diff --git a/src/leapflow/daemon/_transport.py b/src/leapflow/daemon/_transport.py index c4aa7dd..3dca2bd 100644 --- a/src/leapflow/daemon/_transport.py +++ b/src/leapflow/daemon/_transport.py @@ -33,6 +33,10 @@ async def connect(self, runtime_dir: Path) -> Tuple[asyncio.StreamReader, asynci def probe_healthy(self, runtime_dir: Path) -> bool: """Synchronous quick health check — returns True if the daemon is reachable.""" + @abc.abstractmethod + def readiness_path(self, runtime_dir: Path) -> Path: + """Artifact file the daemon publishes when it is ready to serve.""" + @abc.abstractmethod def cleanup(self, runtime_dir: Path) -> None: """Remove transport artifacts (socket file, port file, etc.).""" @@ -70,6 +74,9 @@ def probe_healthy(self, runtime_dir: Path) -> bool: except (OSError, socket.timeout): return False + def readiness_path(self, runtime_dir: Path) -> Path: + return self._sock_path(runtime_dir) + def cleanup(self, runtime_dir: Path) -> None: self._sock_path(runtime_dir).unlink(missing_ok=True) @@ -119,6 +126,9 @@ def probe_healthy(self, runtime_dir: Path) -> bool: except (OSError, socket.timeout): return False + def readiness_path(self, runtime_dir: Path) -> Path: + return self._port_path(runtime_dir) + def cleanup(self, runtime_dir: Path) -> None: self._port_path(runtime_dir).unlink(missing_ok=True) diff --git a/src/leapflow/daemon/lifecycle.py b/src/leapflow/daemon/lifecycle.py index ca3d04d..f94794f 100644 --- a/src/leapflow/daemon/lifecycle.py +++ b/src/leapflow/daemon/lifecycle.py @@ -48,12 +48,14 @@ class DaemonInfo: def discover(cls, runtime_dir: Path) -> DaemonInfo: """Probe the runtime directory to determine daemon state.""" pid = _read_pid(runtime_dir / "leapd.pid") - sock_path = runtime_dir / "leapd.sock" + from leapflow.daemon._transport import get_transport + + sock_path = get_transport().readiness_path(runtime_dir) meta = _read_meta(runtime_dir / "leapd.json") start_time = meta.get("start_time") if meta else None is_running = pid is not None and _process_alive(pid) - is_healthy = is_running and sock_path.exists() and _sock_healthy(sock_path) + is_healthy = is_running and _sock_healthy(runtime_dir) return cls( pid=pid, @@ -439,12 +441,11 @@ def _process_alive(pid: int) -> bool: kernel32.CloseHandle(handle) -def _sock_healthy(sock_path: Path) -> bool: +def _sock_healthy(runtime_dir: Path) -> bool: """Quick health check by connecting to the daemon transport.""" from leapflow.daemon._transport import get_transport - transport = get_transport() - return transport.probe_healthy(sock_path.parent) + return get_transport().probe_healthy(runtime_dir) def _format_duration(seconds: Optional[float]) -> str: diff --git a/src/leapflow/storage/duckdb_connect.py b/src/leapflow/storage/duckdb_connect.py index 3d34a46..c61cd57 100644 --- a/src/leapflow/storage/duckdb_connect.py +++ b/src/leapflow/storage/duckdb_connect.py @@ -24,7 +24,18 @@ logger = logging.getLogger(__name__) -LOCK_KEYWORDS = frozenset({"lock", "locked", "exclusive", "cannot set lock"}) +LOCK_KEYWORDS = frozenset({ + "lock", + "locked", + "exclusive", + "cannot set lock", + # Windows "file held by another process" forms — DuckDB raises these + # instead of a lock error when the DB file is open elsewhere (English + # and Chinese system locales). Must classify as locked, not corrupted. + "cannot open file", + "being used by another process", + "另一个程序正在使用", +}) _CONNECT_RETRIES = int(os.getenv("LEAPFLOW_DB_CONNECT_RETRIES", "5" if sys.platform == "win32" else "3")) _CONNECT_BACKOFF_S = float(os.getenv("LEAPFLOW_DB_CONNECT_BACKOFF_S", "1.0" if sys.platform == "win32" else "0.5")) diff --git a/tests/test_daemon_rpc.py b/tests/test_daemon_rpc.py index 2e4cda6..67793bd 100644 --- a/tests/test_daemon_rpc.py +++ b/tests/test_daemon_rpc.py @@ -11,6 +11,8 @@ import pytest +import leapflow +from leapflow.daemon._transport import get_transport from leapflow.daemon.client import DaemonClient, DaemonUnavailableError, ensure_daemon_client from leapflow.daemon.lifecycle import DaemonInfo from leapflow.daemon.protocol import StreamChunk @@ -75,7 +77,7 @@ async def engine_chat(self, message: str, **kwargs: Any) -> AsyncIterator[Stream class _SlowFirstChunkService(_FakeService): async def engine_chat(self, message: str, **kwargs: Any) -> AsyncIterator[StreamChunk]: - await asyncio.sleep(0.08) + await asyncio.sleep(0.16) yield StreamChunk(request_id="", content=f"slow {message}", event_type="chunk") yield StreamChunk(request_id="", content="done", event_type="final") @@ -128,8 +130,9 @@ async def _start_server(runtime_dir: Path, service=None, *, stream_heartbeat_s: stream_heartbeat_s=stream_heartbeat_s, ) task = asyncio.create_task(server.serve_forever()) + readiness = get_transport().readiness_path(runtime_dir) for _ in range(50): - if (runtime_dir / "leapd.sock").exists(): + if readiness.exists(): return server, task, runtime_dir await asyncio.sleep(0.02) task.cancel() @@ -612,9 +615,9 @@ async def test_daemon_client_stream_heartbeat_prevents_idle_timeout() -> None: server, task, runtime_dir = await _start_server( Path(root) / "runtime", service=_SlowFirstChunkService(), - stream_heartbeat_s=0.01, + stream_heartbeat_s=0.03, ) - client = DaemonClient(runtime_dir / "leapd.sock", timeout_s=0.03) + client = DaemonClient(runtime_dir / "leapd.sock", timeout_s=0.1) try: events = [event async for event in client.engine_chat("world")] @@ -1237,8 +1240,9 @@ async def shutdown(self) -> None: on_shutdown=shutdown_event.set, ) task = asyncio.create_task(server.serve_forever()) + readiness = get_transport().readiness_path(runtime_dir) for _ in range(50): - if (runtime_dir / "leapd.sock").exists(): + if readiness.exists(): break await asyncio.sleep(0.02) else: @@ -1665,7 +1669,7 @@ async def test_runtime_service_hot_reloads_config_before_daemon_chat( assert status["runtime_dir"] == str(settings.runtime_dir) assert status["llm_context_length"] == 700_000 assert status["context_used"] == 0 - assert status["runtime_source"].endswith("leapflow/__init__.py") + assert status["runtime_source"] == str(leapflow.__file__) assert status["runtime_executable"] assert status["runtime_version"] finally: diff --git a/tests/test_process_group.py b/tests/test_process_group.py index 0def35c..fb8611e 100644 --- a/tests/test_process_group.py +++ b/tests/test_process_group.py @@ -70,6 +70,7 @@ def test_second_terminate_is_false() -> None: group = ProcessGroup() group.attach(proc.pid) assert group.terminate() is True + proc.wait(timeout=5) # reap on POSIX: a zombie keeps the group alive assert group.terminate() is False # job handle released / pgid already dead From fa6bfbb12c48cc34c9d503368ecb41727b29c814 Mon Sep 17 00:00:00 2001 From: fanqiNO1 <1848839264@qq.com> Date: Fri, 7 Aug 2026 13:15:21 +0800 Subject: [PATCH 2/4] fix unit tests --- src/leapflow/cli/commands/slash_handlers.py | 4268 +++++++++-------- src/leapflow/daemon/client.py | 7 +- src/leapflow/daemon/server.py | 2 +- src/leapflow/daemon/session_registry.py | 17 +- src/leapflow/dashboard/intent.py | 85 +- src/leapflow/tools/shell_tools.py | 18 +- src/leapflow/utils/shell_lex.py | 18 + tests/_harness/cassette.py | 3 + tests/_harness/leapd.py | 5 +- tests/conftest.py | 25 + tests/journeys/test_r6_lifecycle.py | 4 +- tests/regression/test_test_layer_contracts.py | 2 +- tests/test_cli_entrypoint.py | 3 +- tests/test_code_tools.py | 973 ++-- tests/test_daemon_event_loop_blocking.py | 710 +-- tests/test_daemon_rpc.py | 57 +- tests/test_dev_terminal_tools.py | 20 +- tests/test_tool_call_hardening.py | 728 +-- tools/impact.py | 6 +- 19 files changed, 3536 insertions(+), 3415 deletions(-) create mode 100644 src/leapflow/utils/shell_lex.py diff --git a/src/leapflow/cli/commands/slash_handlers.py b/src/leapflow/cli/commands/slash_handlers.py index cca129e..e2724b0 100644 --- a/src/leapflow/cli/commands/slash_handlers.py +++ b/src/leapflow/cli/commands/slash_handlers.py @@ -1,2131 +1,2137 @@ -"""Slash-command handler implementations. - -Each handler follows the signature ``(ctx, console, args) -> None``. -All display logic uses ``LeapConsole`` for consistent theming. -""" - -from __future__ import annotations - -import logging -import os -import shlex -from pathlib import Path -from typing import TYPE_CHECKING, Any - -if TYPE_CHECKING: - from leapflow.cli.context import Context - from leapflow.cli.tui_app.console import LeapConsole - - -logger = logging.getLogger(__name__) - - -def build_tool_payload(ctx: "Context") -> dict[str, Any]: - """Build a serializable tool summary for local or daemon rendering.""" - from leapflow.cli.banner import _categorize_tools - from leapflow.tools.registry_bootstrap import TOOL_DEFINITIONS - - tool_groups = _categorize_tools(TOOL_DEFINITIONS) - groups = {category: sorted(names) for category, names in tool_groups.items()} - mcp_count = 0 - if hasattr(ctx.rpc, "connected") and ctx.rpc.connected: - mcp_count = len(getattr(ctx, "platform_tools", [])) - return { - "ok": True, - "groups": groups, - "total": sum(len(names) for names in groups.values()), - "mcp_count": mcp_count, - } - - -def render_tool_payload(console: "LeapConsole", payload: dict[str, Any]) -> None: - """Render a serializable tool summary.""" - from rich.table import Table - - if not payload.get("ok", True): - console.warning(str(payload.get("error") or "Tools are not available.")) - return - - groups = dict(payload.get("groups") or {}) - table = Table( - title="Available Tools", - show_header=True, - header_style="bold", - border_style="bright_black", - title_style="bold cyan", - padding=(0, 1), - ) - table.add_column("Category", style="cyan", no_wrap=True) - table.add_column("Tools") - - for category, names in groups.items(): - table.add_row(str(category), ", ".join(sorted(str(name) for name in names))) - - mcp_count = int(payload.get("mcp_count") or 0) - if mcp_count > 0: - table.add_row("mcp", f"{mcp_count} platform tools") - - console.print(table) - console.system(f"{int(payload.get('total') or 0)} tools available") - - -def build_usage_payload(ctx: "Context") -> dict[str, Any]: - """Build a serializable token usage summary.""" - engine = ctx.engine - if engine is None: - return {"ok": False, "error": "No active engine — send a message first."} - - tracker = getattr(engine, "usage_tracker", None) - if tracker is None: - return {"ok": False, "error": "Usage tracking not available."} - - summary = tracker.summary() - cap_registry = getattr(engine, "model_capabilities", None) - context_length = 0 - if cap_registry is not None: - caps = cap_registry.resolve(ctx.settings.llm_model) - context_length = int(caps.context_length) - - return { - "ok": True, - "model": ctx.settings.llm_model, - "prompt_tokens": int(summary.prompt_tokens), - "completion_tokens": int(summary.completion_tokens), - "total_tokens": int(summary.total_tokens), - "turn_count": int(getattr(engine, "turn_count", 0)), - "context_used": int(getattr(engine, "context_token_count", 0)), - "context_length": context_length, - } - - -def render_usage_payload(console: "LeapConsole", payload: dict[str, Any]) -> None: - """Render a serializable token usage summary.""" - from leapflow.cli.tui_app.status import _compact_tokens - - if not payload.get("ok", True): - console.warning(str(payload.get("error") or "Usage tracking not available.")) - return - - console.print() - prompt_tokens = int(payload.get("prompt_tokens") or 0) - completion_tokens = int(payload.get("completion_tokens") or 0) - total_tokens = int(payload.get("total_tokens") or 0) - turn_count = int(payload.get("turn_count") or 0) - context_used = int(payload.get("context_used") or 0) - context_length = int(payload.get("context_length") or 0) - lines = [ - f" Model: {payload.get('model') or ''}", - f" Input tokens: {_compact_tokens(prompt_tokens):>8} ({prompt_tokens:,})", - f" Output tokens: {_compact_tokens(completion_tokens):>8} ({completion_tokens:,})", - f" Total tokens: {_compact_tokens(total_tokens):>8} ({total_tokens:,})", - f" Turns: {turn_count}", - ] - if context_length > 0: - pct = int(context_used * 100 / context_length) - lines.append( - f" Context: {_compact_tokens(context_used)}/{_compact_tokens(context_length)} ({pct}%)" - ) - for line in lines: - console.system(line) - console.print() - - -def build_model_payload(ctx: "Context", args: str = "") -> dict[str, Any]: - """Build a serializable model summary or update the active model.""" - model_arg = args.strip() - if model_arg: - payload = build_config_payload(ctx, f"llm set --model {shlex.quote(model_arg)}") - payload["view"] = "model" - payload["requested_model"] = model_arg - return payload - - engine = ctx.engine - context_length = 0 - if engine is not None: - cap_registry = getattr(engine, "model_capabilities", None) - if cap_registry is not None: - caps = cap_registry.resolve(ctx.settings.llm_model) - context_length = int(caps.context_length) - return { - "ok": True, - "view": "model", - "model": ctx.settings.llm_model, - "context_length": context_length, - "requested_model": "", - } - - -def render_model_payload(console: "LeapConsole", payload: dict[str, Any]) -> None: - """Render a serializable model summary.""" - if not payload.get("ok", True): - console.warning(str(payload.get("error") or payload.get("message") or "Model information is not available.")) - return - requested = str(payload.get("requested_model") or "") - model = str(payload.get("model") or "") - if requested: - console.success(f"Model updated: {model}") - if payload.get("reloaded"): - console.system("Configuration reloaded for this session.") - return - - console.system(f"Current model: {model}") - context_length = int(payload.get("context_length") or 0) - if context_length > 0: - console.system(f"Context length: {context_length:,}") - - -def build_config_payload(ctx: "Context", args: str = "") -> dict[str, Any]: - """Execute a config command and return a serializable payload.""" - from leapflow.config_service import ConfigService - - service = ConfigService(ctx.settings) - try: - tokens = shlex.split(args) - except ValueError as exc: - return {"ok": False, "message": f"Invalid /config syntax: {exc}"} - if not tokens: - tokens = ["show"] - action = tokens[0] - try: - if action == "show": - if len(tokens) > 1: - return { - "ok": True, - "view": "config", - "mode": "show_detail", - "field": _config_field_to_dict(service.describe(tokens[1])), - } - return _config_snapshot_payload(service) - if action == "keys": - return {"ok": True, "view": "config", "mode": "keys", "sources": list(service.writable_keys())} - if action == "list": - category = tokens[1] if len(tokens) > 1 and not tokens[1].startswith("--") else None - return {"ok": True, "view": "config", "mode": "list", "fields": [_config_field_to_dict(item) for item in service.list_fields(category)]} - if action == "sources": - return {"ok": True, "view": "config", "mode": "sources", "sources": list(service.sources())} - if action == "get" and len(tokens) == 2: - value = service.get(tokens[1]) - return {"ok": True, "view": "config", "mode": "get", "values": [_config_value_to_dict(value)]} - if action == "set" and len(tokens) >= 3: - scope = _option_value(tokens[3:], "--scope", "profile") - result = service.set(tokens[1], tokens[2], scope=scope) # type: ignore[arg-type] - return _config_mutation_payload(ctx, service, result) - if action == "unset" and len(tokens) >= 2: - scope = _option_value(tokens[2:], "--scope", "profile") - result = service.unset(tokens[1], scope=scope) # type: ignore[arg-type] - return _config_mutation_payload(ctx, service, result) - if action == "llm": - return _build_llm_config_payload(ctx, service, tokens[1:]) - if action == "secret": - return _build_secret_config_payload(ctx, service, tokens[1:]) - except (KeyError, ValueError, RuntimeError) as exc: - return {"ok": False, "message": f"Config error: {exc}"} - return {"ok": False, "message": "Usage: /config [show|list|keys|sources|get|set|unset|llm|secret] ..."} - - -_CONFIG_LIST_COMPACT_WIDTH = 92 -_CONFIG_LIST_FULL_WIDTH = 118 - - -def _display_width(console: object, default: int = 100) -> int: - """Return the active Rich console width with a safe fallback for tests.""" - width = getattr(console, "width", None) - if isinstance(width, int) and width > 0: - return width - raw = getattr(console, "raw", None) - raw_width = getattr(raw, "width", None) - if isinstance(raw_width, int) and raw_width > 0: - return raw_width - return default - - -def _shorten(text: object, limit: int) -> str: - """Shorten a cell value before Rich allocates table columns.""" - value = str(text) if text is not None else "" - if limit <= 0: - return "" - if len(value) <= limit: - return value - return value[: max(1, limit - 1)] + "…" - - -def _config_scope_text(item: dict[str, Any], *, limit: int) -> str: - scopes = ",".join(str(scope) for scope in item.get("scopes") or []) - return _shorten(scopes, limit) - - -def _config_reload_text(item: dict[str, Any]) -> str: - value = item.get("hot_reload") - if isinstance(value, bool): - return "yes" if value else "no" - return str(value) if value is not None and str(value) else "no" - - -def _config_meta_text(item: dict[str, Any], *, scope_limit: int) -> str: - parts = [str(item.get("type") or "?")] - scope = _config_scope_text(item, limit=scope_limit) - if scope: - parts.append(scope) - parts.append(f"reload:{_config_reload_text(item)}") - return " · ".join(parts) - - -def _build_config_list_table(console: object, fields: list[dict[str, Any]]) -> object: - """Build a width-aware config list table for TUI rendering.""" - from rich.table import Table - - width = _display_width(console) - table = Table(title="Writable config fields", show_lines=False, expand=True) - if width < _CONFIG_LIST_COMPACT_WIDTH: - table.add_column("Key", no_wrap=True, overflow="ellipsis", max_width=32, ratio=2) - table.add_column("Value", overflow="fold", max_width=24, ratio=1) - table.add_column("Meta", overflow="fold", max_width=24, ratio=1) - for item in fields: - table.add_row( - _shorten(item.get("key"), 36), - _shorten(item.get("value"), 32), - _config_meta_text(item, scope_limit=14), - ) - return table - - if width < _CONFIG_LIST_FULL_WIDTH: - table.add_column("Key", no_wrap=True, overflow="ellipsis", max_width=34, ratio=2) - table.add_column("Value", overflow="fold", max_width=30, ratio=1) - table.add_column("Type", no_wrap=True, max_width=8) - table.add_column("Scope", no_wrap=True, overflow="ellipsis", max_width=18) - table.add_column("Reload", no_wrap=True, max_width=6) - for item in fields: - table.add_row( - _shorten(item.get("key"), 38), - _shorten(item.get("value"), 40), - str(item.get("type") or ""), - _config_scope_text(item, limit=18), - _config_reload_text(item), - ) - return table - - table.add_column("Key", no_wrap=True, overflow="ellipsis", max_width=36, ratio=2) - table.add_column("Value", overflow="fold", max_width=36, ratio=1) - table.add_column("Type", no_wrap=True, max_width=10) - table.add_column("Scope", no_wrap=True, overflow="ellipsis", max_width=22) - table.add_column("Reload", no_wrap=True, max_width=6) - table.add_column("Description", overflow="fold", ratio=2) - for item in fields: - table.add_row( - _shorten(item.get("key"), 42), - _shorten(item.get("value"), 48), - str(item.get("type") or ""), - _config_scope_text(item, limit=22), - _config_reload_text(item), - str(item.get("description") or ""), - ) - return table - - -def render_config_payload(console: "LeapConsole", payload: dict[str, Any]) -> None: - """Render config command output.""" - if not payload.get("ok", True): - console.warning(str(payload.get("message") or "Config command failed.")) - return - mode = str(payload.get("mode") or "show") - if mode in {"sources", "keys"}: - label = "Writable config keys" if mode == "keys" else "Config sources" - sources = [str(source) for source in payload.get("sources") or []] - if not sources: - console.system(f"No {label.lower()} found.") - return - console.system(f"{label}:") - for source in sources: - console.system(f" {source}") - return - if mode == "list": - fields = [dict(item) for item in payload.get("fields") or []] - if not fields: - console.system("No writable config fields found.") - return - console.print(_build_config_list_table(console, fields)) - console.system("Use /config show , /config get , /config set , or /config keys for compact output.") - return - if mode == "show_detail": - field = dict(payload.get("field") or {}) - if not field: - console.warning("No config field details found.") - return - from rich.panel import Panel - from rich.text import Text - - details = Text() - details.append(f"value: {field.get('value') or ''}\n") - details.append(f"type: {field.get('type') or ''}\n") - details.append(f"category: {field.get('category') or ''}\n") - details.append(f"scope: {','.join(str(scope) for scope in field.get('scopes') or [])}\n") - details.append(f"reload: {field.get('hot_reload') or ''}\n") - details.append(f"secret: {'true' if field.get('secret') else 'false'}\n") - value_hint = str(field.get("value_hint") or "") - if value_hint: - details.append(f"value hint: {value_hint}\n") - details.append(f"description: {field.get('description') or ''}") - examples = [str(example) for example in field.get("examples") or []] - if examples: - details.append(f"\nexample: {examples[0]}") - console.print(Panel(details, title=str(field.get("key") or "Config field"), border_style="cyan")) - return - if mode == "mutation": - console.success(str(payload.get("message") or "Config updated.")) - for key in payload.get("changed_keys") or []: - console.system(f" {key}") - warnings = payload.get("warnings") or [] - for warning in warnings: - console.warning(str(warning)) - if payload.get("reloaded"): - console.system("Configuration reloaded for this session.") - return - values = payload.get("values") or [] - if values: - for item in values: - console.system(f"{item.get('key')}={item.get('value')}") - warnings = payload.get("warnings") or [] - for warning in warnings: - console.warning(str(warning)) - - -def _config_snapshot_payload(service: Any) -> dict[str, Any]: - snapshot = service.snapshot() - return { - "ok": True, - "view": "config", - "mode": "show", - "values": [_config_value_to_dict(value) for value in snapshot.values], - "sources": list(snapshot.sources), - "warnings": list(snapshot.warnings), - } - - -def _config_value_to_dict(value: Any) -> dict[str, Any]: - return {"key": value.key, "value": value.value, "source": value.source, "secret": value.secret} - - -def _config_field_to_dict(value: Any) -> dict[str, Any]: - return { - "key": value.key, - "value": value.value, - "type": value.value_type, - "category": value.category, - "scopes": list(value.scopes), - "hot_reload": value.hot_reload, - "secret": value.secret, - "description": value.description, - "value_hint": value.value_hint, - "examples": list(value.examples), - } - - -def _build_llm_config_payload(ctx: "Context", service: Any, tokens: list[str]) -> dict[str, Any]: - action = tokens[0] if tokens else "show" - if action == "show": - payload = _config_snapshot_payload(service) - payload["values"] = [item for item in payload["values"] if str(item.get("key") or "").startswith("llm.")] - return payload - if action != "set": - return {"ok": False, "message": "Usage: /config llm [show|set --model NAME --base-url URL --api-key KEY]"} - options = tokens[1:] - if "--ask-api-key" in options: - return {"ok": False, "message": "Use `leap config llm key` in a terminal for secure API key prompts."} - result = service.configure_llm( - api_key=_option_value(options, "--api-key"), - base_url=_option_value(options, "--base-url"), - model=_option_value(options, "--model"), - context_length=_option_int(options, "--context-length"), - max_retries=_option_int(options, "--max-retries"), - scope=_option_value(options, "--scope", "profile"), - ) - return _config_mutation_payload(ctx, service, result) - - -def _build_secret_config_payload(ctx: "Context", service: Any, tokens: list[str]) -> dict[str, Any]: - action = tokens[0] if tokens else "list" - if action == "list": - return {"ok": True, "view": "config", "mode": "sources", "sources": list(service.list_secrets())} - if action == "set" and len(tokens) >= 3: - scope = _option_value(tokens[3:], "--scope", "profile") - result = service.set_secret(tokens[1], tokens[2], scope=scope) # type: ignore[arg-type] - return _config_mutation_payload(ctx, service, result) - if action == "get" and len(tokens) >= 2: - scope = _option_value(tokens[2:], "--scope", "profile") - reveal = "--reveal" in tokens[2:] - value = service.get_secret(tokens[1], scope=scope, reveal=reveal) # type: ignore[arg-type] - return {"ok": True, "view": "config", "mode": "get", "values": [{"key": tokens[1], "value": value, "secret": not reveal}]} - if action == "delete" and len(tokens) >= 2: - scope = _option_value(tokens[2:], "--scope", "profile") - result = service.delete_secret(tokens[1], scope=scope) # type: ignore[arg-type] - return _config_mutation_payload(ctx, service, result) - return {"ok": False, "message": "Usage: /config secret [list|set|get|delete] ..."} - - -def _config_mutation_payload(ctx: "Context", service: Any, result: Any) -> dict[str, Any]: - reloaded = False - if result.changed_keys: - reloaded = bool(ctx.reload_runtime_config_if_changed(force=True)) - return { - "ok": bool(result.ok), - "view": "config", - "mode": "mutation", - "message": result.message, - "changed_keys": list(result.changed_keys), - "warnings": list(getattr(result, "warnings", ()) or ()), - "reloaded": reloaded, - # Runtime values the status bar renders. A daemon-mode TUI is a separate - # process and cannot see the reload, so it needs them echoed back here; - # context length travels with the model because switching models usually - # changes it too. - "model": ctx.settings.llm_model, - "llm_context_length": int(getattr(ctx.settings, "llm_context_length", 0) or 0), - } - - -def _option_value(tokens: list[str], option: str, default: str | None = None) -> str | None: - if option not in tokens: - return default - index = tokens.index(option) - if index + 1 >= len(tokens): - raise ValueError(f"Missing value for {option}") - return tokens[index + 1] - - -def _option_int(tokens: list[str], option: str) -> int | None: - value = _option_value(tokens, option) - return int(value) if value is not None else None - - -def handle_status(ctx: "Context", console: "LeapConsole", args: str) -> None: - """Display session status: model, context, platform, session info.""" - from rich.panel import Panel - from rich.text import Text - - info = Text() - - info.append("Model: ", style="dim") - info.append(f"{ctx.settings.llm_model}\n", style="bold") - - engine = ctx.engine - if engine is not None: - cap_registry = getattr(engine, "model_capabilities", None) - ctx_len = 0 - if cap_registry is not None: - caps = cap_registry.resolve(ctx.settings.llm_model) - ctx_len = caps.context_length - ctx_used = getattr(engine, "context_token_count", 0) - turn_count = getattr(engine, "turn_count", 0) - - if ctx_len: - pct = int(ctx_used * 100 / ctx_len) if ctx_len else 0 - info.append("Context: ", style="dim") - pct_style = "bold red" if pct >= 90 else ("yellow" if pct >= 75 else "") - info.append(f"{ctx_used:,} / {ctx_len:,} ({pct}%)\n", style=pct_style) - - info.append("Turns: ", style="dim") - info.append(f"{turn_count}\n") - - platform_status = "connected" if (hasattr(ctx.rpc, "connected") and ctx.rpc.connected) else "mock" - info.append("Platform: ", style="dim") - p_style = "green" if platform_status == "connected" else "dim" - info.append(f"{platform_status}\n", style=p_style) - - cwd = os.getcwd().replace(os.path.expanduser("~"), "~") - info.append("CWD: ", style="dim") - info.append(f"{cwd}\n") - - info.append("Profile: ", style="dim") - info.append(f"{ctx.settings.profile}\n") - - info.append("Config: ", style="dim") - info.append(f"{str(ctx.settings.profile_layout.config_dir).replace(os.path.expanduser('~'), '~')}\n") - - user_config = str(ctx.settings.layout.user_config_path).replace(os.path.expanduser("~"), "~") - info.append("User cfg: ", style="dim") - info.append(f"{user_config}\n") - - workspace_config = str(ctx.settings.workspace_root / ".leapflow" / "config.yaml") - info.append("Workspace: ", style="dim") - info.append(f"{workspace_config.replace(os.path.expanduser('~'), '~')}\n") - - session_id = getattr(ctx.session, "session_id", "") - if session_id: - info.append("Session: ", style="dim") - info.append(f"{session_id}\n") - - from leapflow.engine.session import SessionMode - mode = "idle" - if ctx.session: - if ctx.session.mode == SessionMode.LEARNING: - mode = "learning" - elif ctx.session.mode == SessionMode.EXECUTING: - mode = "executing" - info.append("Mode: ", style="dim") - info.append(f"{mode}\n") - - gw = getattr(ctx, "gateway_server", None) - if gw is not None: - statuses = gw.platform_status() - connected = [s for s in statuses if s.connected] - info.append("Gateway: ", style="dim") - if connected: - names = [] - for s in connected: - m = gw.manifests.get(s.platform_id) - names.append(m.display_name if m else s.platform_id) - info.append(f"{', '.join(names)}\n", style="green") - else: - info.append("no connections\n", style="dim") - - console.print(Panel( - info, - title="[bold cyan]LeapFlow Status[/]", - border_style="bright_black", - padding=(0, 2), - )) - - -def handle_tool(ctx: "Context", console: "LeapConsole", args: str) -> None: - """Display available tools grouped by category.""" - render_tool_payload(console, build_tool_payload(ctx)) - - -def handle_usage(ctx: "Context", console: "LeapConsole", args: str) -> None: - """Display token usage for the current session.""" - render_usage_payload(console, build_usage_payload(ctx)) - - -def handle_model(ctx: "Context", console: "LeapConsole", args: str) -> None: - """Show or switch the active model.""" - render_model_payload(console, build_model_payload(ctx, args)) - - -def handle_config(ctx: "Context", console: "LeapConsole", args: str) -> None: - """View or update runtime configuration.""" - render_config_payload(console, build_config_payload(ctx, args)) - - -def handle_gateway(ctx: "Context", console: "LeapConsole", args: str) -> None: - """Display gateway status: connected platforms, available integrations.""" - from rich.panel import Panel - from rich.text import Text - - gw = getattr(ctx, "gateway_server", None) - if gw is None: - console.warning("Gateway not initialised.") - return - - statuses = gw.platform_status() - if not statuses: - console.system("No platform manifests discovered.") - return - - info = Text() - connected = [s for s in statuses if s.connected] - configured = [s for s in statuses if not s.connected and s.error == "configured but not connected"] - available = [s for s in statuses if not s.connected and not s.error] - - import time - - if connected: - info.append("Connected\n", style="bold green") - for s in connected: - m = gw.manifests.get(s.platform_id) - name = m.display_name if m else s.platform_id - uptime = "" - if s.connected_since > 0: - secs = int(time.time() - s.connected_since) - if secs < 60: - uptime = f" ({secs}s)" - elif secs < 3600: - uptime = f" ({secs // 60}m)" - else: - uptime = f" ({secs // 3600}h {(secs % 3600) // 60}m)" - info.append(f" ● {name}{uptime}\n", style="green") - - if configured: - info.append("Configured (not connected)\n", style="bold yellow") - for s in configured: - m = gw.manifests.get(s.platform_id) - name = m.display_name if m else s.platform_id - info.append(f" ○ {name}\n", style="yellow") - - if available: - info.append("Available\n", style="bold dim") - names = [gw.manifests[s.platform_id].display_name for s in available if s.platform_id in gw.manifests] - info.append(f" {', '.join(names)}\n", style="dim") - - info.append("\n", style="dim") - info.append('Say "connect to " to set up a new integration.', style="dim italic") - - console.print(Panel( - info, - title="[bold cyan]Gateway[/]", - border_style="bright_black", - padding=(0, 2), - )) - - -def _app_usage() -> dict[str, Any]: - return { - "ok": False, - "error": "Usage: /app [platform] | /app status [platform] | /app connect [--option value] | /app disconnect | /app remove | /app events [status|start|stop] | /app actions ", - "next_actions": ("/app", "/app ", "/app status "), - } - - -def _parse_app_options(tokens: list[str]) -> tuple[dict[str, str], str]: - options: dict[str, str] = {} - index = 0 - while index < len(tokens): - token = tokens[index] - if not token.startswith("--") or token == "--": - return {}, f"Unexpected argument: {token}" - key_value = token[2:] - if "=" in key_value: - key, value = key_value.split("=", 1) - if not key: - return {}, f"Invalid option: {token}" - options[key] = value - index += 1 - continue - if index + 1 >= len(tokens): - return {}, f"Missing value for option: {token}" - key = key_value - if not key: - return {}, f"Invalid option: {token}" - options[key] = tokens[index + 1] - index += 2 - return options, "" - - -def _parse_app_params(args: str) -> dict[str, Any]: - try: - tokens = shlex.split(args) - except ValueError as exc: - return {"ok": False, "error": f"Invalid /app arguments: {exc}"} - - if not tokens or tokens[0].lower() == "list": - if len(tokens) > 1: - return _app_usage() - return {"ok": True, "params": {"action": "list"}, "view": "list"} - - head = tokens[0].lower() - if head == "status": - if len(tokens) > 2: - return _app_usage() - params: dict[str, Any] = {"action": "status"} - if len(tokens) == 2: - params["platform"] = tokens[1].lower() - return {"ok": True, "params": params, "view": "status"} - - if head == "connect": - if len(tokens) < 2: - return _app_usage() - options, error = _parse_app_options(tokens[2:]) - if error: - return { - "ok": False, - "error": error, - "next_actions": ("/app connect --