diff --git a/dashscope/acli/__init__.py b/dashscope/acli/__init__.py index 9a60c83..4f5a0ac 100644 --- a/dashscope/acli/__init__.py +++ b/dashscope/acli/__init__.py @@ -1,7 +1,7 @@ # -*- coding: utf-8 -*- from __future__ import annotations -__version__ = "0.6.0" +__version__ = "0.6.2" # Expose the lightweight programmatic SDK at the package root. try: diff --git a/dashscope/acli/agent.py b/dashscope/acli/agent.py index 68d2ee3..b76364a 100644 --- a/dashscope/acli/agent.py +++ b/dashscope/acli/agent.py @@ -235,9 +235,32 @@ def reset(self): self.messages = [] def load_session(self) -> int: - """Restore self.messages from session_path. Returns the number of - messages loaded (0 if no file, file empty, or parse failed).""" - if not self.session_path or not self.session_path.exists(): + """Restore self.messages. Returns the number of messages loaded + (0 if nothing could be restored). + + The per-topic event stream is the source of truth: the latest + ``messages/snapshot`` wins when present (crash recovery via + append-only replay; torn trailing lines are skipped). The + ``history.json`` file remains as the compat fallback for + sessions written before snapshots existed. + """ + if not self.session_path: + return 0 + try: + from dashscope.acli.session_events import ( + EVENTS_FILENAME, + SessionEventLog, + latest_snapshot_messages, + ) + + log = SessionEventLog(self.session_path.parent / EVENTS_FILENAME) + resumed = latest_snapshot_messages(log.read_raw()) + if resumed: + self.messages = resumed + return len(resumed) + except Exception: + pass + if not self.session_path.exists(): return 0 try: data = json.loads(self.session_path.read_text()) @@ -326,11 +349,25 @@ def _system_prompt_for_turn(self, user_input_text: str) -> str: experience_tracker=self.experience_tracker, disabled_caps_provider=self.disabled_caps_provider, directives_provider=self.directives_provider, + scene_provider=self._scene_section, current_turn_tools=self._current_turn_tools, connected_mcp_services=self._connected_mcp_services, ) return self._prompt_pipeline.render(ctx) + def _scene_section(self) -> str: + """Scene memory of the current session topic (best-effort). + + Subagents and SDK callers may run without a session manager; + any failure simply yields no scene section. + """ + try: + from dashscope.acli.session import get_session_manager + + return get_session_manager().get_scene() + except Exception: + return "" + def _reflection_section(self) -> str: """Inject reflection hints when repeated failures detected.""" tracker = self.memory_manager.session.reflection @@ -524,9 +561,9 @@ async def _run_stream_body(self, user_input) -> AsyncIterator[str]: async for chunk in self.provider.chat_stream( normalize_for_model(messages_with_system, self.model_name), tools_schema, - response_format={"type": "json_object"} - if self.json_mode - else None, + response_format=( + {"type": "json_object"} if self.json_mode else None + ), ): if chunk.delta_content: full_content += chunk.delta_content @@ -765,6 +802,39 @@ async def _run_stream_body(self, user_input) -> AsyncIterator[str]: ), ) + # Record the completed turn as events (session-as-event-log + # direction) BEFORE persisting history.json: the event stream is + # the source of truth, so it must never be older than the + # fallback store if we crash between the two writes. Best-effort: + # subagents/SDK callers may run without a session manager, and + # event recording must never break the loop. + try: + from dashscope.acli.session import get_session_manager + + turn_topic = ( + self.session_path.parent.name if self.session_path else None + ) + get_session_manager().record_turn_event( + user_text=text_of(user_input), + assistant_text=last_content, + tools_used=self._current_turn_tools, + outcome=( + _classify_outcome( + self._turn_tool_successes, + self._turn_tool_failures, + ) + if self._current_turn_tools + else "" + ), + topic=turn_topic, + ) + get_session_manager().record_messages_snapshot( + self.messages, + topic=turn_topic, + ) + except Exception: + pass + # Persist session before memory write so a memory exception can't # cost us the conversation. self.save_session() diff --git a/dashscope/acli/cli/__init__.py b/dashscope/acli/cli/__init__.py index d65b288..d2ace54 100644 --- a/dashscope/acli/cli/__init__.py +++ b/dashscope/acli/cli/__init__.py @@ -52,9 +52,7 @@ _handle_slash_command, dispatch_async_command, ) -from dashscope.acli.cli.examples import ( # noqa: E402 - _handle_example_command, -) +from dashscope.acli.cli.examples import _handle_example_command # noqa: E402 from dashscope.acli.cli.handlers_capability import ( # noqa: F401,E402 _cap_enabled, sync_extensions_into_catalog, @@ -62,9 +60,7 @@ from dashscope.acli.cli.handlers_misc import ( # noqa: F401,E402 _handle_report_command, ) -from dashscope.acli.cli.handlers_setup import ( # noqa: F401,E402 - _handle_setup, -) +from dashscope.acli.cli.handlers_setup import _handle_setup # noqa: F401,E402 # Import MCP management from submodule from dashscope.acli.cli.mcp import ( # noqa: F401,E402 diff --git a/dashscope/acli/cli/constants.py b/dashscope/acli/cli/constants.py index 5ba5b7a..ba8b370 100644 --- a/dashscope/acli/cli/constants.py +++ b/dashscope/acli/cli/constants.py @@ -186,6 +186,7 @@ "/report", "/feedback", "/history", + "/undo", "/json", "/save", "/privacy", @@ -228,7 +229,7 @@ "/mcp": ["list", "add", "remove"], "/cron": ["add", "list", "remove", "pause", "resume"], "/feedback": ["good", "bad"], - "/history": ["stats", "list", "export", "clear"], + "/history": ["stats", "list", "search", "export", "clear"], "/json": ["on", "off"], "/privacy": ["on", "off", "status"], "/audit": ["recent", "query", "clear"], diff --git a/dashscope/acli/cli/dispatch.py b/dashscope/acli/cli/dispatch.py index 9f45c6a..d38c525 100644 --- a/dashscope/acli/cli/dispatch.py +++ b/dashscope/acli/cli/dispatch.py @@ -403,6 +403,11 @@ def _handle_slash_command( elif cmd.startswith("/history"): _handle_history_command(cmd) return True + elif cmd == "/undo": + from dashscope.acli.tools.checkpoint import handle_undo_command + + handle_undo_command() + return True elif cmd.startswith("/privacy"): _handle_privacy_command(cmd, config) return True diff --git a/dashscope/acli/cli/handlers_misc.py b/dashscope/acli/cli/handlers_misc.py index b54e62e..94ffe93 100644 --- a/dashscope/acli/cli/handlers_misc.py +++ b/dashscope/acli/cli/handlers_misc.py @@ -1,10 +1,14 @@ # -*- coding: utf-8 -*- """Miscellaneous command handlers (trust, history, report).""" # pylint: disable=protected-access,too-many-branches,too-many-statements +# pylint: disable=too-many-return-statements from __future__ import annotations +from typing import Any + from rich.console import Console +from rich.markup import escape from dashscope.acli.agent import Agent @@ -77,6 +81,86 @@ def _handle_trust_command(cmd: str, agent: Agent) -> None: ) +def _message_text(msg: dict[str, Any]) -> str: + """Flatten a chat message's content into one-line plain text.""" + content = msg.get("content", "") + if isinstance(content, list): + content = " ".join( + part.get("text", "") for part in content if isinstance(part, dict) + ) + if not isinstance(content, str): + return "" + return " ".join(content.split()) + + +def _search_snippet(text: str, idx: int, kw_len: int) -> str: + """Build a short one-line snippet centered on a match position.""" + start = max(0, idx - 20) + end = min(len(text), idx + kw_len + 40) + prefix = "..." if start > 0 else "" + suffix = "..." if end < len(text) else "" + return prefix + text[start:end] + suffix + + +def _highlight_keyword(text: str, keyword: str) -> str: + """Wrap keyword occurrences in rich markup (case-insensitive).""" + lower_kw = keyword.lower() + if not lower_kw: + return escape(text) + lower_text = text.lower() + out: list[str] = [] + pos = 0 + while True: + idx = lower_text.find(lower_kw, pos) + if idx < 0: + out.append(escape(text[pos:])) + break + out.append(escape(text[pos:idx])) + out.append("[bold yellow]") + out.append(escape(text[idx : idx + len(keyword)])) + out.append("[/bold yellow]") + pos = idx + len(keyword) + return "".join(out) + + +def _history_search_matches( + keyword: str, + limit: int = 20, +) -> list[dict[str, str]]: + """Case-insensitive substring search across all session history. + + Scans every stored session's messages and returns up to ``limit`` + matches, each carrying the session topic, a timestamp, the message + role, and a one-line snippet with match context. + """ + from dashscope.acli.session import get_session_manager + + needle = keyword.lower() + if not needle or limit <= 0: + return [] + mgr = get_session_manager() + matches: list[dict[str, str]] = [] + for meta in mgr.list_topics(): + if len(matches) >= limit: + break + for msg in mgr.load_messages(meta.topic): + text = _message_text(msg) + idx = text.lower().find(needle) if text else -1 + if idx < 0: + continue + matches.append( + { + "session": meta.topic, + "timestamp": meta.last_accessed or "", + "role": str(msg.get("role", "?")), + "snippet": _search_snippet(text, idx, len(needle)), + }, + ) + if len(matches) >= limit: + break + return matches + + def _handle_history_command(cmd: str) -> None: """Manage conversation history.""" from dashscope.acli.platforms.local.history import ( @@ -107,6 +191,8 @@ def _handle_history_command(cmd: str) -> None: "\n[dim]Usage:\n" " /history stats — show stats\n" " /history list [n] — list recent n\n" + " /history search [limit] — full-text " + "search\n" " /history export [--format json|md|html] — export\n" " /history clear — clear " "history[/dim]", @@ -147,6 +233,41 @@ def _handle_history_command(cmd: str) -> None: ) return + if sub == "search": + if len(parts) < 3: + console.print( + "[dim]Usage: /history search [limit][/dim]", + ) + return + keyword = parts[2] + limit = 20 + if len(parts) >= 4: + try: + limit = int(parts[3]) + except ValueError: + limit = 0 + if limit <= 0: + console.print( + "[red]limit must be a positive integer[/red]", + ) + return + matches = _history_search_matches(keyword, limit=limit) + if not matches: + console.print(f"[dim]No matches for '{keyword}'[/dim]") + return + header = f"[bold]{len(matches)} match(es) for '{keyword}':[/bold]" + console.print(header) + for i, m in enumerate(matches, 1): + ts = m["timestamp"][:16] + head = ( + f" {i}. [cyan]{m['session']}[/cyan] " + f"[dim]{ts}[/dim] [bold]{m['role']}[/bold]" + ) + console.print(head) + snippet = _highlight_keyword(m["snippet"], keyword) + console.print(f" {snippet}") + return + if sub == "export" and len(parts) >= 3: output_path = parts[2] fmt = "html" @@ -174,7 +295,8 @@ def _handle_history_command(cmd: str) -> None: console.print(f"[green]✓ Cleared {count} history records[/green]") return - console.print("[dim]Usage: /history [stats|list|export|clear][/dim]") + usage = "[dim]Usage: /history [stats|list|search|export|clear][/dim]" + console.print(usage) def _handle_report_command(agent: Agent) -> None: diff --git a/dashscope/acli/cli/handlers_session.py b/dashscope/acli/cli/handlers_session.py index 76fcbfa..b4d0d7b 100644 --- a/dashscope/acli/cli/handlers_session.py +++ b/dashscope/acli/cli/handlers_session.py @@ -8,6 +8,21 @@ console = Console() +_SESSION_USAGE = ( + "\n[dim]Usage:\n" + " /session — show current topic\n" + " /session new [name] — new session (default: default)\n" + " /session list — list all sessions\n" + " /session switch — switch to a topic\n" + " /session rename — rename\n" + " /session remove — remove session " + "(default cannot be removed)\n" + " /session scene — show scene memory of the current topic\n" + " /session scene — append a note to scene memory\n" + " /session scene set — replace scene memory\n" + " /session scene clear — clear scene memory[/dim]" +) + def _handle_session_command(cmd: str, config, agent) -> None: """Handle /session commands for multi-topic session management.""" @@ -20,16 +35,7 @@ def _handle_session_command(cmd: str, config, agent) -> None: # Show current topic current = session_mgr.get_current_topic() console.print(f"[bold]Current session[/bold]: {current}") - console.print( - "\n[dim]Usage:\n" - " /session — show current topic\n" - " /session new [name] — new session (default: default)\n" - " /session list — list all sessions\n" - " /session switch — switch to a topic\n" - " /session rename — rename\n" - " /session remove — remove session " - "(default cannot be removed)[/dim]", - ) + console.print(_SESSION_USAGE) return subcmd = parts[1] @@ -137,14 +143,39 @@ def _handle_session_command(cmd: str, config, agent) -> None: else: console.print(f"[red]Session '{topic}' does not exist[/red]") + elif subcmd == "scene": + topic = session_mgr.get_current_topic() + rest = parts[2].strip() if len(parts) > 2 else "" + if not rest: + text = session_mgr.get_scene() + if text: + console.print( + f"[bold]Scene memory[/bold] [dim]({topic})[/dim]:", + ) + console.print(text) + else: + console.print( + f"[dim]No scene memory for '{topic}' yet. Add one " + f"with: /session scene [/dim]", + ) + elif rest == "clear": + session_mgr.set_scene("") + console.print(f"[green]Scene memory cleared ({topic})[/green]") + elif rest.startswith("set "): + text = rest[len("set ") :].strip() + if session_mgr.set_scene(text): + console.print( + f"[green]Scene memory replaced ({topic})[/green]", + ) + else: + console.print("[red]Failed to write scene memory[/red]") + else: + if session_mgr.append_scene(rest): + console.print( + f"[green]Scene note appended ({topic})[/green]", + ) + else: + console.print("[red]Failed to write scene memory[/red]") + else: - console.print( - "[dim]Usage:\n" - " /session — show current topic\n" - " /session new [name] — new session (default: default)\n" - " /session list — list all sessions\n" - " /session switch — switch to a topic\n" - " /session rename — rename\n" - " /session remove — remove session " - "(default cannot be removed)[/dim]", - ) + console.print(_SESSION_USAGE) diff --git a/dashscope/acli/cli/mcp.py b/dashscope/acli/cli/mcp.py index b570665..6343786 100644 --- a/dashscope/acli/cli/mcp.py +++ b/dashscope/acli/cli/mcp.py @@ -7,6 +7,7 @@ from rich.status import Status from dashscope.acli.config import Config, MCPServerConfig +from dashscope.acli.mcp_stdio import StdioMCPClient from dashscope.acli.platforms.bailian import MCPClient, MCPError from dashscope.acli.skills import list_known_services from dashscope.acli.tools.registry import registry @@ -14,22 +15,34 @@ console = Console() # Active MCP clients - shared state -_mcp_clients: dict[str, MCPClient] = {} +_mcp_clients: dict[str, MCPClient | StdioMCPClient] = {} -async def _connect_mcp(service: str, config: Config, url: str = "") -> str: +async def _connect_mcp( + service: str, + config: Config, + url: str = "", + server: MCPServerConfig | None = None, +) -> str: """Connect to an MCP service and register its tools. Returns empty string on success, error message on failure.""" if service in _mcp_clients: return "" + use_stdio = bool(server) and ( + (server.transport or "").lower() == "stdio" or bool(server.command) + ) try: - client = MCPClient( - service=service, - api_key=config.tongyi_api_key, - url=url, - ) + client: MCPClient | StdioMCPClient + if use_stdio: + client = StdioMCPClient(server.command, server.args) + else: + client = MCPClient( + service=service, + api_key=config.tongyi_api_key, + url=url, + ) except MCPError as e: return str(e) @@ -143,7 +156,12 @@ async def _handle_mcp_command(cmd: str, config: Config): async def _init_mcp_servers(config: Config): """Connect to configured MCP servers on startup.""" for mcp_cfg in config.mcp_servers: - error = await _connect_mcp(mcp_cfg.service, config, url=mcp_cfg.url) + error = await _connect_mcp( + mcp_cfg.service, + config, + url=mcp_cfg.url, + server=mcp_cfg, + ) if not error: client = _mcp_clients[mcp_cfg.service] summary = f"{len(client.tools)} tools" diff --git a/dashscope/acli/cli/startup.py b/dashscope/acli/cli/startup.py index 0d02aa1..760bf29 100644 --- a/dashscope/acli/cli/startup.py +++ b/dashscope/acli/cli/startup.py @@ -7,7 +7,7 @@ from dashscope.acli import __version__ from dashscope.acli.cli.constants import ALL_CAPABILITY_KEYS -from dashscope.acli.config import WORKSPACE_DIR, Config +from dashscope.acli.config import Config from dashscope.acli.tools.registry import registry from dashscope.acli.utils import mask_secret @@ -21,7 +21,8 @@ def _load_system_prompt() -> str | None: are discovered separately by Agent.__init__ and passed to the prompt pipeline for proper stable/ephemeral separation. """ - from dashscope.acli.config import CONFIG_DIR + # Imported at call time so tests can patch acli.config.WORKSPACE_DIR. + from dashscope.acli.config import CONFIG_DIR, WORKSPACE_DIR base: str | None = None for d in (WORKSPACE_DIR, CONFIG_DIR): @@ -50,7 +51,7 @@ def _load_references() -> str | None: Unlike skills (invoked on demand), references are knowledge docs that must always be in the system prompt — e.g. generated SDK API indexes. """ - from dashscope.acli.config import CONFIG_DIR + from dashscope.acli.config import CONFIG_DIR, WORKSPACE_DIR parts: dict[str, str] = {} for d in (CONFIG_DIR, WORKSPACE_DIR): @@ -82,7 +83,9 @@ def _compose_system_prompt(base: str | None) -> str | None: return f"{base}\n\n---\n\n{section}" if base else section -def _print_banner(config: Config | None = None): +def _print_banner(config: Config | None = None) -> None: + from dashscope.acli.config import WORKSPACE_DIR + logo = ( " _ _ _ ____ _ ___\n" " / \\ __ _ ___ _ __ | |_(_) ___ / ___| | |_ _|\n" diff --git a/dashscope/acli/commands.py b/dashscope/acli/commands.py index 0907661..111bd1d 100644 --- a/dashscope/acli/commands.py +++ b/dashscope/acli/commands.py @@ -38,7 +38,11 @@ "JSON output mode (replies forced to JSON when on)", ), ("/compress", "Compress context (LLM summary replaces history)"), - ("/history", "Conversation history (stats/list/export/clear)"), + ( + "/history", + "Conversation history (stats/list/search/export/clear)", + ), + ("/undo", "Undo the last file write/delete (checkpoint)"), ( "/feedback good|bad", "Rate task satisfaction (stored in experience memory)", @@ -106,7 +110,10 @@ [ ("/profile", "User profile (list/search/add/remove/clear)"), ("/memory", "Chat history (list/search/remove /clear)"), - ("/session", "Session management (new/list/switch/rename/remove)"), + ( + "/session", + "Session management (new/list/switch/rename/remove/scene)", + ), ( "/summarize", "Summarize the current task; record key steps and lessons", diff --git a/dashscope/acli/config.py b/dashscope/acli/config.py index 99f2fca..3ff97c1 100644 --- a/dashscope/acli/config.py +++ b/dashscope/acli/config.py @@ -165,6 +165,10 @@ def _add(m: str) -> None: class MCPServerConfig: service: str url: str = "" + # "sse" (remote, default) or "stdio" (local subprocess). + transport: str = "sse" + command: str = "" + args: list[str] = field(default_factory=list) @dataclass @@ -282,6 +286,9 @@ class Config: debug: bool = ( False # When True, log final LLM prompts to .acli/logs/llm.log ) + sandbox: bool = ( + False # When True, run shell commands in an OS sandbox if available + ) skill_registry: str = ( "" # Optional registry index URL/path for /skill search/install ) @@ -550,6 +557,9 @@ def _load_workspace_from(self, path: Path): if "debug" in data: val = str(data["debug"]).lower() self.debug = val not in ("false", "0", "no") + if "sandbox" in data: + val = str(data["sandbox"]).lower() + self.sandbox = val not in ("false", "0", "no") if "voice_silence_duration" in data: try: self.voice_silence_duration = float( @@ -659,7 +669,18 @@ def _load_workspace_from(self, path: Path): if "mcp_servers" in data: for mcp_data in data["mcp_servers"]: if isinstance(mcp_data, dict) and "service" in mcp_data: - self.mcp_servers.append(MCPServerConfig(**mcp_data)) + kwargs = { + k: v + for k, v in mcp_data.items() + if k + in ("service", "url", "transport", "command", "args") + } + raw_args = kwargs.get("args") + if isinstance(raw_args, list): + kwargs["args"] = [str(a) for a in raw_args] + else: + kwargs.pop("args", None) + self.mcp_servers.append(MCPServerConfig(**kwargs)) if "examples_repo" in data: self.examples_repo = str(data["examples_repo"]) if "examples_branch" in data: @@ -798,6 +819,8 @@ def _workspace_lines(self) -> list[str]: lines.append("privacy_mode = true") if self.debug: lines.append("debug = true") + if self.sandbox: + lines.append("sandbox = true") lines.append(f"tts_enabled = {str(self.tts_enabled).lower()}") if self.tts_model and self.tts_model != "cosyvoice-v2": lines.append(f"tts_model = {toml_str(self.tts_model)}") @@ -846,4 +869,11 @@ def _workspace_lines(self) -> list[str]: lines.append(f"service = {toml_str(mcp.service)}") if mcp.url: lines.append(f"url = {toml_str(mcp.url)}") + if mcp.transport and mcp.transport != "sse": + lines.append(f"transport = {toml_str(mcp.transport)}") + if mcp.command: + lines.append(f"command = {toml_str(mcp.command)}") + if mcp.args: + arg_list = ", ".join(toml_str(a) for a in mcp.args) + lines.append(f"args = [{arg_list}]") return lines diff --git a/dashscope/acli/eval/__init__.py b/dashscope/acli/eval/__init__.py index 4af79d5..9d22d53 100644 --- a/dashscope/acli/eval/__init__.py +++ b/dashscope/acli/eval/__init__.py @@ -214,11 +214,13 @@ async def compare( / max(len(results_a), 1), "avg_duration_b": sum(r.duration for r in results_b) / max(len(results_b), 1), - "winner": label_a - if avg_a > avg_b - else label_b - if avg_b > avg_a - else "tie", + "winner": ( + label_a + if avg_a > avg_b + else label_b + if avg_b > avg_a + else "tie" + ), "results_a": [r.to_dict() for r in results_a], "results_b": [r.to_dict() for r in results_b], } diff --git a/dashscope/acli/hooks.py b/dashscope/acli/hooks.py index bc35e1b..d0e58a5 100644 --- a/dashscope/acli/hooks.py +++ b/dashscope/acli/hooks.py @@ -99,9 +99,11 @@ def _build_variables(ctx: HookContext) -> dict[str, str]: "filename_stem": p.stem if p else "", "exit_code": "", "content": "", - "args": json.dumps(ctx.arguments, ensure_ascii=False) - if ctx.arguments - else "", + "args": ( + json.dumps(ctx.arguments, ensure_ascii=False) + if ctx.arguments + else "" + ), "result": (ctx.result or "")[:1000], "error": (ctx.result or "")[:1000] if ctx.success is False else "", } diff --git a/dashscope/acli/mcp_stdio.py b/dashscope/acli/mcp_stdio.py new file mode 100644 index 0000000..bd92e13 --- /dev/null +++ b/dashscope/acli/mcp_stdio.py @@ -0,0 +1,312 @@ +# -*- coding: utf-8 -*- +"""MCP stdio transport client. + +Speaks newline-delimited JSON-RPC 2.0 to a local MCP server subprocess +over stdin/stdout. Mirrors the SSE ``MCPClient`` interface (initialize / +list_tools / list_prompts / call_tool / close plus ``tools``, ``prompts`` +and ``last_error`` attributes) so ``cli/mcp.py`` can use either transport +interchangeably. +""" + +from __future__ import annotations + +import asyncio +import json + +from dashscope.acli.platforms.bailian.mcp import MCP_PROTOCOL_VERSION, MCPError + +__all__ = ["StdioMCPClient", "MCPError"] + +_CLIENT_INFO = {"name": "acli", "version": "0.1.0"} +_CLOSE_TIMEOUT = 3.0 +_STDERR_TAIL = 500 + + +class StdioMCPClient: + """MCP client that drives a local server over stdio. + + The server process is spawned with ``command`` + ``args``; requests + and responses are single-line JSON documents. A background reader + matches response ids to pending request futures; server notifications + (messages without an ``id``) are ignored. + """ + + def __init__( + self, + command: str, + args: list[str] | None = None, + timeout: float = 30.0, + ): + if not command: + raise MCPError("stdio MCP server requires a command") + self.command = command + self.args = list(args or []) + self.timeout = timeout + self.tools: list[dict] = [] + self.prompts: list[dict] = [] + self.last_error = "" + self._proc: asyncio.subprocess.Process | None = None + self._reader_task: asyncio.Task | None = None + self._pending: dict[int, asyncio.Future] = {} + self._request_id = 0 + self._stderr_tail = "" + + # ------------------------------------------------------------------ # + # Lifecycle + # ------------------------------------------------------------------ # + async def initialize(self) -> bool: + """Spawn the server, run the MCP initialize handshake.""" + try: + self._proc = await asyncio.create_subprocess_exec( + self.command, + *self.args, + stdin=asyncio.subprocess.PIPE, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + except (OSError, ValueError) as exc: + self.last_error = f"Failed to start MCP server: {exc}" + return False + + self._reader_task = asyncio.create_task(self._read_loop()) + + params = { + "protocolVersion": MCP_PROTOCOL_VERSION, + "capabilities": {}, + "clientInfo": dict(_CLIENT_INFO), + } + try: + resp = await self._request("initialize", params) + except MCPError as exc: + self.last_error = str(exc) + await self.close() + return False + if "error" in resp: + error = resp["error"] + message = error.get("message", str(error)) + self.last_error = f"MCP initialize failed: {message}" + await self.close() + return False + try: + await self._notify("notifications/initialized") + except MCPError as exc: + self.last_error = str(exc) + await self.close() + return False + return True + + async def close(self) -> None: + """Stop the reader, close stdin, terminate the subprocess.""" + self._fail_all_pending("connection closed") + if self._reader_task and not self._reader_task.done(): + self._reader_task.cancel() + try: + await self._reader_task + except (asyncio.CancelledError, Exception): + pass + self._reader_task = None + + proc = self._proc + self._proc = None + if proc is None: + return + if proc.stdin and not proc.stdin.is_closing(): + try: + proc.stdin.close() + except (OSError, RuntimeError): + pass + if proc.returncode is None: + try: + await asyncio.wait_for(proc.wait(), timeout=_CLOSE_TIMEOUT) + except asyncio.TimeoutError: + try: + proc.terminate() + await asyncio.wait_for( + proc.wait(), + timeout=_CLOSE_TIMEOUT, + ) + except (asyncio.TimeoutError, ProcessLookupError, OSError): + try: + proc.kill() + except (ProcessLookupError, OSError): + pass + await self._drain_stderr(proc) + + # ------------------------------------------------------------------ # + # MCP operations + # ------------------------------------------------------------------ # + async def list_tools(self) -> list[dict]: + resp = await self._request("tools/list") + if "error" in resp: + error = resp["error"] + message = error.get("message", str(error)) + raise MCPError(f"tools/list failed: {message}") + result = resp.get("result", {}) + self.tools = result.get("tools", []) + return self.tools + + async def list_prompts(self) -> list[dict]: + """Discover prompts/skills; returns [] when unsupported.""" + try: + resp = await self._request("prompts/list") + except MCPError: + return [] + if "error" in resp: + return [] + result = resp.get("result", {}) + self.prompts = result.get("prompts", []) + return self.prompts + + async def call_tool(self, tool_name: str, arguments: dict) -> str: + params = {"name": tool_name, "arguments": arguments} + resp = await self._request("tools/call", params) + if "error" in resp: + error = resp["error"] + message = error.get("message", str(error)) + return f"MCP Error: {message}" + result = resp.get("result", {}) + content = result.get("content", []) + parts: list[str] = [] + for item in content: + if isinstance(item, dict) and item.get("type") == "text": + parts.append(str(item.get("text", ""))) + else: + parts.append(json.dumps(item, ensure_ascii=False)) + return "\n".join(parts) if parts else str(result) + + # ------------------------------------------------------------------ # + # JSON-RPC plumbing + # ------------------------------------------------------------------ # + def _next_id(self) -> int: + self._request_id += 1 + return self._request_id + + async def _request( + self, + method: str, + params: dict | None = None, + ) -> dict: + """Send a request and await the matching response.""" + proc = self._proc + if proc is None or proc.stdin is None: + raise MCPError("stdio MCP client is not connected") + req_id = self._next_id() + payload = { + "jsonrpc": "2.0", + "method": method, + "params": params or {}, + "id": req_id, + } + future: asyncio.Future = asyncio.get_running_loop().create_future() + self._pending[req_id] = future + try: + await self._write(payload) + except MCPError: + self._pending.pop(req_id, None) + raise + try: + return await asyncio.wait_for(future, timeout=self.timeout) + except asyncio.TimeoutError: + self._pending.pop(req_id, None) + message = ( + f"MCP request '{method}' timed out " f"after {self.timeout:g}s" + ) + raise MCPError(message) from None + + async def _notify( + self, + method: str, + params: dict | None = None, + ) -> None: + """Send a notification (no id, no response expected).""" + if self._proc is None or self._proc.stdin is None: + raise MCPError("stdio MCP client is not connected") + payload = {"jsonrpc": "2.0", "method": method} + if params: + payload["params"] = params + await self._write(payload) + + async def _write(self, payload: dict) -> None: + proc = self._proc + if proc is None or proc.stdin is None: + raise MCPError("stdio MCP client is not connected") + line = json.dumps(payload, ensure_ascii=False) + "\n" + try: + proc.stdin.write(line.encode("utf-8")) + await proc.stdin.drain() + except ( + ConnectionResetError, + BrokenPipeError, + OSError, + RuntimeError, + ) as exc: + raise MCPError( + f"MCP server process is gone: {exc}", + ) from exc + + async def _read_loop(self) -> None: + """Read stdout lines and resolve pending futures by id.""" + proc = self._proc + if proc is None or proc.stdout is None: + return + try: + while True: + raw = await proc.stdout.readline() + if not raw: + self._fail_all_pending( + "MCP server closed stdout" + self._stderr_suffix(), + ) + return + line = raw.decode("utf-8", errors="replace").strip() + if not line: + continue + try: + msg = json.loads(line) + except json.JSONDecodeError: + # stdout is reserved for JSON-RPC messages; anything + # else is a protocol violation — fail in-flight + # requests instead of hanging until timeout. + self._stderr_tail = ( + self._stderr_tail + "non-JSON: " + line + "\n" + )[-_STDERR_TAIL:] + message = ( + "MCP server sent non-JSON output: " f"{line[:100]}" + ) + self._fail_all_pending(message) + return + if not isinstance(msg, dict): + continue + msg_id = msg.get("id") + if msg_id is None: + continue # server notification — ignore + future = self._pending.pop(msg_id, None) + if future is not None and not future.done(): + future.set_result(msg) + except asyncio.CancelledError: # pylint: disable=try-except-raise + raise + except Exception as exc: # stream broke unexpectedly + self._fail_all_pending(f"MCP stdio stream error: {exc}") + + def _fail_all_pending(self, reason: str) -> None: + pending, self._pending = self._pending, {} + for future in pending.values(): + if not future.done(): + future.set_exception(MCPError(reason)) + + async def _drain_stderr(self, proc: asyncio.subprocess.Process) -> None: + if proc.stderr is None: + return + try: + data = await asyncio.wait_for( + proc.stderr.read(_STDERR_TAIL), + timeout=1.0, + ) + if data: + text = data.decode("utf-8", errors="replace").strip() + self._stderr_tail = (self._stderr_tail + text)[-_STDERR_TAIL:] + except (asyncio.TimeoutError, OSError): + pass + + def _stderr_suffix(self) -> str: + tail = self._stderr_tail.strip() + return f" (stderr: {tail})" if tail else "" diff --git a/dashscope/acli/memory/directives_learning.py b/dashscope/acli/memory/directives_learning.py index 3dfe971..3595015 100644 --- a/dashscope/acli/memory/directives_learning.py +++ b/dashscope/acli/memory/directives_learning.py @@ -65,6 +65,33 @@ def _save_patterns(patterns: dict[str, Any]) -> None: _patterns_cache = None +# Recency decay: a pattern's influence fades over time so stale habits +# stop generating proposals. Weight is 1.0 now, 0.5 after the half-life. +_DECAY_HALF_LIFE_DAYS = 14.0 +# A pair is proposal-worthy when its recency-weighted frequency reaches this. +_MIN_WEIGHTED_FREQUENCY = 3.0 + + +def _recency_weight(timestamp_iso: str) -> float: + """Exponential-decay weight for a recorded sequence timestamp. + + Returns 1.0 for a just-recorded entry and halves every + ``_DECAY_HALF_LIFE_DAYS`` days. Missing or malformed timestamps + degrade to 1.0 (no decay) rather than discarding the data. + """ + try: + ts = datetime.fromisoformat(timestamp_iso) + except (ValueError, TypeError): + return 1.0 + if ts.tzinfo is None: + ts = ts.replace(tzinfo=timezone.utc) + age_days = max( + (datetime.now(timezone.utc) - ts).total_seconds() / 86400.0, + 0.0, + ) + return 0.5 ** (age_days / _DECAY_HALF_LIFE_DAYS) + + def record_tool_sequence(tools: list[str]) -> None: """Record a sequence of tools used in a successful turn.""" if len(tools) < 2: @@ -100,21 +127,29 @@ def analyze_patterns() -> list[dict[str, Any]]: if len(sequences) < 5: return [] # Not enough data - # Count tool sequence patterns - sequence_counter: Counter = Counter() + # Count tool sequence patterns. Two tallies are kept per adjacent + # pair: a raw occurrence count (reported as "frequency") and a + # recency-weighted score (recent repetitions count more, stale ones + # decay) used for the proposal threshold and confidence. + raw_counter: Counter = Counter() + weighted_counter: Counter = Counter() for seq in sequences: tools = tuple(seq.get("tools", [])) - if len(tools) >= 2: - # Look for adjacent pairs - for i in range(len(tools) - 1): - pair = (tools[i], tools[i + 1]) - sequence_counter[pair] += 1 - - # Find frequent patterns (>= 3 occurrences) + if len(tools) < 2: + continue + weight = _recency_weight(seq.get("timestamp", "")) + # Look for adjacent pairs + for i in range(len(tools) - 1): + pair = (tools[i], tools[i + 1]) + raw_counter[pair] += 1 + weighted_counter[pair] += weight + + # Find frequent patterns (weighted frequency >= threshold) proposals = [] - for (tool1, tool2), count in sequence_counter.most_common(10): - if count >= 3: - confidence = min(count / 10, 1.0) + for (tool1, tool2), weighted in weighted_counter.most_common(10): + if weighted >= _MIN_WEIGHTED_FREQUENCY: + confidence = min(weighted / 10, 1.0) + count = raw_counter[(tool1, tool2)] directive = _generate_directive(tool1, tool2, count) proposals.append( { diff --git a/dashscope/acli/memory/experience.py b/dashscope/acli/memory/experience.py index 4bebefa..781e42b 100644 --- a/dashscope/acli/memory/experience.py +++ b/dashscope/acli/memory/experience.py @@ -115,6 +115,9 @@ def search_experiences( if score > 0: if exp.get("lesson"): score += 1 + # Cautionary lessons are high-value recall targets. + if exp.get("outcome") == "failure": + score += 1 scored.append((score, index, exp)) # Highest score first; ties favor the most recently recorded entry. diff --git a/dashscope/acli/memory/skill_evolution.py b/dashscope/acli/memory/skill_evolution.py index ec8ece1..b9a3f32 100644 --- a/dashscope/acli/memory/skill_evolution.py +++ b/dashscope/acli/memory/skill_evolution.py @@ -74,6 +74,21 @@ def analyze_trajectory( if any(kw in last_assistant.lower() for kw in error_keywords): return None # Failed trajectory, not skill-worthy + # Tool-outcome check: the final tool call must have succeeded. Earlier + # tool errors are tolerated (a workflow that fails then recovers — e.g. + # edit-and-test — is still a valuable skill), but a trajectory whose + # last tool call errored did not complete cleanly, so distilling it + # would produce a low-quality skill. + for msg in reversed(messages): + if msg.get("role") != "tool": + continue + content = msg.get("content", "") + if isinstance(content, str) and content.startswith( + ("Error", "错误"), + ): + return None + break # only the final tool outcome matters + # Identify common patterns pattern_name = _identify_pattern(tool_sequence) if not pattern_name: diff --git a/dashscope/acli/memory/trace.py b/dashscope/acli/memory/trace.py index b3e41dc..ff41e20 100644 --- a/dashscope/acli/memory/trace.py +++ b/dashscope/acli/memory/trace.py @@ -189,9 +189,9 @@ def generate_report(trace_logger: TraceLogger | None) -> dict | None: return { "total_llm_calls": llm_calls, "total_tool_calls": tool_calls, - "tool_success_rate": (tool_successes / tool_calls) - if tool_calls - else 0.0, + "tool_success_rate": ( + (tool_successes / tool_calls) if tool_calls else 0.0 + ), "avg_response_time": avg_response_time, "top_tools": top_tools, } diff --git a/dashscope/acli/platforms/base.py b/dashscope/acli/platforms/base.py index e28d071..39b2e7f 100644 --- a/dashscope/acli/platforms/base.py +++ b/dashscope/acli/platforms/base.py @@ -206,7 +206,11 @@ def list_files( ) -> list[FileInfo]: ... - def delete_file(self, file_id: str, category_id: str = "default") -> bool: + def delete_file( + self, + file_id: str, + category_id: str = "default", + ) -> bool: ... def list_categories( diff --git a/dashscope/acli/prompt_pipeline.py b/dashscope/acli/prompt_pipeline.py index 2f98a12..40aa13f 100644 --- a/dashscope/acli/prompt_pipeline.py +++ b/dashscope/acli/prompt_pipeline.py @@ -33,6 +33,7 @@ class PromptContext: experience_tracker: Any = None disabled_caps_provider: Callable[[], str] | None = None directives_provider: Callable[[], list[str]] | None = None + scene_provider: Callable[[], str] | None = None current_turn_tools: list[str] = field(default_factory=list) connected_mcp_services: Callable[[], list[str]] | None = None @@ -124,6 +125,26 @@ def render(self, ctx: PromptContext) -> str: return "\n".join(lines) +class SceneSection: + """Persistent per-topic scene memory (see SessionManager.get_scene).""" + + name = "scene" + + def render(self, ctx: PromptContext) -> str: + if not ctx.scene_provider: + return "" + try: + text = (ctx.scene_provider() or "").strip() + except Exception: + return "" + if not text: + return "" + return ( + "\n\n## Scene memory (persistent notes for the current " + "session topic; treat as standing context)\n" + text + ) + + class PlanSection: name = "plan" @@ -273,6 +294,7 @@ def default_pipeline( .add_ephemeral(SkillPackagesSection(active_prompts_fn)) .add_ephemeral(DisabledCapsSection()) .add_ephemeral(DirectivesSection()) + .add_ephemeral(SceneSection()) .add_ephemeral(PlanSection()) .add_ephemeral(ExperienceSection()) .add_ephemeral(ToolChainsSection()) diff --git a/dashscope/acli/providers/openai.py b/dashscope/acli/providers/openai.py index a956cf8..ad5c1b1 100644 --- a/dashscope/acli/providers/openai.py +++ b/dashscope/acli/providers/openai.py @@ -148,9 +148,11 @@ async def chat( "output_tokens": getattr(resp_usage, "completion_tokens", 0) or 0, "total_tokens": getattr(resp_usage, "total_tokens", 0) or 0, - "cached_tokens": (getattr(details, "cached_tokens", 0) or 0) - if details - else 0, + "cached_tokens": ( + (getattr(details, "cached_tokens", 0) or 0) + if details + else 0 + ), } return LLMResponse( @@ -208,10 +210,10 @@ async def chat_stream( or 0, "total_tokens": getattr(usage, "total_tokens", 0) or 0, "cached_tokens": ( - getattr(details, "cached_tokens", 0) or 0 - ) - if details - else 0, + (getattr(details, "cached_tokens", 0) or 0) + if details + else 0 + ), } continue diff --git a/dashscope/acli/sandbox.py b/dashscope/acli/sandbox.py new file mode 100644 index 0000000..912cedc --- /dev/null +++ b/dashscope/acli/sandbox.py @@ -0,0 +1,181 @@ +# -*- coding: utf-8 -*- +"""Optional OS-level sandbox for shell command execution (roadmap P0-L3). + +Defense-in-depth layered on top of the permission engine and the +dangerous-command blocklist. When enabled *and* a sandbox backend is +present, ``run_command`` executes commands inside an OS sandbox that +confines filesystem writes to the current workspace. + +Design constraints (see the roadmap): + +* **Opt-in** — disabled by default (``sandbox = false`` in config). Local + users already get the permission prompts + blocklist; the sandbox is an + extra layer for ``auto_approve`` / untrusted-skill scenarios. +* **Graceful degradation** — if no backend is detected the command runs + normally under the existing permission layer. The sandbox never blocks + startup and never raises out of ``run_command``. +* **Best-effort boundary** — this raises the cost of a destructive or + runaway command; it is *not* a hardened security boundary. Treat it as + confinement, not isolation. + +Backends: + +* macOS: Seatbelt via ``sandbox-exec`` (built into macOS). The profile + keeps the default-allow policy but denies filesystem writes outside the + workspace and temp areas, so process execution, reads, and networking are + unaffected and normal dev commands keep working. +* Linux: bubblewrap (``bwrap``) when installed. bwrap confines by + remapping the filesystem, so it is more restrictive; commands that write + outside the workspace/temp will fail. +""" + +from __future__ import annotations + +import os +import shutil +import sys +from typing import Optional + +# Backend identifiers. +SEATBELT = "seatbelt" +BWRAP = "bwrap" + + +def detect_backend() -> Optional[str]: + """Return the available sandbox backend name, or ``None``. + + macOS prefers ``sandbox-exec`` (Seatbelt); Linux prefers ``bwrap``. + Windows has no supported backend. Detection only checks for the tool's + presence on PATH — it never starts a sandbox. + """ + if sys.platform == "darwin": + return SEATBELT if shutil.which("sandbox-exec") else None + if sys.platform.startswith("linux"): + return BWRAP if shutil.which("bwrap") else None + return None + + +def available() -> bool: + """True when a sandbox backend is detected on this machine.""" + return detect_backend() is not None + + +# Optional override (mainly for tests / explicit CLI control). ``None`` +# means "read the setting from config". +_enabled_override: Optional[bool] = None + + +def set_enabled(value: Optional[bool]) -> None: + """Override sandbox enablement; pass ``None`` to use the config value.""" + global _enabled_override + _enabled_override = value + + +def is_enabled() -> bool: + """Whether sandboxing is enabled (config-driven, overridable). + + Reads ``sandbox`` from the loaded config on each call so config + changes take effect; degrades to ``False`` on any error so the + sandbox can never break command execution. + """ + if _enabled_override is not None: + return bool(_enabled_override) + try: + from dashscope.acli.config import Config + + return bool(Config.load().sandbox) + except Exception: + return False + + +def seatbelt_profile(cwd: str) -> str: + """Build a macOS Seatbelt profile confining writes to the workspace. + + The base policy stays default-allow; only filesystem writes are + denied, then re-allowed for the workspace and standard temp/cache + locations. Reads, process execution, and networking are untouched so + ordinary development commands keep working. + """ + safe_cwd = cwd.replace('"', '\\"') + return "\n".join( + [ + "(version 1)", + "(allow default)", + "(deny file-write*)", + f'(allow file-write* (subpath "{safe_cwd}"))', + '(allow file-write* (subpath "/private/tmp"))', + '(allow file-write* (subpath "/tmp"))', + '(allow file-write* (subpath "/private/var/folders"))', + '(allow file-write* (literal "/dev/null"))', + ], + ) + + +def bwrap_argv(command: str, cwd: str) -> list[str]: + """Build a ``bwrap`` argv that runs *command* with a read-only root. + + The whole filesystem is bound read-only, then the workspace and a + fresh ``/tmp`` are made writable. Commands that need to write elsewhere + (e.g. package caches) will fail — this is the intended confinement. + """ + return [ + "bwrap", + "--ro-bind", + "/", + "/", + "--bind", + cwd, + cwd, + "--tmpfs", + "/tmp", + "--proc", + "/proc", + "--dev", + "/dev", + "--chdir", + cwd, + "/bin/sh", + "-c", + command, + ] + + +def build_argv( + command: str, + cwd: str, + backend: Optional[str] = None, +) -> Optional[list[str]]: + """Return an argv to run *command* inside the sandbox, or ``None``. + + ``None`` means "no sandbox available — run the command normally". + ``backend`` may be passed explicitly (mainly for tests); otherwise it + is detected. + """ + backend = backend or detect_backend() + if backend == SEATBELT: + return [ + "sandbox-exec", + "-p", + seatbelt_profile(cwd), + "/bin/sh", + "-c", + command, + ] + if backend == BWRAP: + return bwrap_argv(command, cwd) + return None + + +def is_sandboxed_path(path: str, cwd: str) -> bool: + """True when *path* is inside the writable workspace or temp areas. + + Informational helper (e.g. for messages); not a security check. + """ + try: + abs_path = os.path.abspath(path) + except (OSError, ValueError): + return False + for base in (cwd, "/tmp", "/private/tmp"): + if abs_path == base or abs_path.startswith(base + os.sep): + return True + return False diff --git a/dashscope/acli/session.py b/dashscope/acli/session.py index 07e690c..98b42c4 100644 --- a/dashscope/acli/session.py +++ b/dashscope/acli/session.py @@ -5,6 +5,10 @@ - history.json: message history - input-history.txt: command input history - meta.json: metadata (created, last_accessed, message_count) + - scene.md: persistent scene memory (topic-scoped notes injected into + the system prompt every turn) + - events.jsonl: append-only event sidecar (lifecycle; see + session_events.SessionEventLog) """ from __future__ import annotations @@ -13,10 +17,24 @@ from dataclasses import asdict, dataclass from datetime import datetime from pathlib import Path +from typing import Any from dashscope.acli.config import WORKSPACE_DIR +from dashscope.acli.session_events import ( + EVENTS_FILENAME, + SessionEventLog, + latest_snapshot_messages, +) DEFAULT_TOPIC = "default" +SCENE_FILENAME = "scene.md" + +# Snapshot pruning (event-log completeness Phase 2): snapshots carry a +# full message copy, so unbounded retention grows the log quadratically. +# Compact once the log holds more than _SNAPSHOT_COMPACT_AT snapshots, +# keeping the newest _SNAPSHOT_KEEP. +_SNAPSHOT_KEEP = 2 +_SNAPSHOT_COMPACT_AT = 8 @dataclass @@ -102,6 +120,182 @@ def _meta_file(self, topic: str) -> Path: """Get the meta.json path for a topic.""" return self._topic_dir(topic) / "meta.json" + def _events_file(self, topic: str) -> Path: + """Get the events.jsonl sidecar path for a topic.""" + return self._topic_dir(topic) / EVENTS_FILENAME + + def _scene_file(self, topic: str) -> Path: + """Get the scene.md path for a topic.""" + return self._topic_dir(topic) / SCENE_FILENAME + + def get_scene(self, topic: str | None = None) -> str: + """Return the scene memory text for a topic (default: current). + + Returns an empty string when the file is missing or unreadable. + """ + path = self._scene_file(topic or self.current_topic) + if not path.exists(): + return "" + try: + return path.read_text(encoding="utf-8").strip() + except OSError: + return "" + + def set_scene(self, text: str, topic: str | None = None) -> bool: + """Replace the scene memory for a topic (default: current). + + An empty/whitespace-only *text* clears the scene file. Returns + False when the topic name is unsafe or the write fails. + """ + topic = topic or self.current_topic + topic_dir = self._safe_topic_dir(topic) + if topic_dir is None: + return False + path = self._scene_file(topic) + content = text.strip() + try: + if not content: + if path.exists(): + path.unlink() + else: + topic_dir.mkdir(parents=True, exist_ok=True) + path.write_text(content + "\n", encoding="utf-8") + except OSError: + return False + self.event_log(topic).append( + "scene/updated", + {"topic": topic, "chars": len(content)}, + ) + return True + + def append_scene(self, text: str, topic: str | None = None) -> bool: + """Append a note line to the scene memory of a topic.""" + topic = topic or self.current_topic + note = text.strip() + if not note: + return False + existing = self.get_scene(topic) + merged = f"{existing}\n{note}" if existing else note + return self.set_scene(merged, topic) + + def event_log(self, topic: str | None = None) -> SessionEventLog: + """Return the append-only event log for a topic (default: current). + + The sidecar is advisory: callers may append lifecycle/turn events + without affecting the history.json read/write path. + """ + return SessionEventLog( + self._events_file(topic or self.current_topic), + ) + + def record_turn_event( + self, + user_text: str, + assistant_text: str, + tools_used: list[str] | None = None, + outcome: str = "", + topic: str | None = None, + ) -> None: + """Append a ``turn/end`` event to the topic's event sidecar. + + Records a compact summary of one turn (truncated user/assistant + text, tools used, outcome). Best-effort: never raises, so event + recording can't break the agent loop. This is a step toward the + session-as-event-log direction (later: projection / fork / + resume built on the event stream). + """ + try: + topic = topic or self.current_topic + self.event_log(topic).append( + "turn/end", + { + "topic": topic, + "user": (user_text or "")[:200], + "assistant": (assistant_text or "")[:200], + "tools": list(tools_used or []), + "outcome": outcome, + }, + ) + except Exception: + pass + + def record_messages_snapshot( + self, + messages: list[dict[str, Any]], + topic: str | None = None, + ) -> None: + """Append a full-fidelity ``messages/snapshot`` event. + + Stores the complete current message list so the session can later + be reconstructed (resume/fork) from the event log alone. Old + snapshots are pruned once more than ``_SNAPSHOT_COMPACT_AT`` + accumulate, keeping the newest ``_SNAPSHOT_KEEP``. Best-effort: + never raises. + """ + try: + topic = topic or self.current_topic + log = self.event_log(topic) + log.append( + "messages/snapshot", + {"topic": topic, "messages": list(messages or [])}, + ) + if len(log.read("messages/snapshot")) > _SNAPSHOT_COMPACT_AT: + log.compact_snapshots(keep=_SNAPSHOT_KEEP) + except Exception: + pass + + def resume_from_events( + self, + topic: str | None = None, + ) -> list[dict[str, Any]]: + """Rebuild the message list from the latest snapshot event. + + Returns an empty list when the topic has no snapshot. This is the + projection side of the event-sourced session: combined with the + append-only log it enables resume / fork / crash recovery. + """ + try: + events = self.event_log(topic).read_raw() + except Exception: + return [] + return latest_snapshot_messages(events) + + def fork_topic(self, src: str, dst: str) -> bool: + """Create topic *dst* seeded with a copy of *src*'s event log. + + The forked topic resumes from the same state as the source. Both + names must be safe and the destination must not already exist. + Returns False otherwise. + """ + src_dir = self._safe_topic_dir(src) + dst_dir = self._safe_topic_dir(dst) + if src_dir is None or dst_dir is None: + return False + if not src_dir.exists() or dst_dir.exists(): + return False + try: + dst_dir.mkdir(parents=True) + self._save_meta( + dst, + SessionMeta( + topic=dst, + created=datetime.now().isoformat(), + last_accessed=datetime.now().isoformat(), + ), + ) + src_events = self._events_file(src) + if src_events.exists(): + import shutil + + shutil.copy2(src_events, self._events_file(dst)) + except OSError: + return False + self.event_log(dst).append( + "topic/forked", + {"from": src, "to": dst}, + ) + return True + def list_topics(self) -> list[SessionMeta]: """List all available topics with metadata.""" topics = [] @@ -164,6 +358,7 @@ def set_current_topic(self, topic: str) -> bool: return False self.current_topic = topic self._update_last_accessed(topic) + self.event_log(topic).append("topic/switched", {"topic": topic}) return True def create_topic(self, topic: str) -> bool: @@ -179,6 +374,7 @@ def create_topic(self, topic: str) -> bool: ) self._save_meta(topic, meta) self.current_topic = topic + self.event_log(topic).append("topic/created", {"topic": topic}) return True def rename_topic(self, old_name: str, new_name: str) -> bool: @@ -221,6 +417,10 @@ def rename_topic(self, old_name: str, new_name: str) -> bool: if self.current_topic == old_name: self.current_topic = new_name + self.event_log(new_name).append( + "topic/renamed", + {"from": old_name, "to": new_name}, + ) return True except OSError: return False @@ -254,6 +454,24 @@ def get_input_history_path(self, topic: str | None = None) -> Path: topic = topic or self.current_topic return self._input_history_file(topic) + def load_messages( + self, + topic: str | None = None, + ) -> list[dict[str, Any]]: + """Load stored chat messages for a topic (default: current). + + Returns an empty list when the history file is missing, + unreadable, or not a JSON list. + """ + path = self._history_file(topic or self.current_topic) + if not path.exists(): + return [] + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (json.JSONDecodeError, OSError): + return [] + return data if isinstance(data, list) else [] + def update_message_count( self, count: int, diff --git a/dashscope/acli/session_events.py b/dashscope/acli/session_events.py new file mode 100644 index 0000000..777845e --- /dev/null +++ b/dashscope/acli/session_events.py @@ -0,0 +1,210 @@ +# -*- coding: utf-8 -*- +"""Append-only session event log (roadmap P2-Phase2, first increment). + +An event sidecar that records session lifecycle (and, in later +increments, turn) events as an append-only JSONL stream, without +touching the existing ``history.json`` read/write path. This lays the +groundwork for future projection / fork / resume built on an immutable +event source. + +Storage layout (per topic):: + + .acli/session//events.jsonl + +Each line is a self-describing event:: + + {"v": 1, "seq": 3, "ts": "", "type": "topic/created", + "data": {...}} + +Writes are best-effort: an event-log failure never breaks the session, +mirroring the checkpoint/history philosophy. +""" + +from __future__ import annotations + +import json +import os +from datetime import datetime +from pathlib import Path +from typing import Any + +# Bump when the on-disk event schema changes in a breaking way. +SCHEMA_VERSION = 1 + +EVENTS_FILENAME = "events.jsonl" + + +def latest_snapshot_messages( + entries: list[dict[str, Any]], +) -> list[dict[str, Any]]: + """Return the message list from the newest ``messages/snapshot``. + + Scans *entries* (as produced by :meth:`SessionEventLog.read_raw`) + backwards; returns an empty list when no snapshot is present. + """ + for entry in reversed(entries): + if entry.get("type") != "messages/snapshot": + continue + data = entry.get("data") or {} + msgs = data.get("messages") + if isinstance(msgs, list): + return msgs + return [] + + +class SessionEventLog: + """Append-only JSONL event log bound to one file. + + Events are appended, never edited in place — with one exception: + :meth:`compact_snapshots` atomically rewrites the file to drop + stale full-history snapshots (all other events are preserved). + ``seq`` is a 1-based monotonically increasing position derived + from the current line count, so it stays correct even if another + process appends. + """ + + def __init__(self, events_file: Path): + self._file = Path(events_file) + + @property + def path(self) -> Path: + return self._file + + def append(self, event_type: str, data: dict | None = None) -> None: + """Append one event. Never raises (best-effort sidecar).""" + try: + self._append(event_type, data or {}) + except Exception: + # The event log is advisory; never break the caller. + pass + + def _append(self, event_type: str, data: dict) -> None: + self._file.parent.mkdir(parents=True, exist_ok=True) + entry = { + "v": SCHEMA_VERSION, + "seq": self._next_seq(), + "ts": datetime.now().isoformat(), + "type": event_type, + "data": data, + } + with open(self._file, "a", encoding="utf-8") as f: + f.write(json.dumps(entry, ensure_ascii=False) + "\n") + + def _next_seq(self) -> int: + """1 + number of non-blank lines currently in the log.""" + if not self._file.exists(): + return 1 + count = 0 + with open(self._file, "r", encoding="utf-8") as f: + for line in f: + if line.strip(): + count += 1 + return count + 1 + + def read( + self, + event_type: str | None = None, + ) -> list[dict[str, Any]]: + """Return events in order, optionally filtered by type. + + Malformed or wrong-schema lines are skipped silently. + """ + if not self._file.exists(): + return [] + events: list[dict[str, Any]] = [] + with open(self._file, "r", encoding="utf-8") as f: + for line in f: + line = line.strip() + if not line: + continue + try: + entry = json.loads(line) + except json.JSONDecodeError: + continue + if not isinstance(entry, dict): + continue + if entry.get("v") != SCHEMA_VERSION: + continue + if event_type is not None and entry.get("type") != event_type: + continue + events.append(entry) + return events + + def tail( + self, + n: int, + event_type: str | None = None, + ) -> list[dict[str, Any]]: + """Return the most recent ``n`` events (optionally by type).""" + if n <= 0: + return [] + return self.read(event_type)[-n:] + + def read_raw(self) -> list[dict[str, Any]]: + """Return every well-formed entry, any schema version. + + Unlike :meth:`read` (which filters to the current schema), this + yields all parseable entries — used when projecting large events + such as ``messages/snapshot``. + """ + if not self._file.exists(): + return [] + entries: list[dict[str, Any]] = [] + with open(self._file, "r", encoding="utf-8") as f: + for line in f: + line = line.strip() + if not line: + continue + try: + entry = json.loads(line) + except json.JSONDecodeError: + continue + if isinstance(entry, dict): + entries.append(entry) + return entries + + def compact_snapshots( + self, + keep: int = 2, + snapshot_type: str = "messages/snapshot", + ) -> bool: + """Drop all but the newest ``keep`` snapshot events, atomically. + + Snapshots carry a full copy of the message list, so retaining + every one grows the log quadratically in history length. + Compaction rewrites the file with all non-snapshot events plus + the newest ``keep`` snapshots, renumbering ``seq`` to stay + line-ordered. The rewrite goes through a tmp file and + ``os.replace``, so a crash mid-compaction leaves the original + log intact. + + Best-effort: returns True only when the file was rewritten. + """ + try: + entries = self.read_raw() + except OSError: + return False + snapshot_idx = [ + i for i, e in enumerate(entries) if e.get("type") == snapshot_type + ] + if len(snapshot_idx) <= keep: + return False + drop = set(snapshot_idx[: len(snapshot_idx) - keep]) + kept = [e for i, e in enumerate(entries) if i not in drop] + tmp = self._file.with_suffix(self._file.suffix + ".tmp") + try: + with open(tmp, "w", encoding="utf-8") as f: + for seq, entry in enumerate(kept, start=1): + entry["seq"] = seq + f.write(json.dumps(entry, ensure_ascii=False) + "\n") + os.replace(tmp, self._file) + except OSError: + try: + tmp.unlink(missing_ok=True) + except OSError: + pass + return False + return True + + def __len__(self) -> int: + return len(self.read()) diff --git a/dashscope/acli/tools/checkpoint.py b/dashscope/acli/tools/checkpoint.py new file mode 100644 index 0000000..e6f84af --- /dev/null +++ b/dashscope/acli/tools/checkpoint.py @@ -0,0 +1,167 @@ +# -*- coding: utf-8 -*- +"""File checkpoint/undo support for mutating file tools. + +Backups live under ``/.acli/checkpoints/`` next to a JSONL +index. ``snapshot()`` records the pre-mutation state of a file and +``undo()`` reverses the most recent recorded mutation. +""" + +from __future__ import annotations + +import json +import os +import shutil +import time +import uuid +from pathlib import Path + +from rich.console import Console + +console = Console() + +# Maximum number of checkpoint entries kept in the index. +_MAX_ENTRIES = 50 + +_INDEX_NAME = "index.jsonl" + +# Maps a checkpoint action to the tool name shown in undo messages. +_TOOL_BY_ACTION = { + "overwrite": "write_file", + "create": "write_file", + "delete": "delete_file", +} + + +def _checkpoint_dir() -> Path: + """Directory holding backup files and the JSONL index.""" + # Lazy import so loading this module never triggers config cycles. + from dashscope.acli.config import WORKSPACE_DIR + + return Path(WORKSPACE_DIR) / "checkpoints" + + +def _read_entries(cp_dir: Path) -> list[dict]: + """Load index entries, skipping blank or malformed lines.""" + index = cp_dir / _INDEX_NAME + if not index.is_file(): + return [] + entries: list[dict] = [] + with open(index, "r", encoding="utf-8") as f: + for line in f: + line = line.strip() + if not line: + continue + try: + entries.append(json.loads(line)) + except json.JSONDecodeError: + continue + return entries + + +def _write_entries(cp_dir: Path, entries: list[dict]) -> None: + """Rewrite the index atomically (temp file + rename).""" + index = cp_dir / _INDEX_NAME + tmp = index.with_name(_INDEX_NAME + ".tmp") + with open(tmp, "w", encoding="utf-8") as f: + for entry in entries: + f.write(json.dumps(entry, ensure_ascii=False) + "\n") + os.replace(tmp, index) + + +def snapshot(path: str, action: str) -> None: + """Record a pre-mutation checkpoint for *path*. + + *action* is one of ``"overwrite"``, ``"create"`` or ``"delete"``. + Never raises: a checkpoint failure must not break the write tool. + """ + try: + _snapshot(path, action) + except Exception: + # Checkpointing is best-effort; skip silently on any error. + pass + + +def _snapshot(path: str, action: str) -> None: + """Internal snapshot implementation (may raise).""" + abs_path = os.path.abspath(path) + cp_dir = _checkpoint_dir() + cp_dir.mkdir(parents=True, exist_ok=True) + entry_id = uuid.uuid4().hex + backup = None + if os.path.isfile(abs_path): + backup = entry_id + ".bak" + shutil.copy2(abs_path, cp_dir / backup) + entry = { + "id": entry_id, + "path": abs_path, + "backup": backup, + "action": action, + "ts": time.time(), + } + entries = _read_entries(cp_dir) + entries.append(entry) + overflow = len(entries) - _MAX_ENTRIES + dropped = entries[:overflow] if overflow > 0 else [] + _write_entries(cp_dir, entries[-_MAX_ENTRIES:]) + for old in dropped: + name = old.get("backup") + if not name: + continue + try: + os.remove(cp_dir / name) + except OSError: + pass + + +def undo() -> str: + """Undo the most recent checkpointed file mutation. + + Returns a human-readable result string; never raises. + """ + try: + return _undo() + except Exception as e: + return f"Error: undo failed - {e}" + + +def _undo() -> str: + """Internal undo implementation (may raise).""" + cp_dir = _checkpoint_dir() + entries = _read_entries(cp_dir) + if not entries: + return "Nothing to undo" + entry = entries.pop() + _write_entries(cp_dir, entries) + + path = entry.get("path") or "" + action = entry.get("action") or "" + backup = entry.get("backup") + tool_name = _TOOL_BY_ACTION.get(action, action or "unknown") + + if backup: + src = cp_dir / backup + if not src.is_file(): + return f"Error: backup file missing for {path}" + parent = os.path.dirname(path) + if parent: + os.makedirs(parent, exist_ok=True) + shutil.copy2(src, path) + try: + os.remove(src) + except OSError: + pass + return f"Undid {tool_name}: restored {path}" + + if action == "create": + try: + os.remove(path) + except FileNotFoundError: + pass + return f"Undid {tool_name}: removed created {path}" + + return f"Error: no backup recorded for {path}" + + +def handle_undo_command() -> None: + """CLI-facing wrapper: undo the last change and print the result.""" + console.print(undo()) diff --git a/dashscope/acli/tools/filesystem.py b/dashscope/acli/tools/filesystem.py index 9dbf595..54985f1 100644 --- a/dashscope/acli/tools/filesystem.py +++ b/dashscope/acli/tools/filesystem.py @@ -91,6 +91,12 @@ def write_file(path: str, content: str) -> str: except (OSError, UnicodeDecodeError): existed = False # treat as new for diff purposes + # Checkpoint the current state so /undo can reverse this write. + # Lazy import to avoid tool-module import cycles. + from dashscope.acli.tools import checkpoint + + checkpoint.snapshot(path, "overwrite" if existed else "create") + with open(path, "w", encoding="utf-8") as f: f.write(content) @@ -187,6 +193,10 @@ def delete_file(path: str) -> str: return f"Error: {e}" if not os.path.isfile(path): return f"Error: file not found - {path}" + # Checkpoint so /undo can restore the deleted file. + from dashscope.acli.tools import checkpoint + + checkpoint.snapshot(path, "delete") os.remove(path) return f"Deleted file: {path}" diff --git a/dashscope/acli/tools/shell.py b/dashscope/acli/tools/shell.py index 1fb6832..f376b05 100644 --- a/dashscope/acli/tools/shell.py +++ b/dashscope/acli/tools/shell.py @@ -21,6 +21,10 @@ "dd if=", "> /dev/sd", "chmod -R 777 /", + "shred", + "wipefs", + # rm with this flag always targets `/` — no benign use exists + "--no-preserve-root", # Windows "rd /s /q", "del /f /q /s", @@ -35,6 +39,73 @@ _RM_ROOT_RE = re.compile(r"\brm\s+-\w*[rf]\w*\s+/(?:\s*\*?\s*(?:$|[;&|]))") _FORMAT_CMD_RE = re.compile(r"(?:^|[;&|]\s*)format(?:\s|$)") +# Home / cwd wipes: `rm -rf ~`, `rm -rf .`, but NOT `rm -rf ./build`. +_RM_HOME_RE = re.compile( + r"\brm\s+(?:-\w+\s+)*-\w*[rf]\w*\s+(?:--\s+)?(?:~|\.\.?)/?" + r"(?=\s|$|[;&|])", +) +# Fork bombs: a function that pipes itself into itself in the background, +# e.g. `:(){ :|:& };:` or `bomb(){ bomb|bomb& };bomb`. +_FORK_BOMB_RE = re.compile( + r"(\S+)\s*\(\)\s*\{\s*\1\s*\|\s*\1\s*&\s*\}\s*;", +) +# Redirects writing to raw block devices (`> /dev/sda`, `>>/dev/nvme0n1`). +_DEV_WRITE_RE = re.compile( + r">\s*/dev/(?:sd|hd|vd|xvd|nvme|mmcblk|disk)\w*", +) +# Recursive chown of whole system trees (`chown -R u:g /etc`, `/`, ...). +# `/home` & co only match bare: `chown -R u /home/lzs` is everyday work. +_CHOWN_SYSTEM_RE = re.compile( + r"\bchown\s+(?:-\w+\s+)*-R\s+\S+\s+" + r"(?:/(?:etc|usr|bin|sbin|lib64|lib|boot|dev|sys)(?:[/\s;&|]|$)" + r"|/(?:home|var|opt|srv|root)?(?:[\s;&|]|$))", +) +# Power-state commands as actual command tokens (incl. `sudo reboot`). +_SYSTEM_STATE_RE = re.compile( + r"(?:^|[;&|]\s*)(?:sudo\s+)?" + r"(?:shutdown|reboot|halt|poweroff)(?:\s|$|[;&|])", +) +_INIT_HALT_RE = re.compile( + r"(?:^|[;&|]\s*)(?:sudo\s+)?init\s+[06](?:\s|$|[;&|])", +) +_SYSTEMCTL_HALT_RE = re.compile( + r"\bsystemctl\s+(?:-\S+\s+)*(?:poweroff|reboot|halt)(?:\s|$|[;&|])", +) +# Killing PID 1 drags the whole system down with it. +_KILL_PID1_RE = re.compile(r"\bkill\s+(?:-\S+\s+)*1(?:\s|$|[;&|])") +# Shell-history destruction (`history -c`, `history -cw`). +_HISTORY_CLEAR_RE = re.compile(r"\bhistory\s+-\w*c") +# Remote-exec pipes: `curl ... | sh`, `wget ... | bash`, with any flags +# between the downloader and the pipe. +_PIPE_TO_SHELL_RE = re.compile( + r"\b(?:curl|wget)\b[^\n]*\|\s*(?:sudo\s+)?(?:sh|bash|zsh|dash|ksh)\b", +) +# Partition-table editors as actual command tokens. +_PARTITION_CMD_RE = re.compile( + r"(?:^|[;&|]\s*)(?:sudo\s+)?(?:fdisk|parted)(?:\s|$)", +) +# macOS: whole-disk erase and Secure Boot bypass. +_DISKUTIL_ERASE_RE = re.compile(r"\bdiskutil\s+eraseDisk\b") +_CSRUTIL_DISABLE_RE = re.compile(r"\bcsrutil\s+disable\b") + +# (label, regex) pairs checked by run_command after BLOCKED_PATTERNS; +# the label is quoted in the block error message. +_BLOCKED_REGEXES: list[tuple[str, re.Pattern[str]]] = [ + ("rm -rf ~", _RM_HOME_RE), + ("fork bomb", _FORK_BOMB_RE), + ("> /dev/", _DEV_WRITE_RE), + ("chown -R on system paths", _CHOWN_SYSTEM_RE), + ("shutdown/reboot/halt/poweroff", _SYSTEM_STATE_RE), + ("init 0/init 6", _INIT_HALT_RE), + ("systemctl poweroff/reboot/halt", _SYSTEMCTL_HALT_RE), + ("kill PID 1", _KILL_PID1_RE), + ("history -c", _HISTORY_CLEAR_RE), + ("curl/wget piped into a shell", _PIPE_TO_SHELL_RE), + ("fdisk/parted", _PARTITION_CMD_RE), + ("diskutil eraseDisk", _DISKUTIL_ERASE_RE), + ("csrutil disable", _CSRUTIL_DISABLE_RE), +] + # ----- Read-only command classifier --------------------------------------- # Used by the executor to auto-approve obvious inspection commands so the # user isn't asked to confirm every `grep`/`ls`/`git status`. Conservative @@ -439,6 +510,12 @@ async def run_command(command: str, timeout: int | None = None) -> str: return "Error: command blocked (contains dangerous pattern: rm -rf /)" if _FORMAT_CMD_RE.search(command): return "Error: command blocked (contains dangerous pattern: format)" + for name, regex in _BLOCKED_REGEXES: + if regex.search(command): + return ( + f"Error: command blocked " + f"(contains dangerous pattern: {name})" + ) # Belt-and-suspenders on top of utils.validation.coerce_types: if the model # still managed to slip something un-castable through (e.g. "auto"), @@ -461,12 +538,27 @@ async def run_command(command: str, timeout: int | None = None) -> str: cwd=os.getcwd(), ) else: - proc = await asyncio.create_subprocess_shell( - command, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE, - cwd=os.getcwd(), - ) + # Optional OS sandbox (opt-in; degrades to normal execution + # when disabled or when no backend is available). + from dashscope.acli import sandbox + + sandbox_argv = None + if sandbox.is_enabled(): + sandbox_argv = sandbox.build_argv(command, os.getcwd()) + if sandbox_argv is not None: + proc = await asyncio.create_subprocess_exec( + *sandbox_argv, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + cwd=os.getcwd(), + ) + else: + proc = await asyncio.create_subprocess_shell( + command, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + cwd=os.getcwd(), + ) stdout, stderr = await asyncio.wait_for( proc.communicate(), timeout=timeout, diff --git a/dashscope/acli/ui/tui.py b/dashscope/acli/ui/tui.py index e3deafe..2516754 100644 --- a/dashscope/acli/ui/tui.py +++ b/dashscope/acli/ui/tui.py @@ -35,6 +35,18 @@ os.environ.get("TERMINAL_EMULATOR", "").startswith("JetBrains") or "jediterm" in os.environ.get("TERM_PROGRAM", "").lower() ) + +# Temporary instrumentation for the iTerm2 drag auto-scroll investigation. +_ACLI_DEBUG_SELECT = bool(os.environ.get("ACLI_DEBUG_SELECT")) + + +def _select_debug(message: str) -> None: + if not _ACLI_DEBUG_SELECT: + return + with open("/tmp/acli_select_debug.log", "a", encoding="utf-8") as log_file: + log_file.write(f"{time.monotonic():.3f} {message}\n") + + # Stream flush: time window + line-count threshold (the threshold only # guards against over-frequent flushes on bursty bulk output) _STREAM_FLUSH_INTERVAL = 0.8 if _IS_JEDITERM else 0.3 @@ -42,6 +54,13 @@ # Wheel batching window: full repaints are costly on JediTerm; trade # frame rate for stability _WHEEL_FLUSH_INTERVAL = 0.12 if _IS_JEDITERM else 0.03 +# Trackpad gestures decelerate: the tail of a wheel gesture arrives as +# isolated arrows 20-200ms after the last burst key, indistinguishable +# from real keypresses by timing alone. Within this window after the +# last wheel-classified key, deferred arrows are absorbed as scrolls +# instead of history/completion (which visibly raced the input box and +# popup while scrolling). +_WHEEL_TAIL_WINDOW = 0.35 from rich.cells import cell_len # noqa: E402 from rich.console import Console # noqa: E402 @@ -58,12 +77,7 @@ from textual.screen import Screen # noqa: E402 from textual.selection import SelectEnd, Selection # noqa: E402 from textual.strip import Strip # noqa: E402 -from textual.widgets import ( # noqa: E402 - OptionList, - RichLog, - Static, - TextArea, -) +from textual.widgets import OptionList, RichLog, Static, TextArea # noqa: E402 from textual.widgets.option_list import Option # noqa: E402 from dashscope.acli.commands import ( # noqa: E402 @@ -179,6 +193,24 @@ class OutputLog(RichLog): widget-level select-all that copies nothing. """ + # Explicit follow state: writes keep the view pinned to the bottom only + # while following. An upward scroll (mouse wheel, PyCharm's arrow-key + # wheel queue, page keys) stops following; a downward scroll re-engages + # it once it lands back at the bottom; scroll_end() (command submit, + # confirmation prompts) always re-engages. A position band or a + # time-based suppression both misbehave at the bottom while streaming: + # the band yanks the view on fine 1-line adjustments and a timeout + # re-arms follow while the user is still reading. + _follow_output: bool = True + + # A read-only view must never take keyboard focus: on PyCharm the + # trackpad wheel arrives as arrow keys, and a focused OutputLog would + # route them to ScrollView's *animated* key-scroll bindings — bypassing + # CommandInput's burst classifier and tearing on JediTerm. Keyboard + # scrolling is handled globally by CommandInput (pageup/shift+arrows); + # mouse wheel and drag selection need no focus. + can_focus = False + def write( self, content, @@ -189,16 +221,38 @@ def write( animate: bool = False, ): if scroll_end is None and self.auto_scroll: - # Sticky follow: only scroll to the end when already at the bottom. - scroll_end = self.scroll_offset.y >= max(self.max_scroll_y - 1, 0) - return super().write( + scroll_end = self._follow_output + if not self._size_known: + return super().write( + content, + width=width, + expand=expand, + shrink=shrink, + scroll_end=scroll_end, + animate=animate, + ) + result = super().write( content, width=width, expand=expand, shrink=shrink, - scroll_end=scroll_end, + scroll_end=False, animate=animate, ) + if scroll_end: + # Follow with a fire-time check rather than RichLog's queued + # scroll_end: a wheel-up landing between the write and the + # refresh must win over the write's queued scroll. + self.call_after_refresh(self._scroll_to_follow) + return result + + def _scroll_to_follow(self) -> None: + if self._follow_output: + self.scroll_to(y=self.max_scroll_y, animate=False) + + def scroll_end(self, *args, **kwargs): + self._follow_output = True + return super().scroll_end(*args, **kwargs) def render_line(self, y: int) -> Strip: scroll_x, scroll_y = self.scroll_offset @@ -281,11 +335,29 @@ def get_selection(self, selection: Selection) -> tuple[str, str] | None: return None def _on_mouse_scroll_down(self, event: events.MouseScrollDown) -> None: + _select_debug( + f"wheel down scroll_y={self.scroll_offset.y}/{self.max_scroll_y}", + ) + # The pump dispatches this event to every MRO class defining the + # handler; stop() doesn't suppress that, only prevent_default() + # does — without it each wheel notch scrolls twice. + event.prevent_default() super()._on_mouse_scroll_down(event) + # Pointer scrolls apply synchronously (animate=False), so the + # landed position is readable here: re-engage follow only when the + # scroll actually reached the bottom. + if self.scroll_offset.y >= self.max_scroll_y: + self._follow_output = True self._extend_selection_after_wheel(event) def _on_mouse_scroll_up(self, event: events.MouseScrollUp) -> None: + _select_debug( + f"wheel up scroll_y={self.scroll_offset.y}/{self.max_scroll_y}", + ) + event.prevent_default() super()._on_mouse_scroll_up(event) + if self.scroll_offset.y < self.max_scroll_y: + self._follow_output = False self._extend_selection_after_wheel(event) def _extend_selection_after_wheel(self, event: events.MouseEvent) -> None: @@ -331,39 +403,121 @@ class AcliScreen(Screen): """ _auto_scroll_pointer: Offset | None = None + _auto_scroll_target: Any = None # widget our auto-scroll timer scrolls _select_state: Any # textual Screen internal + def _start_auto_scroll( + self, + widget, + direction, + speed: float = 1.0, + ) -> None: + # super() calls _stop_auto_scroll() first (clearing the target), so + # record the target afterwards. + super()._start_auto_scroll(widget, direction, speed) + self._auto_scroll_target = widget + _select_debug( + f"arm target={getattr(widget, 'id', widget)} " + f"direction={direction} speed={speed:.2f}", + ) + + def _stop_auto_scroll(self) -> None: + if self._auto_select_scroll_timer is not None: + target = self._auto_scroll_target + _select_debug( + f"stop (was target={getattr(target, 'id', target)})", + ) + self._auto_scroll_target = None + super()._stop_auto_scroll() + def _forward_event(self, event) -> None: + if ( + isinstance(event, events.MouseDown) + and self.app.mouse_captured is not None + ): + # A MouseUp that never reaches the captor (released outside the + # window, intercepted by a modal, ...) leaks the capture: every + # later MouseDown skips selection setup and routes to the stale + # captor, so a drag can never start a new selection and copy + # keeps serving the old one. A fresh MouseDown always begins a + # new gesture — drop the stale capture. + _select_debug( + f"down: dropping stale capture {self.app.mouse_captured}", + ) + self.app.capture_mouse(None) super()._forward_event(event) + if isinstance(event, events.MouseDown): + _select_debug( + f"down y={event.pointer_screen_y} " + f"selecting={self._selecting} " + f"state={self._select_state is not None}", + ) + # Anchor the pointer stash to the new drag: a wheel flush firing + # before this drag's first MouseMove must not extend the + # selection toward the previous drag's parked position. + self._auto_scroll_pointer = Offset( + int(event.pointer_screen_x), + int(event.pointer_screen_y), + ) + return + if isinstance(event, events.MouseUp): + _select_debug(f"up y={event.pointer_screen_y}") + self._auto_scroll_pointer = None + return if not (isinstance(event, events.MouseMove) and self._selecting): return self._auto_scroll_pointer = Offset( int(event.pointer_screen_x), int(event.pointer_screen_y), ) - # When dragging to the top/bottom screen edge, the pointer may land - # on a non-scrollable widget (the fixed input box below) or a no-hit - # area (hit-test on the output area's padding rows returns None) — - # Textual then stops the auto-scroll timer and the selection freezes - # within one screen (always reproducible in iTerm2 when a drag past - # the window edge is clamped to the first/last row). Fall back to - # scrolling #output directly here. - if self._auto_select_scroll_timer is not None: + # Edge auto-scroll must scroll the widget being selected, not the + # widget under the pointer. Below the output area (spinner, input + # box, hint rows) Textual's ancestor walk either finds no scrollable + # widget and stops the timer, or arms it for that widget — the + # CommandInput is a TextArea, so it is *always* scrollable from + # Textual's viewpoint even when the draft fits on one line, and the + # timer then ticks against an input whose scroll offset is clamped + # at 0 while the output never moves (this is exactly what a drag + # into the prompt area hits in iTerm2). Whenever the drag started + # in #output and the pointer sits in an edge zone, make sure the + # armed auto-scroll targets #output. + state = self._select_state + if state is None: return try: output = self.query_one("#output") except Exception: return + start_widget = state.start.content_widget or state.start.container + if start_widget is not output: + # Selecting inside another widget (e.g. the input draft) — + # leave Textual's native auto-scroll alone. + return lines = self.app.SELECT_AUTO_SCROLL_LINES y = event.pointer_screen_y - if y < lines and output.scroll_y > 0: - self._start_auto_scroll(output, -1, (lines - y) / lines) - elif ( - y >= self.size.height - lines - and output.scroll_y < output.max_scroll_y - ): - speed = (y - (self.size.height - lines) + 1) / lines - self._start_auto_scroll(output, +1, speed) + if y < output.region.y + lines: + direction = -1 + speed = min((output.region.y + lines - y) / lines, 1.0) + can_scroll = output.scroll_y > 0 + elif y >= output.region.bottom: + direction = +1 + speed = min((y - output.region.bottom + 1) / lines, 1.0) + can_scroll = output.scroll_y < output.max_scroll_y + else: + return + target = self._auto_scroll_target + _select_debug( + f"move y={y} dy={event.delta_y} " + f"zone={'up' if direction < 0 else 'down'} " + f"target={getattr(target, 'id', target)} " + f"scroll_y={output.scroll_y:.0f}/{output.max_scroll_y}", + ) + if self._auto_scroll_target is output: + # Already scrolling the right widget (armed here or natively). + return + self._stop_auto_scroll() + if can_scroll: + self._start_auto_scroll(output, direction, speed) def extend_selection_to(self, pointer: Offset) -> None: """Move an in-progress selection's end to the content under pointer.""" @@ -713,6 +867,7 @@ def __init__(self, history_path: Path | None = None, **kwargs): self._prev_arrow_ts: float = 0.0 self._wheel_pending: int = 0 self._wheel_flush_timer = None + self._last_wheel_ts: float = 0.0 self._arrow_pending_key: str = "" self._arrow_timer = None # Password masking: real chars live in _password_real, the widget @@ -788,6 +943,20 @@ def _on_key(self, event) -> None: if output is not None: event.prevent_default() event.stop() + if isinstance(output, OutputLog): + # The scroll below is deferred, so decide follow from + # the computed landing position, not the live offset. + if event.key in ("pageup", "shift+up"): + output._follow_output = False + else: + delta = ( + output.scrollable_content_region.height + if event.key == "pagedown" + else 1 + ) + landed = output.scroll_offset.y + delta + if landed >= output.max_scroll_y: + output._follow_output = True if event.key == "pageup": output.scroll_page_up(animate=False) elif event.key == "pagedown": @@ -909,6 +1078,7 @@ def _on_key(self, event) -> None: def _queue_wheel_scroll(self, key: str) -> None: self._wheel_pending += 1 if key == "down" else -1 + self._last_wheel_ts = time.monotonic() if self._wheel_flush_timer is None: self._wheel_flush_timer = self.set_timer( _WHEEL_FLUSH_INTERVAL, @@ -922,11 +1092,15 @@ def _flush_wheel_scroll(self) -> None: if not lines: return try: - output = self.app.query_one("#output") + output = self.app.query_one("#output", OutputLog) screen = self.app.screen except Exception: return + if lines < 0: + output._follow_output = False output.scroll_to(y=output.scroll_offset.y + lines, animate=False) + if lines > 0 and output.scroll_offset.y >= output.max_scroll_y: + output._follow_output = True # Wheel scrolling mid-drag (PyCharm wheel = arrow keys) must also # keep the selection following the pointer, or the selection cannot # grow past one screen @@ -944,6 +1118,13 @@ def _apply_deferred_arrow(self) -> None: self._arrow_timer = None if not key or self.password_mode: return + # A decelerating trackpad gesture keeps emitting isolated arrows + # past the 20ms burst window; while a wheel gesture is (or was + # just) active, these stragglers are scrolls, not history/completion + # keys — routing them there visibly raced the input box and popup. + if time.monotonic() - self._last_wheel_ts < _WHEEL_TAIL_WINDOW: + self._queue_wheel_scroll(key) + return popup = None try: popup = self.app.query_one("#completion-popup", CompletionPopup) @@ -1327,6 +1508,13 @@ def action_copy_selection(self) -> bool: tool = copy_to_clipboard(text) n_lines = len(text.splitlines()) + # Clear the selection highlight BEFORE notify: a failing toast must + # never leave the highlight stuck (clearing also prevents the + # terminal from overwriting the clipboard with visible-screen + # content on Cmd+C). The explicit refresh forces the cleared frame + # out immediately instead of relying on the async selections watcher. + self.screen.clear_selection() + self.screen.refresh() if tool: self.notify(f"Copied {n_lines} lines to clipboard", timeout=2) else: @@ -1339,9 +1527,6 @@ def action_copy_selection(self) -> bool: "does not support it)", timeout=3, ) - # Clear the selection highlight: prevents the terminal from - # overwriting the clipboard with visible-screen content on Cmd+C - self.screen.clear_selection() return True def action_smart_quit(self) -> None: @@ -1575,16 +1760,16 @@ def _write_output(self, content) -> None: def on_text_selected(self, event: events.TextSelected) -> None: # Do not write the clipboard automatically on mouse-up — that would - # clobber the user's clipboard content. Only hint at Cmd+C for an - # explicit copy (the super+c binding writes the full selection, + # clobber the user's clipboard content. Only hint at Ctrl+C for an + # explicit copy (the smart_quit binding writes the full selection, # including the scrolled-off part, to the clipboard). text = self.screen.get_selected_text() if not text or not text.strip(): return n_lines = len(text.splitlines()) self.notify( - f"Selected {n_lines} lines: Cmd+C to copy (Ctrl+C on " - "PyCharm-style terminals); Ctrl+Q to quote into input box", + f"Selected {n_lines} lines: Ctrl+C to copy; " + "Ctrl+Q to quote into input box", timeout=3, ) @@ -1793,9 +1978,9 @@ def _render_banner(self) -> None: "Ctrl+C cancel/quit [/dim]", ) info_lines.append( - "[dim]Output: wheel to scroll; drag to select, then Cmd+C " - "to copy (Ctrl+C on PyCharm-style terminals); Ctrl+Q to " - "quote the selection into the input box[/dim]", + "[dim]Output: wheel to scroll; drag to select, then Ctrl+C " + "to copy; Ctrl+Q to quote the selection into the input box" + "[/dim]", ) panel_border = ( @@ -2822,4 +3007,6 @@ def run_tui(config, agent, input_history_path: Path | None = None): # screen (including the input line) appears to move. Set tui_mouse = # false for terminal-native selection; bypass capture with Alt/Option # drag to copy. - app.run(mouse=getattr(config, "tui_mouse", True)) + mouse = getattr(config, "tui_mouse", True) + _select_debug(f"run_tui start mouse={mouse} pid={os.getpid()}") + app.run(mouse=mouse) diff --git a/dashscope/version.py b/dashscope/version.py index 1925367..e8bac9f 100644 --- a/dashscope/version.py +++ b/dashscope/version.py @@ -1,4 +1,4 @@ # -*- coding: utf-8 -*- # Copyright (c) Alibaba, Inc. and its affiliates. -__version__ = "1.27.0" +__version__ = "1.27.1"