From 17ec8361634445fbef32a745b96b26858b795e43 Mon Sep 17 00:00:00 2001 From: Matlan Date: Fri, 16 Oct 2026 15:05:25 +0200 Subject: [PATCH 1/5] Overhaul CLI coder terminal UI/UX and interactive approval workflow --- CHANGELOG.md | 16 + localm/plugins/coder/agent/context.py | 108 ++++++- localm/plugins/coder/agent/execution.py | 26 +- localm/plugins/coder/agent/loop.py | 62 +++- localm/plugins/coder/cli/_main.py | 18 +- localm/plugins/coder/cli/repl.py | 115 ++++++- localm/plugins/coder/display.py | 378 +++++++++++++++++++++--- tests/plugins/coder/test_coder_ui_ux.py | 260 ++++++++++++++++ 8 files changed, 892 insertions(+), 91 deletions(-) create mode 100644 tests/plugins/coder/test_coder_ui_ux.py diff --git a/CHANGELOG.md b/CHANGELOG.md index ab08b7975..3b183751c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,22 @@ permanent public record of what shipped and are never rewritten; the in-progress ## [Unreleased] +### Added +- **CLI coder terminal UI and interactive approval upgrades.** `localm coder` now displays a + scrollback-safe inline spinner while waiting for the first token or running tools, shows live + line- and character-count progress when the model streams large file writes or patches inside + `` blocks, renders `set_todos` as a color-coded terminal checklist, supports inline + `@path` and `@path:start-end` context mentions in prompts, coalesces multi-line pastes in the + REPL, and expands write/edit/bash approval prompts to `[y/N/a=always/f=full]` (or + `[y/N/a=always]` for commands) with optional rejection feedback routed back to the agent. + +### Fixed +- **CLI coder console output during file creation, streaming, and tool failures.** New-file + previews now cap to a compact preview instead of dumping entire files line by line, streamed + assistant responses no longer print a second time as a Markdown panel after completion, and + failed tool calls now print a compact error excerpt in normal mode instead of hiding the + failure behind `--verbose`. + ## [0.3.0] - 2026-10-14 ### Added diff --git a/localm/plugins/coder/agent/context.py b/localm/plugins/coder/agent/context.py index 67d1d72df..9da2d160c 100644 --- a/localm/plugins/coder/agent/context.py +++ b/localm/plugins/coder/agent/context.py @@ -6,12 +6,14 @@ from __future__ import annotations import json +import re from typing import Optional import localm.plugins.coder.agent as _agent from ..display import ( - print_assistant_label, print_info, print_reasoning_token, print_status, - print_streaming_done, print_streaming_token, print_thinking, + ActivitySpinner, console, print_assistant_label, print_info, + print_reasoning_token, print_status, print_streaming_done, + print_streaming_token, print_thinking, ) from ..parser import _EXPLICIT_FENCE_LANGS, _try_parse_body from .constants import _COMPACT_AUTO_RATIO, _COMPACT_WARN_RATIO, _DEFAULT_CTX_TOKENS @@ -19,6 +21,11 @@ _JSON_WS = " \t\r\n" +_TOOL_NAME_PROGRESS_RE = re.compile(r'"name"\s*:\s*"([A-Za-z0-9_]+)"') +_TOOL_TARGET_PROGRESS_RE = re.compile( + r'"(?:path|command|pattern|url|task)"\s*:\s*"([^"\\]{1,48})' +) + class _TerminalReply: """The interactive terminal's view of one streamed reply: each new status @@ -30,26 +37,60 @@ def __init__(self, name: str) -> None: self._name = name self._labelled = False self._last_status = None + self._in_reasoning = False + self._reasoning_ended_nl = False + self._at_line_start = True + self.printed_tokens = False + self._spinner = ActivitySpinner("Thinking...") + self._spinner.start() def _label(self) -> None: if not self._labelled: + self._spinner.stop() self._labelled = True print_assistant_label(self._name) + self._at_line_start = False def status(self, text: str, code=None) -> None: if self._labelled or not text or text == self._last_status: return self._last_status = text + self._spinner.stop() print_status(text) + self._spinner.start(text) def token(self, piece: str) -> None: + self._spinner.stop() + if self._in_reasoning: + self._in_reasoning = False + if not self._reasoning_ended_nl and not piece.startswith("\n"): + console.print() self._label() + self.printed_tokens = True print_streaming_token(piece) + if piece: + self._at_line_start = piece.endswith("\n") def reasoning(self, piece: str) -> None: + self._spinner.stop() self._label() + self._in_reasoning = True + if piece: + self._reasoning_ended_nl = piece.endswith("\n") + self._at_line_start = self._reasoning_ended_nl print_reasoning_token(piece) + def tool_progress(self, label: str, detail: str = "") -> None: + if not self._at_line_start: + console.print() + self._at_line_start = True + if self._spinner._thread is None or not self._spinner._thread.is_alive(): + self._spinner.start(label) + self._spinner.update(label, detail) + + def finish(self) -> None: + self._spinner.stop() + class _ProgressLines: """Each new status of one model call, as a progress line through @@ -515,8 +556,13 @@ def _pass(earlier, excerpt) -> str: except Exception: return raw - summary, failure = summarise_in_passes( - groups, _pass, reply_tokens=reply_tokens, measure=measure) + if getattr(self, "on_event", None) is None and getattr(self, "progress_sink", None) is None: + with ActivitySpinner("Compacting session history..."): + summary, failure = summarise_in_passes( + groups, _pass, reply_tokens=reply_tokens, measure=measure) + else: + summary, failure = summarise_in_passes( + groups, _pass, reply_tokens=reply_tokens, measure=measure) if summary: body = compose("[Session summary]\n", summary) else: @@ -944,7 +990,8 @@ def _llm_kwargs(self) -> dict: return kw def _stream_and_record(self, messages: list[dict], *, on_token, on_reasoning, - on_interrupt=None, on_status=None) -> str: + on_interrupt=None, on_status=None, + on_tool_progress=None) -> str: """ Consume backend.chat_stream, hiding tool-call blocks from *on_token*, routing reasoning deltas to *on_reasoning*, honouring a mid-stream stop @@ -973,13 +1020,37 @@ def _capture_reasoning(piece: str) -> None: # hider hides are exactly the set parse_tool_calls will later execute. tool_names = set(_agent.TOOL_REGISTRY) - self.disabled_tools + raw_stream = self.backend.chat_stream( + messages, on_reasoning=_capture_reasoning, **status_kw, + **self._llm_kwargs()) + + def _tapped_stream(): + raw_seen = "" + for chunk in raw_stream: + raw_seen += chunk + yield chunk + if on_tool_progress is not None and len(raw_seen) - len(full) >= 12: + pending = raw_seen[len(full):] + if any(m in pending for m in ("= 2: + details.append(f"{lines} lines") + on_tool_progress(lbl, " Β· ".join(details)) + try: # _llm_kwargs, not raw gen_kwargs: every dispatch branch gets the lazy # tool-call grammar through this shared helper. for piece, hidden in self._stream_hiding_tool_calls( - self.backend.chat_stream( - messages, on_reasoning=_capture_reasoning, **status_kw, - **self._llm_kwargs()), + _tapped_stream(), tool_names=tool_names, ): full += piece @@ -1004,6 +1075,7 @@ def _capture_reasoning(piece: str) -> None: def _call_llm(self, messages: list[dict], interactive: bool) -> str: from ..backends.http import CoderAuthError, CoderServerError first_attempt = True + self._streamed_to_terminal = False while True: try: if self.on_event is not None: @@ -1028,16 +1100,22 @@ def _call_llm(self, messages: list[dict], interactive: bool) -> str: def _on_interrupt() -> None: nonlocal interrupted interrupted = True + terminal.finish() print_streaming_done() print_info("(interrupted)") - full = self._stream_and_record( - messages, - on_token=terminal.token, - on_reasoning=terminal.reasoning, - on_interrupt=_on_interrupt, - on_status=terminal.status, - ) + try: + full = self._stream_and_record( + messages, + on_token=terminal.token, + on_reasoning=terminal.reasoning, + on_interrupt=_on_interrupt, + on_status=terminal.status, + on_tool_progress=terminal.tool_progress, + ) + finally: + terminal.finish() + self._streamed_to_terminal = terminal.printed_tokens if not interrupted: print_streaming_done() return full diff --git a/localm/plugins/coder/agent/execution.py b/localm/plugins/coder/agent/execution.py index 76f846559..1cae121df 100644 --- a/localm/plugins/coder/agent/execution.py +++ b/localm/plugins/coder/agent/execution.py @@ -16,7 +16,8 @@ import localm.plugins.coder.agent as _agent from ..display import ( - console, print_progress_tool_call, print_tool_call, print_tool_error, + ActivitySpinner, console, consume_confirm_always, consume_reject_reason, + print_info, print_progress_tool_call, print_tool_call, print_tool_error, print_tool_result, tool_call_text, ) from ..diffutil import ( @@ -584,6 +585,9 @@ def _execute_tool(self, call: ToolCall, interactive: bool) -> ToolResult: agent=self._confirm_agent_label()) elif interactive: approved = self._confirm_tool(call) + if approved and consume_confirm_always(): + self.auto_approve = True + print_info(" βœ“ Auto-approve enabled for the rest of this session.") else: # Fail CLOSED: this tool requires confirmation and there is no way # to obtain it (non-interactive run, no approval handler), so deny @@ -615,11 +619,16 @@ def _execute_tool(self, call: ToolCall, interactive: bool) -> ToolResult: summary="denied: confirmation required, none available") return result if not approved: - result = ToolResult.error("Rejected by user.") + reject_reason = consume_reject_reason() if interactive else "" + err_msg = ( + f"Rejected by user: {reject_reason}" + if reject_reason else "Rejected by user." + ) + result = ToolResult.error(err_msg) if interactive: print_tool_result(call.name, result, verbose=False) self._emit("tool_result", tool=call.name, ok=False, - summary="rejected by user") + summary=err_msg.lower()) return result # Snapshot file content before undoable writes so /undo can restore it and @@ -695,7 +704,11 @@ def _execute_tool(self, call: ToolCall, interactive: bool) -> ToolResult: # how long the tool took - the number the GUI shows next to the card. t_start = time.monotonic() try: - result = tool_def.fn(self.cwd, **args) + if interactive and call.name not in _PARENT_AGENT_TOOLS: + with ActivitySpinner(f"Running {call.name}..."): + result = tool_def.fn(self.cwd, **args) + else: + result = tool_def.fn(self.cwd, **args) except TypeError as e: result = ToolResult.error(f"Bad arguments for {call.name}: {e}") except Exception as e: @@ -712,7 +725,10 @@ def _execute_tool(self, call: ToolCall, interactive: bool) -> ToolResult: self._audit.tool_result(call.name, result.ok, result.summary) if interactive: - print_tool_result(call.name, result, verbose=self.verbose) + print_tool_result( + call.name, result, verbose=self.verbose, + elapsed=duration_s, args=call.args, + ) self._emit("tool_result", tool=call.name, ok=result.ok, summary=result.summary, output=result.output[:4000], duration_s=duration_s) diff --git a/localm/plugins/coder/agent/loop.py b/localm/plugins/coder/agent/loop.py index 3316a63d4..9ca498d14 100644 --- a/localm/plugins/coder/agent/loop.py +++ b/localm/plugins/coder/agent/loop.py @@ -13,9 +13,9 @@ import localm.plugins.coder.agent as _agent from ..display import ( - console, print_assistant_response, print_info, print_success, - print_tool_call, print_tool_error, print_tool_result, print_turn_divider, - safe_markup, + ActivitySpinner, console, print_assistant_response, print_info, + print_success, print_task_footer, print_tool_call, print_tool_error, + print_tool_result, print_turn_divider, safe_markup, ) from ..parser import looks_like_tool_attempt, split_response, strip_orphan_closers from ..tools import ToolResult @@ -387,13 +387,9 @@ def _loop(self, interactive: bool) -> str: verify_retries=0, verify_settled=False, verify_checked_at=self._write_total(), partial_notice_count=0, partial_notice_cap_announced=False, - # Zero-tool-call escalation: how many rungs of the - # ladder this task has used, and whether the model - # has produced any call yet. nocall_escalation=0, nocall_declined=False, tool_calls_made=0, writes_at_start=self._write_total(), - # The one-shot re-prompt for code that is not in - # the file the reply names (unfounded_code). + start_turns=start_turns, unfounded_checked=False) try: @@ -548,9 +544,9 @@ def _loop(self, interactive: bool) -> str: # the model can produce a CALL at all, not whether it succeeded. st.tool_calls_made += len(calls) - # Show the non-tool-call text parts first + # Show the non-tool-call text parts when not already streamed segments = split_response(response, calls) - if interactive: + if interactive and not getattr(self, "_streamed_to_terminal", False): for seg in segments: if isinstance(seg, str) and seg.strip(): console.print(safe_markup(seg.strip())) @@ -880,7 +876,22 @@ def _handle_no_tool_calls(self, response, interactive, st) -> tuple[bool, str]: footer = self._grounding_footer() sensitive = self._sensitive_changes_notice() final_text = response + enforcement + unverified + (footer or "") + sensitive - if not interactive and self.on_event is None: + if interactive and self.on_event is None: + print_warning = _agent.print_warning + if enforcement.strip(): + print_warning(enforcement.strip()) + if unverified.strip(): + print_warning(unverified.strip()) + if sensitive.strip(): + print_warning(sensitive.strip()) + if st.tool_calls_made > 0: + print_task_footer( + turns_used=max(1, self._turns - getattr(st, "start_turns", self._turns - 1)), + files_changed=[c["path"] for c in self.changed_files()], + verify_status=self._last_verify_state, + ctx_ratio=self._fill_ratio(), + ) + elif not interactive and self.on_event is None: print_assistant_response(final_text, name=self.name) self._add_assistant(response) return (True, final_text) @@ -1083,7 +1094,11 @@ def _run_pre_done_review(self, diff: str) -> str: except Exception as e: from localm.debuglog import logger logger.debug("sensitive-changes reviewer note skipped: %r", e) # See test_classify_sensitive_changes_tolerates_malformed_entries. - result = self._reviewer.review(diff, self._review_task, sensitive_note=sensitive_note) + if getattr(self, "_interactive", False): + with ActivitySpinner("Running pre-done code review..."): + result = self._reviewer.review(diff, self._review_task, sensitive_note=sensitive_note) + else: + result = self._reviewer.review(diff, self._review_task, sensitive_note=sensitive_note) warning = self._reviewer.failure_warning(result) if warning: print_warning(warning) @@ -1134,7 +1149,10 @@ def _run_verify_gate(self, response, interactive, st): self._emit("info", text=f"verification: running `{label}`") if interactive: print_info(f"(verification: running `{label}`)") - outcome = _verify.run_verify(cmd, self.cwd) + with ActivitySpinner(f"Verifying with `{label}`..."): + outcome = _verify.run_verify(cmd, self.cwd) + else: + outcome = _verify.run_verify(cmd, self.cwd) code, output = outcome # Whether the command STARTED is carried on the outcome, never inferred # from the exit code. @@ -1351,6 +1369,11 @@ def _execute_tools(self, calls: list, interactive: bool) -> list[str]: pool.submit(self._execute_tool, call, False): (i, call) for i, call in enumerate(group) } + batch_spinner = ActivitySpinner( + f"Running {len(group)} tools in parallel..." + ) + if interactive: + batch_spinner.start() try: for fut in as_completed(futures, timeout=self._PARALLEL_BATCH_TIMEOUT_S): @@ -1359,12 +1382,19 @@ def _execute_tools(self, calls: list, interactive: bool) -> list[str]: result = fut.result() except Exception as exc: result = ToolResult.error(f"Parallel execution error: {exc}") - # Print results in original order when interactive ordered[i] = self._result_block(call, result) if interactive: + batch_spinner.stop() print_tool_call(call.name, call.args) - print_tool_result(call.name, result, verbose=self.verbose) + print_tool_result( + call.name, result, verbose=self.verbose, args=call.args + ) + rem = len(group) - len(ordered) + if rem > 0: + batch_spinner.start(f"Running {rem} remaining tool(s)...") except TimeoutError: + if interactive: + batch_spinner.stop() for fut, (i, call) in futures.items(): if i not in ordered: fut.cancel() @@ -1382,6 +1412,8 @@ def _execute_tools(self, calls: list, interactive: bool) -> list[str]: if interactive: print_tool_error(call.name, result.output) finally: + if interactive: + batch_spinner.stop() pool.shutdown(wait=False) for i in range(len(group)): result_blocks.append(ordered[i]) diff --git a/localm/plugins/coder/cli/_main.py b/localm/plugins/coder/cli/_main.py index 89e6257e1..6916d0f85 100644 --- a/localm/plugins/coder/cli/_main.py +++ b/localm/plugins/coder/cli/_main.py @@ -883,15 +883,17 @@ def _child_log_tail(limit: int = 2000) -> str: # flushes anything buffered before the early-exit read # below. child_log.close() + from ..display import ActivitySpinner _tgt = None - for _ in range(40): - time.sleep(0.5) - if proc.poll() is not None: - break - _root = instances.resolve_root_dir(start=str(work_dir)) - _tgt = instances.attach_target(home_dir(), _root) - if _tgt: - break + with ActivitySpinner("Waiting for localm server to start..."): + for _ in range(40): + time.sleep(0.5) + if proc.poll() is not None: + break + _root = instances.resolve_root_dir(start=str(work_dir)) + _tgt = instances.attach_target(home_dir(), _root) + if _tgt: + break if _tgt: attach_key = api_key if not _api_key_explicit: diff --git a/localm/plugins/coder/cli/repl.py b/localm/plugins/coder/cli/repl.py index 3e8a42473..fec89d055 100644 --- a/localm/plugins/coder/cli/repl.py +++ b/localm/plugins/coder/cli/repl.py @@ -24,12 +24,34 @@ from ..review_guard import classify_sensitive_changes, render_warning from .goal import _run_goal_loop +import re +import sys + +_AT_MENTION_RE = re.compile( + r"(?:^|(?<=\s))@([A-Za-z0-9_./\\-]+(?::\d+(?:-\d+)?)?)(?=\s|$|[.,;!?])" +) +_MAX_AT_FILE_CHARS = 24_000 + + +def _stdin_has_buffered_data() -> bool: + """Return True when interactive stdin already has buffered pasted bytes waiting.""" + try: + if not getattr(sys.stdin, "isatty", lambda: False)(): + return False + if os.name == "nt": + import msvcrt + return bool(msvcrt.kbhit()) + import select + ready, _, _ = select.select([sys.stdin], [], [], 0.0) + return bool(ready) + except Exception: + return False + + def _read_multiline() -> str: """ - Read one user message, supporting backslash line continuation. - - End a line with \\ to keep typing on the next line. The backslash is - stripped and the lines are joined with a newline before being sent. + Read one user message, supporting backslash line continuation and + coalescing multi-line terminal pastes into a single message. """ lines: list[str] = [] first = True @@ -44,10 +66,78 @@ def _read_multiline() -> str: first = False else: lines.append(line) + pasted = 0 + while _stdin_has_buffered_data(): + extra = sys.stdin.readline() + if not extra: + break + lines.append(extra.rstrip("\r\n")) + pasted += 1 + if pasted >= 2: + print_info(f" (pasted {pasted + 1} lines)") break return "\n".join(lines) +def _expand_at_mentions(user_input: str, cwd: Path) -> str: + """Attach contents of existing workspace files referenced as @path[:start-end].""" + if "@" not in user_input: + return user_input + from ..tools.files import _confine + + attachments: list[str] = [] + seen: set[str] = set() + for match in _AT_MENTION_RE.finditer(user_input): + spec = match.group(1) + if spec in seen: + continue + seen.add(spec) + path_part = spec + line_range: tuple[int, int] | None = None + if ":" in spec: + base, rng = spec.rsplit(":", 1) + if re.fullmatch(r"\d+(?:-\d+)?", rng): + path_part = base + if "-" in rng: + s_str, e_str = rng.split("-", 1) + line_range = (max(1, int(s_str)), max(1, int(e_str))) + else: + ln = max(1, int(rng)) + line_range = (ln, ln) + try: + resolved = _confine(cwd, path_part) + except PermissionError: + continue + if not resolved.is_file(): + continue + try: + raw = resolved.read_text(encoding="utf-8", errors="replace") + except Exception: + continue + rel = resolved.relative_to(cwd) if resolved.is_relative_to(cwd) else resolved + all_lines = raw.splitlines() + if line_range is not None: + start_ln, end_ln = line_range + sliced = all_lines[start_ln - 1:end_ln] + body = "\n".join(sliced) + label = f"{rel}:{start_ln}-{end_ln}" + n_lines = len(sliced) + else: + body = raw + label = str(rel) + n_lines = len(all_lines) + trunc_note = "" + if len(body) > _MAX_AT_FILE_CHARS: + body = body[:_MAX_AT_FILE_CHARS] + "\n... [truncated]" + trunc_note = " (truncated)" + print_info(f" πŸ“Ž Attached {label} ({n_lines} lines{trunc_note})") + attachments.append(f'\n{body}\n') + + if not attachments: + return user_input + return user_input + "\n\n" + "\n\n".join(attachments) + + # Every REPL slash command, for tab completion and /help _SLASH_COMMANDS = ( "/help", "/exit", "/quit", "/clear", "/model", "/mode", "/cwd", "/cd", @@ -85,16 +175,19 @@ def complete(text: str, state: int): except Exception: options = [] else: - # Complete the current token as a path relative to the project + # Complete the current token as a path relative to the project (with or without @ prefix) import glob as _g + at_prefix = "@" if text.startswith("@") else "" + lookup = text[1:] if at_prefix else text try: - matches = _g.glob(str(agent.cwd / (text + "*"))) + matches = _g.glob(str(agent.cwd / (lookup + "*"))) except Exception: matches = [] options = [] for m in matches[:50]: rel = os.path.relpath(m, agent.cwd) - options.append(rel + os.sep if os.path.isdir(m) else rel) + suffix = os.sep if os.path.isdir(m) else "" + options.append(at_prefix + rel + suffix) try: return options[state] except IndexError: @@ -135,7 +228,7 @@ def _warn_goal_run_sensitive_changes(agent: Agent, before: dict) -> None: See test_goal_run_warning_is_scoped_to_that_run_not_the_whole_session.""" try: touched = [f for f in agent.changed_files() - if f["writes"] > before.get(f["path"], 0)] + if f["writes"] > before.get(f["path"], 0)] message = render_warning(classify_sensitive_changes(touched)) if message: print_warning(message) @@ -166,6 +259,8 @@ def _repl(agent: Agent) -> None: break continue + expanded_input = _expand_at_mentions(user_input, agent.cwd) + try: # The saved checkpoint is not cleared before the turn; the turn # itself clears or rewrites it. See @@ -173,12 +268,12 @@ def _repl(agent: Agent) -> None: if agent.goal_cmd is not None: before = {f["path"]: f["writes"] for f in agent.changed_files()} try: - _run_goal_loop(agent, user_input, agent.goal_cmd, + _run_goal_loop(agent, expanded_input, agent.goal_cmd, agent.goal_max_iters, agent.cwd) finally: _warn_goal_run_sensitive_changes(agent, before) else: - agent.chat(user_input) + agent.chat(expanded_input) except KeyboardInterrupt: # _loop has already dealt with the checkpoint; swallow the # interrupt here. diff --git a/localm/plugins/coder/display.py b/localm/plugins/coder/display.py index 8197fb2a1..fcf358d05 100644 --- a/localm/plugins/coder/display.py +++ b/localm/plugins/coder/display.py @@ -8,6 +8,9 @@ from __future__ import annotations import difflib +import re +import threading +import time from pathlib import Path from localm.console import SafeConsole as Console, strip_controls @@ -20,6 +23,137 @@ console = Console(highlight=False) err_console = Console(stderr=True, highlight=False) +_SPINNER_FRAMES = ("β ‹", "β ™", "β Ή", "β Έ", "β Ό", "β ΄", "β ¦", "β §", "β ‡", "⠏") + +_last_diff_preview: tuple[str, str, str, bool] | None = None +_last_confirm_always: bool = False +_last_reject_reason: str = "" + + +def consume_confirm_always() -> bool: + """Return and clear the session auto-approve flag set by the last confirmation prompt.""" + global _last_confirm_always + val = _last_confirm_always + _last_confirm_always = False + return val + + +def consume_reject_reason() -> str: + """Return and clear the optional rejection reason entered at the last confirmation prompt.""" + global _last_reject_reason + val = _last_reject_reason + _last_reject_reason = "" + return val + + +class ActivitySpinner: + """Single-line carriage-return activity indicator using SGR styling only. See test_coder_stream_renderer_streams_without_viewport_control.""" + + def __init__( + self, + label: str = "Thinking...", + *, + console_obj: Console | None = None, + interval: float = 0.1, + ) -> None: + self._console_override = console_obj + self._label = label + self._detail = "" + self._interval = interval + self._stop_evt = threading.Event() + self._lock = threading.Lock() + self._thread: threading.Thread | None = None + self._start_time = 0.0 + self._last_len = 0 + self._frame_idx = 0 + + def _active_console(self) -> Console: + return self._console_override if self._console_override is not None else console + + def _is_interactive_tty(self, c: Console) -> bool: + is_term = getattr(c, "is_terminal", None) + return isinstance(is_term, bool) and is_term + + def start(self, label: str | None = None) -> "ActivitySpinner": + with self._lock: + if label is not None: + self._label = label + self._detail = "" + if self._thread is not None and self._thread.is_alive(): + return self + self._stop_evt.clear() + self._start_time = time.monotonic() + self._frame_idx = 0 + c = self._active_console() + if not self._is_interactive_tty(c): + return self + self._render_frame_locked(c) + self._thread = threading.Thread( + target=self._run, name="coder-spinner", daemon=True + ) + self._thread.start() + return self + + def update(self, label: str | None = None, detail: str | None = None) -> None: + with self._lock: + if label is not None: + self._label = label + if detail is not None: + self._detail = detail + + def _run(self) -> None: + while not self._stop_evt.wait(self._interval): + with self._lock: + if self._stop_evt.is_set(): + break + c = self._active_console() + if self._is_interactive_tty(c): + self._render_frame_locked(c) + + def _render_frame_locked(self, c: Console) -> None: + elapsed = time.monotonic() - self._start_time + frame = _SPINNER_FRAMES[self._frame_idx % len(_SPINNER_FRAMES)] + self._frame_idx += 1 + max_w = max(20, (getattr(c, "width", 80) or 80) - 2) + line = Text(" ") + line.append(f"{frame} ", style="cyan") + line.append(_strip_esc(self._label), style="dim") + if self._detail: + line.append(f" Β· {_strip_esc(self._detail)}", style="dim cyan") + if elapsed >= 0.4: + line.append(f" ({elapsed:.1f}s)", style="dim") + if len(line) > max_w: + line.truncate(max_w) + plain_len = len(line.plain) + if plain_len < self._last_len: + line.append(" " * (self._last_len - plain_len)) + self._last_len = plain_len + c.print(line, end="\r", soft_wrap=True, highlight=False) + + def stop(self) -> None: + self._stop_evt.set() + t = self._thread + if t is not None and t.is_alive() and t is not threading.current_thread(): + t.join(timeout=0.3) + with self._lock: + self._thread = None + c = self._active_console() + if self._last_len > 0 and self._is_interactive_tty(c): + c.print( + Text(" " * self._last_len), + end="\r", + soft_wrap=True, + highlight=False, + ) + self._last_len = 0 + + def __enter__(self) -> "ActivitySpinner": + return self.start() + + def __exit__(self, *exc) -> None: + self.stop() + + def _strip_esc(s: str) -> str: """Remove terminal control characters (ESC and every other C0/C1 control except newline and tab) so externally-controlled text (model output, tool @@ -72,7 +206,7 @@ def print_banner( f"[bold cyan]{agent_name}[/bold cyan] Β· " f"[dim]model:[/dim] [bold]{model}[/bold] Β· " f"[dim]cwd:[/dim] [dim]{cwd}[/dim]{map_info}{mode_info}\n" - "[dim]Type your task, or [bold]/help[/bold] for commands. " + "[dim]Type your task, [bold]@file[/bold] to attach context, or [bold]/help[/bold] for commands. " "Ctrl+C or [bold]/exit[/bold] to quit.[/dim]", border_style="dim cyan", padding=(0, 1), @@ -113,42 +247,119 @@ def _tool_call_line(tool_name: str, args: dict) -> Text: line.append("(") first = True for k, v in args.items(): - if k == "content": # don't show full file contents inline + if k in ("content", "diff"): continue if not first: line.append(", ") first = False line.append(_strip_esc(k), style="cyan") line.append("=") - line.append(_strip_esc(repr(str(v)[:60])), style="dim") - if "content" in args: - # Show content length instead of dumping it - n = len(args["content"].splitlines()) - if not first: - line.append(", ") - line.append("content", style="cyan") - line.append(f"=<{n} lines>", style="dim") + if k in ("old", "new") and isinstance(v, str) and ("\n" in v or len(v) > 48): + n = len(v.splitlines()) or 1 + line.append(f"<{n} lines>", style="dim") + elif k == "edits" and isinstance(v, list): + line.append(f"<{len(v)} edits>", style="dim") + elif k == "items" and isinstance(v, list): + line.append(f"<{len(v)} items>", style="dim") + else: + line.append(_strip_esc(repr(str(v)[:60])), style="dim") + for bulk_key in ("content", "diff"): + if bulk_key in args and isinstance(args[bulk_key], str): + n = len(args[bulk_key].splitlines()) + if not first: + line.append(", ") + first = False + line.append(bulk_key, style="cyan") + line.append(f"=<{n} lines>", style="dim") line.append(")") return line -def print_tool_result(tool_name: str, result, verbose: bool = False) -> None: - """Print the one-line summary, optionally expanding the full output.""" +_XML_WRAPPER_TAG_RE = re.compile( + r"^\s*(?:[^<]*|<(?:stdout|stderr|output)>|)\s*$" +) + + +def _clean_output_lines(raw_output: str) -> list[str]: + """Return non-wrapper lines from tool output for display.""" + lines = [] + for raw_line in raw_output.splitlines(): + if _XML_WRAPPER_TAG_RE.match(raw_line): + continue + lines.append(raw_line) + while lines and not lines[0].strip(): + lines.pop(0) + while lines and not lines[-1].strip(): + lines.pop() + return lines + + +def print_tool_result( + tool_name: str, + result, + verbose: bool = False, + *, + elapsed: float | None = None, + args: dict | None = None, +) -> None: + """Print the one-line summary, todo checklist, error excerpt on failure, or verbose output.""" icon, icon_style = ("βœ“", "green") if result.ok else ("βœ—", "red") - trunc = " (truncated)" if result.truncated else "" + trunc = " (truncated)" if getattr(result, "truncated", False) else "" line = Text(" ") line.append(icon + " ", style=icon_style) - line.append(_strip_esc(result.summary + trunc), style="dim") + line.append(_strip_esc(str(result.summary) + trunc), style="dim") + if elapsed is not None and elapsed >= 0.05: + line.append(f" ({elapsed:.1f}s)", style="dim") console.print(line) - if verbose and result.ok and result.output: - # Collapse tool result behind a divider for readability + if tool_name == "set_todos" and result.ok and isinstance(args, dict): + items = args.get("items") + if isinstance(items, list): + for item in items[:20]: + if not isinstance(item, dict): + continue + status = str(item.get("status", "pending")).lower() + task_text = str(item.get("task", item.get("content", ""))).strip() + if not task_text: + continue + todo_line = Text(" ") + if status in ("done", "completed"): + todo_line.append("[βœ“] ", style="green") + todo_line.append(_strip_esc(task_text), style="dim strike") + elif status == "in_progress": + todo_line.append("[β–Έ] ", style="bold cyan") + todo_line.append(_strip_esc(task_text), style="bold") + else: + todo_line.append("[ ] ", style="dim") + todo_line.append(_strip_esc(task_text), style="dim") + console.print(todo_line) + + raw_output = getattr(result, "output", "") + if not isinstance(raw_output, str) or not raw_output.strip(): + return + + if not result.ok: + cleaned = _clean_output_lines(raw_output) + summary_str = str(result.summary).strip() + if cleaned and not (len(cleaned) == 1 and cleaned[0].strip() == summary_str): + cap = 40 if verbose else 12 + excerpt = cleaned[-cap:] if len(cleaned) > cap else cleaned + if len(cleaned) > cap: + console.print(_sanitized_text( + f" ... ({len(cleaned) - cap} earlier lines omitted)", style="dim" + )) + for out_line in excerpt: + console.print(_sanitized_text(" " + out_line, style="red dim")) + return + + if verbose and result.ok: console.print(_sanitized_text(" " + "─" * 60, style="dim")) - for out_line in result.output.splitlines()[:40]: + out_lines = raw_output.splitlines() + for out_line in out_lines[:40]: console.print(_sanitized_text(" " + out_line, style="dim")) - if len(result.output.splitlines()) > 40: + if len(out_lines) > 40: console.print(_sanitized_text( - " ... (use /verbose to see full output)", style="dim")) + " ... (output capped at 40 lines)", style="dim")) def print_tool_error(tool_name: str, message: str) -> None: @@ -162,7 +373,7 @@ def print_tool_error(tool_name: str, message: str) -> None: # Agent response # --------------------------------------------------------------------------- -def print_thinking(label: str = "Thinking…") -> None: +def print_thinking(label: str = "Thinking...") -> None: console.print(f"\n[dim]{label}[/dim]") @@ -172,12 +383,17 @@ def print_status(text: str) -> None: console.print(_sanitized_text(f" {text}", style="dim")) +def print_reasoning_header() -> None: + """Print a dim header before streaming reasoning tokens.""" + console.print("\n[dim] πŸ’­ Thinking:[/dim]\n ", end="") + + def print_assistant_label(name: str = "Agent") -> None: console.print(f"\n[bold blue]{name}[/bold blue]: ", end="") def print_assistant_response(text: str, name: str = "Agent") -> None: - """Render the final response text. Uses Markdown if it looks like Markdown.""" + """Render the final response text. Uses Markdown if it looks like Markdown.""" text = text.strip() if not text: return @@ -212,7 +428,7 @@ def print_streaming_done() -> None: # --------------------------------------------------------------------------- -# Turn divider +# Turn divider & task summary footer # --------------------------------------------------------------------------- def print_turn_divider(turn: int, total_tokens: int = 0, turn_tokens: int = 0, @@ -230,6 +446,31 @@ def print_turn_divider(turn: int, total_tokens: int = 0, turn_tokens: int = 0, console.print(f"\n[dim]{body} ──────────────────────────────────────[/dim]") +def print_task_footer( + turns_used: int, + files_changed: list[str] | None = None, + verify_status: str | None = None, + ctx_ratio: float | None = None, +) -> None: + """Print a one-line turn summary footer at the end of an interactive task.""" + parts: list[str] = [f"{turns_used} turn{'s' if turns_used != 1 else ''}"] + if files_changed: + n = len(files_changed) + names = ", ".join(safe_markup(f) for f in files_changed[:3]) + if n > 3: + names += f", +{n - 3} more" + parts.append(f"{n} file{'s' if n != 1 else ''} changed ({names})") + if verify_status: + v_col = "green" if verify_status == "passed" else "yellow" + parts.append(f"[{v_col}]verify: {safe_markup(verify_status)}[/{v_col}]") + if ctx_ratio is not None and ctx_ratio > 0: + pct = min(ctx_ratio, 1.0) * 100 + color = "red" if ctx_ratio >= 0.85 else ("yellow" if ctx_ratio >= 0.65 else "dim") + parts.append(f"[{color}]ctx {pct:.0f}%[/{color}]") + body = " Β· ".join(parts) + console.print(f"[dim] ── {body} ──[/dim]") + + # --------------------------------------------------------------------------- # Status messages # --------------------------------------------------------------------------- @@ -260,7 +501,7 @@ def print_success(msg: str) -> None: # --------------------------------------------------------------------------- def print_server_starting(model: str, port: int) -> None: - console.print(f"[dim]Starting localm serve {model} on port {port}…[/dim]") + console.print(f"[dim]Starting localm serve {model} on port {port}...[/dim]") def print_server_ready(port: int) -> None: @@ -282,7 +523,7 @@ def print_diff_preview( old_text: str, new_text: str, path_label: str = "", - max_lines: int = 200, + max_lines: int = 30, ) -> None: """ Print a coloured unified diff between *old_text* and *new_text*. @@ -298,6 +539,7 @@ def print_diff_preview( max_lines: Truncate the displayed diff at this many lines. """ + global _last_diff_preview from_label = f"a/{path_label}" if path_label else "a/current" to_label = f"b/{path_label}" if path_label else "b/proposed" @@ -309,27 +551,77 @@ def print_diff_preview( )) if not diff_lines: + _last_diff_preview = None console.print("[dim] (no changes)[/dim]") return + truncated = len(diff_lines) > max_lines + _last_diff_preview = (old_text, new_text, path_label, truncated) + + added = sum(1 for ln in diff_lines if ln.startswith("+") and not ln.startswith("+++")) + removed = sum(1 for ln in diff_lines if ln.startswith("-") and not ln.startswith("---")) + header = Text("\n ") + if not old_text: + header.append("+ new file ", style="bold green") + header.append(_strip_esc(path_label or "proposed"), style="bold") + header.append(f" ({added} lines)", style="dim") + else: + header.append("Ξ” ", style="bold cyan") + header.append(_strip_esc(path_label or "proposed"), style="bold") + header.append(f" (+{added} -{removed} lines)", style="dim") + diff_text = "".join(diff_lines[:max_lines]) - if len(diff_lines) > max_lines: - diff_text += f"\n... ({len(diff_lines) - max_lines} more lines truncated)\n" + if truncated: + rem = len(diff_lines) - max_lines + diff_text += f"\n... ({rem} more lines truncated - press [f] at prompt for full diff)\n" - console.print() + console.print(header) console.print(Syntax(diff_text, "diff", theme="monokai", line_numbers=False)) +def _parse_confirm_answer(raw: str) -> tuple[str, str]: + """Parse raw confirmation input into (action, reason).""" + s = raw.strip() + lower = s.lower() + if lower in ("y", "yes"): + return "yes", "" + if lower in ("a", "always", "all"): + return "always", "" + if lower in ("f", "full", "d", "diff"): + return "full", "" + if lower in ("n", "no", ""): + return "no", "" + for prefix in ("n:", "no:", "n ", "no "): + if lower.startswith(prefix): + return "no", s[len(prefix):].strip() + return "no", "" + + def confirm_diff(path_label: str) -> bool: - """Prompt 'Apply changes to ? [y/N]' and return True on yes.""" + """Prompt to apply changes to *path_label* with support for [y/N/a/f].""" + global _last_confirm_always, _last_reject_reason + _last_confirm_always = False + _last_reject_reason = "" try: - # path_label is model-chosen text, composed via Text rather than - # embedded in a markup string. - prompt = Text("Apply changes to ", style="yellow") - prompt.append(_strip_esc(path_label), style="bold yellow") - prompt.append("? [y/N] ", style="yellow") - answer = console.input(prompt).strip().lower() - return answer in ("y", "yes") + while True: + has_full = bool(_last_diff_preview and _last_diff_preview[3]) + opts = "[y/N/a=always/f=full] " if has_full else "[y/N/a=always] " + prompt = Text("Apply changes to ", style="yellow") + prompt.append(_strip_esc(path_label), style="bold yellow") + prompt.append(f"? {opts}", style="yellow") + raw = console.input(prompt) + action, reason = _parse_confirm_answer(raw) + if action == "full" and _last_diff_preview is not None: + old_t, new_t, lbl, _ = _last_diff_preview + print_diff_preview(old_t, new_t, path_label=lbl, max_lines=100_000) + continue + if action == "always": + _last_confirm_always = True + return True + if action == "yes": + return True + _last_reject_reason = reason + return False except (KeyboardInterrupt, EOFError): return False @@ -339,10 +631,20 @@ def confirm_diff(path_label: str) -> bool: # --------------------------------------------------------------------------- def confirm(prompt: str) -> bool: + global _last_confirm_always, _last_reject_reason + _last_confirm_always = False + _last_reject_reason = "" try: - line = _sanitized_text(f"{prompt} [y/N] ", style="yellow") - answer = console.input(line).strip().lower() - return answer in ("y", "yes") + line = _sanitized_text(f"{prompt} [y/N/a=always] ", style="yellow") + raw = console.input(line) + action, reason = _parse_confirm_answer(raw) + if action == "always": + _last_confirm_always = True + return True + if action == "yes": + return True + _last_reject_reason = reason + return False except (KeyboardInterrupt, EOFError): return False diff --git a/tests/plugins/coder/test_coder_ui_ux.py b/tests/plugins/coder/test_coder_ui_ux.py new file mode 100644 index 000000000..08d4e74ff --- /dev/null +++ b/tests/plugins/coder/test_coder_ui_ux.py @@ -0,0 +1,260 @@ +# SPDX-License-Identifier: AGPL-3.0-or-later +"""Tests for the CLI coder UI/UX improvements: scrollback-safe spinner, live +hidden tool-call progress, single-print streaming, compact diff previews with +[f]ull/[a]lways/rejection-reason support, failed-tool error excerpts, todo +checklist rendering, and @file context expansion.""" + +from __future__ import annotations + +import io +from pathlib import Path +from unittest.mock import MagicMock, patch + +from localm.console import SafeConsole +from localm.plugins.coder import display +from localm.plugins.coder.cli.repl import _expand_at_mentions, _read_multiline +from localm.plugins.coder.parser import ToolCall +from localm.plugins.coder.tools import ToolResult +from tests.test_cli_stream_scroll import _has_viewport_control + + +def _make_agent(tmp_path: Path, *, auto_approve: bool = False): + from localm.plugins.coder.agent import Agent + backend = MagicMock() + backend.model_id = "test-model" + backend.native_tools = False + backend.supports_grammar = False + backend.last_usage = {} + with patch("localm.plugins.coder.agent.ProjectMap") as MockPM, \ + patch("localm.plugins.coder.agent.make_audit_log"), \ + patch("localm.plugins.coder.agent.load_memory", return_value=""): + MockPM.build.return_value.file_count.return_value = 0 + agent = Agent(backend=backend, cwd=tmp_path, auto_approve=auto_approve) + agent._audit = MagicMock() + return agent + + +class TestActivitySpinner: + def test_spinner_emits_no_viewport_control_sequences(self): + buf = io.StringIO() + tty_console = SafeConsole(force_terminal=True, file=buf, highlight=False, width=80) + spinner = display.ActivitySpinner( + "Thinking...", console_obj=tty_console, interval=0.02 + ) + spinner.start() + spinner.update("Preparing write_file", "src/app.py Β· 42 lines") + spinner._render_frame_locked(tty_console) + spinner.stop() + + out = buf.getvalue() + assert "Thinking..." in out + assert "Preparing write_file" in out + assert "src/app.py" in out + assert "\r" in out + assert not _has_viewport_control(out), repr(out) + + def test_spinner_is_noop_on_non_tty_console(self): + buf = io.StringIO() + non_tty = SafeConsole(force_terminal=False, file=buf, highlight=False) + with display.ActivitySpinner("Thinking...", console_obj=non_tty): + pass + assert buf.getvalue() == "" + + +class TestHiddenToolCallLiveProgress: + def test_hidden_write_file_reports_live_progress_without_leaking_tokens(self, tmp_path): + agent = _make_agent(tmp_path) + py_lines = "\\n".join(f"def fn_{i}(): return {i}" for i in range(40)) + raw_reply = ( + "I will create the module now.\n" + '\n{"name": "write_file", "args": ' + f'{{"path": "src/module.py", "content": "{py_lines}"}}}}\n' + ) + chunks = [raw_reply[i:i + 15] for i in range(0, len(raw_reply), 15)] + + def fake_stream(messages, on_reasoning=None, **kw): + yield from chunks + + agent.backend.chat_stream.side_effect = fake_stream + progress_updates = [] + streamed_tokens = [] + + orig_init = display._SPINNER_FRAMES # keep module loaded + assert orig_init + + from localm.plugins.coder.agent import context + orig_tp = context._TerminalReply.tool_progress + + def spy_tp(self, label, detail=""): + progress_updates.append((label, detail)) + return orig_tp(self, label, detail) + + with patch.object(context._TerminalReply, "tool_progress", spy_tp), \ + patch.object(context, "print_streaming_token", side_effect=streamed_tokens.append), \ + patch.object(context, "print_thinking"), \ + patch.object(context, "print_assistant_label"), \ + patch.object(context, "print_streaming_done"): + agent._call_llm([{"role": "user", "content": "write module"}], interactive=True) + + visible = "".join(streamed_tokens) + assert "I will create the module now." in visible + assert "def fn_0" not in visible + assert any("Preparing write_file" in lbl for lbl, _ in progress_updates) + assert any("src/module.py" in dtl for _, dtl in progress_updates) + + +class TestNoDoublePrintInInteractiveLoop: + def test_streamed_prose_before_tool_call_is_printed_only_once(self, tmp_path, capsys): + agent = _make_agent(tmp_path, auto_approve=True) + (tmp_path / "hello.py").write_text("x = 1\n", encoding="utf-8") + + turn_replies = [ + 'Checking the file first.\n\n{"name": "read_file", "args": {"path": "hello.py"}}\n', + "All done.", + ] + + def fake_stream(messages, on_reasoning=None, **kw): + reply = turn_replies.pop(0) + for i in range(0, len(reply), 12): + yield reply[i:i + 12] + + agent.backend.chat_stream.side_effect = fake_stream + agent.chat("read hello.py") + out = capsys.readouterr().out + assert out.count("Checking the file first.") == 1 + + +class TestDiffPreviewAndConfirmationUX: + def test_large_new_file_diff_is_capped_at_30_lines_by_default(self): + big_py = "".join(f"line_{i} = {i}\n" for i in range(120)) + with patch("localm.plugins.coder.display.console"), \ + patch("localm.plugins.coder.display.Syntax") as mock_syntax: + display.print_diff_preview("", big_py, "big_module.py") + diff_text = mock_syntax.call_args[0][0] + assert "more lines truncated - press [f] at prompt for full diff" in diff_text + assert "line_119" not in diff_text + + def test_confirm_diff_f_expands_full_diff_then_approves(self, monkeypatch): + big_py = "".join(f"line_{i} = {i}\n" for i in range(80)) + answers = iter(["f", "y"]) + monkeypatch.setattr(display.console, "input", lambda _prompt="": next(answers)) + + printed_syntaxes = [] + with patch("localm.plugins.coder.display.Syntax", + side_effect=lambda text, *a, **kw: printed_syntaxes.append(text) or text), \ + patch.object(display.console, "print"): + display.print_diff_preview("", big_py, "app.py") + approved = display.confirm_diff("app.py") + + assert approved is True + assert len(printed_syntaxes) == 2 + assert "truncated" in printed_syntaxes[0] + assert "line_79 = 79" in printed_syntaxes[1] + + def test_confirm_always_enables_session_auto_approve(self, tmp_path, monkeypatch): + agent = _make_agent(tmp_path, auto_approve=False) + monkeypatch.setattr(display.console, "input", lambda _prompt="": "a") + + call = ToolCall( + name="write_file", + args={"path": "new.py", "content": "x = 1\n"}, + raw="", + start=0, + end=0, + ) + res = agent._execute_tool(call, interactive=True) + assert res.ok is True + assert agent.auto_approve is True + + def test_confirm_rejection_reason_is_passed_back_to_agent(self, tmp_path, monkeypatch): + agent = _make_agent(tmp_path, auto_approve=False) + monkeypatch.setattr( + display.console, "input", lambda _prompt="": "n: keep existing config" + ) + + call = ToolCall( + name="write_file", + args={"path": "cfg.py", "content": "x = 2\n"}, + raw="", + start=0, + end=0, + ) + res = agent._execute_tool(call, interactive=True) + assert res.ok is False + assert "Rejected by user: keep existing config" in res.output + + +class TestToolResultAndCallFormatting: + def test_failed_tool_prints_error_excerpt_even_when_not_verbose(self, capsys): + res = ToolResult( + ok=False, + output="1\n\nFAILED tests/test_foo.py::test_bar - AssertionError: 1 != 2\n", + summary="exit 1", + ) + display.print_tool_result("run_shell", res, verbose=False, elapsed=0.42) + out = capsys.readouterr().out + assert "exit 1" in out + assert "(0.4s)" in out + assert "AssertionError: 1 != 2" in out + assert "" not in out + + def test_set_todos_renders_checklist(self, capsys): + res = ToolResult.success("3 items", summary="3 items (1 done, 1 in_progress, 1 pending)") + items = [ + {"task": "Inspect display.py", "status": "done"}, + {"task": "Add ActivitySpinner", "status": "in_progress"}, + {"task": "Run pytest", "status": "pending"}, + ] + display.print_tool_result("set_todos", res, verbose=False, args={"items": items}) + out = capsys.readouterr().out + assert "[βœ“] Inspect display.py" in out + assert "[β–Έ] Add ActivitySpinner" in out + assert "[ ] Run pytest" in out + + def test_multiline_edit_file_args_summarized_by_line_count(self): + text = display.tool_call_text( + "edit_file", + {"path": "a.py", "old": "line1\nline2\nline3", "new": "line1\nline2"}, + ) + assert "old=<3 lines>" in text + assert "new=<2 lines>" in text + + +class TestReplAtMentionsAndPaste: + def test_expand_at_mentions_attaches_file_and_line_slice(self, tmp_path): + f = tmp_path / "sample.py" + f.write_text("a = 1\nb = 2\nc = 3\nd = 4\n", encoding="utf-8") + + expanded = _expand_at_mentions( + "Check @sample.py:2-3 and ignore @property decorator", tmp_path + ) + assert '' in expanded + assert "b = 2\nc = 3" in expanded + assert "a = 1" not in expanded + + def test_expand_at_mentions_blocks_outside_cwd(self, tmp_path): + outside = tmp_path / "secret.txt" + outside.write_text("top-secret\n", encoding="utf-8") + sub = tmp_path / "proj" + sub.mkdir() + + expanded = _expand_at_mentions("Read @../secret.txt", sub) + assert "top-secret" not in expanded + assert " 0: + state["count"] -= 1 + return True + return False + + monkeypatch.setattr("localm.plugins.coder.cli.repl._stdin_has_buffered_data", has_data) + monkeypatch.setattr("sys.stdin.readline", lambda: next(buffered)) + + msg = _read_multiline() + assert msg == "first line\nsecond line\nthird line" From 2b15781cba3a3201efbc8e83f6c5c083710a31c1 Mon Sep 17 00:00:00 2001 From: Matlan Date: Tue, 20 Oct 2026 14:26:26 +0200 Subject: [PATCH 2/5] Coder: 'a' at a prompt allows only that tool (or exact shell command) for the session; @file reads through read_file and honours /scope Answering 'a' used to switch on session-wide auto-approve, so one 'a' on a file write silently approved every later shell command, push and network call. It now allows only the tool it was asked about, or the exact command for a shell tool, and never waives a shell-guard verdict, a loosely formatted call or a configured always_confirm. @file attachments go through read_file (same confinement, notebook rendering and truncation the model's own read gets) and skip paths outside the session's /scope. Also fixes the ruff findings (B023, UP037) and stops printing a static 'Thinking...' line next to the live spinner on a terminal. Co-Authored-By: Claude Opus 5.5 --- localm/plugins/coder/agent/context.py | 9 ++- localm/plugins/coder/agent/core.py | 4 ++ localm/plugins/coder/agent/execution.py | 42 +++++++++++- localm/plugins/coder/cli/repl.py | 77 ++++++++++----------- localm/plugins/coder/display.py | 16 +++-- tests/plugins/coder/test_coder_ui_ux.py | 89 +++++++++++++++++++++---- 6 files changed, 172 insertions(+), 65 deletions(-) diff --git a/localm/plugins/coder/agent/context.py b/localm/plugins/coder/agent/context.py index 9da2d160c..e9408aedc 100644 --- a/localm/plugins/coder/agent/context.py +++ b/localm/plugins/coder/agent/context.py @@ -13,7 +13,7 @@ from ..display import ( ActivitySpinner, console, print_assistant_label, print_info, print_reasoning_token, print_status, print_streaming_done, - print_streaming_token, print_thinking, + print_streaming_token, print_thinking, spinner_available, ) from ..parser import _EXPLICIT_FENCE_LANGS, _try_parse_body from .constants import _COMPACT_AUTO_RATIO, _COMPACT_WARN_RATIO, _DEFAULT_CTX_TOKENS @@ -1091,13 +1091,16 @@ def _call_llm(self, messages: list[dict], interactive: bool) -> str: ) if interactive: if first_attempt: - print_thinking() + if spinner_available(): + console.print() + else: + print_thinking() first_attempt = False terminal = _TerminalReply(self.name) interrupted = False - def _on_interrupt() -> None: + def _on_interrupt(terminal: _TerminalReply = terminal) -> None: nonlocal interrupted interrupted = True terminal.finish() diff --git a/localm/plugins/coder/agent/core.py b/localm/plugins/coder/agent/core.py index 56c024506..34a555df4 100644 --- a/localm/plugins/coder/agent/core.py +++ b/localm/plugins/coder/agent/core.py @@ -338,6 +338,10 @@ def __init__( # removed when the identical call later runs. _loop re-arms it. self._denied_unconfirmed: list[tuple[str, str, str]] = [] self._denied_lock = threading.Lock() + # Keys the user answered "a" (always) for at an interactive prompt this + # session: a tool name, or (tool, exact command) for shell-executing + # tools. See _ExecutionMixin._session_allow_key. + self._session_allowed: set = set() self._undo_stack: list[dict] = [] self._unverified_writes: set[str] = set() # code files changed since last test run # Changed-files tracker: rel path -> {original: bytes|None, writes: int, diff --git a/localm/plugins/coder/agent/execution.py b/localm/plugins/coder/agent/execution.py index 1cae121df..b8cf4b9c7 100644 --- a/localm/plugins/coder/agent/execution.py +++ b/localm/plugins/coder/agent/execution.py @@ -579,15 +579,32 @@ def _execute_tool(self, call: ToolCall, interactive: bool) -> ToolResult: lenient_call and gated and not shell_unchecked and self.auto_approve and call.name not in self.always_confirm ) + # A session allow ("a" at an interactive prompt) never waives a + # shell-guard verdict, a lenient call or a configured always_confirm. + # See test_session_allow_never_waives_always_confirm_or_lenient. + allow_key = self._session_allow_key(call) + session_allowable = ( + allow_key is not None and not shell_unchecked and not lenient_call + and call.name not in self.always_confirm + ) + session_allowed = getattr(self, "_session_allowed", None) + if session_allowed is None: + session_allowed = self._session_allowed = set() + if needs_confirm and session_allowable and allow_key in session_allowed: + needs_confirm = False if needs_confirm: if self.confirm_handler is not None: approved = invoke_confirm(self.confirm_handler, call, agent=self._confirm_agent_label()) elif interactive: approved = self._confirm_tool(call) - if approved and consume_confirm_always(): - self.auto_approve = True - print_info(" βœ“ Auto-approve enabled for the rest of this session.") + if consume_confirm_always() and approved: + if session_allowable: + session_allowed.add(allow_key) + print_info(" " + self._session_allow_notice(call)) + else: + print_info(" Approved once: this call cannot be " + "allowed for the whole session.") else: # Fail CLOSED: this tool requires confirmation and there is no way # to obtain it (non-interactive run, no approval handler), so deny @@ -910,6 +927,25 @@ def _confirm_agent_label(self) -> Optional[str]: return None return self.name or "sub-agent" + @staticmethod + def _session_allow_key(call: ToolCall): + """The key an "always" answer stores for *call*, or None if it cannot + be session-allowed. Shell-executing tools are keyed by the exact + command string, every other tool by its name.""" + if call.name in _SHELL_GUARDED_TOOLS: + cmd = call.args.get("command") if isinstance(call.args, dict) else None + if not isinstance(cmd, str) or not cmd.strip(): + return None + return (call.name, cmd.strip()) + return call.name + + @staticmethod + def _session_allow_notice(call: ToolCall) -> str: + if call.name in _SHELL_GUARDED_TOOLS: + return (f"Allowing this exact {call.name} command without asking " + "for the rest of this session.") + return f"Allowing {call.name} without asking for the rest of this session." + def _confirm_tool(self, call: ToolCall) -> bool: """ Ask the user to approve a destructive tool call. diff --git a/localm/plugins/coder/cli/repl.py b/localm/plugins/coder/cli/repl.py index fec89d055..ea3a00cf3 100644 --- a/localm/plugins/coder/cli/repl.py +++ b/localm/plugins/coder/cli/repl.py @@ -5,6 +5,8 @@ from __future__ import annotations import os +import re +import sys import time from pathlib import Path @@ -24,13 +26,9 @@ from ..review_guard import classify_sensitive_changes, render_warning from .goal import _run_goal_loop -import re -import sys - _AT_MENTION_RE = re.compile( r"(?:^|(?<=\s))@([A-Za-z0-9_./\\-]+(?::\d+(?:-\d+)?)?)(?=\s|$|[.,;!?])" ) -_MAX_AT_FILE_CHARS = 24_000 def _stdin_has_buffered_data() -> bool: @@ -79,59 +77,50 @@ def _read_multiline() -> str: return "\n".join(lines) -def _expand_at_mentions(user_input: str, cwd: Path) -> str: - """Attach contents of existing workspace files referenced as @path[:start-end].""" +def _expand_at_mentions(user_input: str, cwd: Path, scope_allows=None) -> str: + """Attach existing workspace files referenced as @path or @path:start-end. + + Each file is read through the read_file tool, so confinement, notebook + rendering and size truncation match what the model would see from its own + read. A path outside *scope_allows* (the session's /scope) is not attached. + Tokens that name no file (a decorator, an @handle) are left alone.""" if "@" not in user_input: return user_input - from ..tools.files import _confine + from ..tools.files import tool_read_file attachments: list[str] = [] seen: set[str] = set() for match in _AT_MENTION_RE.finditer(user_input): - spec = match.group(1) - if spec in seen: + spec = match.group(1).rstrip(".") + if not spec or spec in seen: continue seen.add(spec) - path_part = spec - line_range: tuple[int, int] | None = None + path_part, offset, limit = spec, 0, 0 if ":" in spec: base, rng = spec.rsplit(":", 1) - if re.fullmatch(r"\d+(?:-\d+)?", rng): + m = re.fullmatch(r"(\d+)(?:-(\d+))?", rng) + if m: path_part = base - if "-" in rng: - s_str, e_str = rng.split("-", 1) - line_range = (max(1, int(s_str)), max(1, int(e_str))) - else: - ln = max(1, int(rng)) - line_range = (ln, ln) + offset = max(1, int(m.group(1))) + end_ln = max(offset, int(m.group(2))) if m.group(2) else offset + limit = end_ln - offset + 1 + candidate = cwd / path_part try: - resolved = _confine(cwd, path_part) - except PermissionError: + if not candidate.is_file(): + continue + except OSError: continue - if not resolved.is_file(): + if scope_allows is not None and not scope_allows(path_part): + print_info(f" @{path_part} is outside this session's scope - not attached") continue - try: - raw = resolved.read_text(encoding="utf-8", errors="replace") - except Exception: + result = tool_read_file(cwd, path_part, offset, limit) + if not result.ok: + print_info(f" @{spec} not attached: {result.output}") continue - rel = resolved.relative_to(cwd) if resolved.is_relative_to(cwd) else resolved - all_lines = raw.splitlines() - if line_range is not None: - start_ln, end_ln = line_range - sliced = all_lines[start_ln - 1:end_ln] - body = "\n".join(sliced) - label = f"{rel}:{start_ln}-{end_ln}" - n_lines = len(sliced) - else: - body = raw - label = str(rel) - n_lines = len(all_lines) - trunc_note = "" - if len(body) > _MAX_AT_FILE_CHARS: - body = body[:_MAX_AT_FILE_CHARS] + "\n... [truncated]" - trunc_note = " (truncated)" - print_info(f" πŸ“Ž Attached {label} ({n_lines} lines{trunc_note})") - attachments.append(f'\n{body}\n') + trunc_note = " (truncated)" if result.truncated else "" + print_info(f" Attached {result.summary}{trunc_note}") + attachments.append( + f'\n{result.output}\n') if not attachments: return user_input @@ -259,7 +248,9 @@ def _repl(agent: Agent) -> None: break continue - expanded_input = _expand_at_mentions(user_input, agent.cwd) + expanded_input = _expand_at_mentions( + user_input, agent.cwd, + scope_allows=agent._scope_allows if agent.scope else None) try: # The saved checkpoint is not cleared before the turn; the turn diff --git a/localm/plugins/coder/display.py b/localm/plugins/coder/display.py index fcf358d05..92e5b7c90 100644 --- a/localm/plugins/coder/display.py +++ b/localm/plugins/coder/display.py @@ -74,7 +74,7 @@ def _is_interactive_tty(self, c: Console) -> bool: is_term = getattr(c, "is_terminal", None) return isinstance(is_term, bool) and is_term - def start(self, label: str | None = None) -> "ActivitySpinner": + def start(self, label: str | None = None) -> ActivitySpinner: with self._lock: if label is not None: self._label = label @@ -147,13 +147,20 @@ def stop(self) -> None: ) self._last_len = 0 - def __enter__(self) -> "ActivitySpinner": + def __enter__(self) -> ActivitySpinner: return self.start() def __exit__(self, *exc) -> None: self.stop() +def spinner_available(console_obj: Console | None = None) -> bool: + """True when an ActivitySpinner would animate on *console_obj* (default: + the coder console), i.e. it is a real terminal.""" + return ActivitySpinner(console_obj=console_obj)._is_interactive_tty( + console_obj if console_obj is not None else console) + + def _strip_esc(s: str) -> str: """Remove terminal control characters (ESC and every other C0/C1 control except newline and tab) so externally-controlled text (model output, tool @@ -605,7 +612,8 @@ def confirm_diff(path_label: str) -> bool: try: while True: has_full = bool(_last_diff_preview and _last_diff_preview[3]) - opts = "[y/N/a=always/f=full] " if has_full else "[y/N/a=always] " + opts = ("[y/N/a=allow this session/f=full diff] " if has_full + else "[y/N/a=allow this session] ") prompt = Text("Apply changes to ", style="yellow") prompt.append(_strip_esc(path_label), style="bold yellow") prompt.append(f"? {opts}", style="yellow") @@ -635,7 +643,7 @@ def confirm(prompt: str) -> bool: _last_confirm_always = False _last_reject_reason = "" try: - line = _sanitized_text(f"{prompt} [y/N/a=always] ", style="yellow") + line = _sanitized_text(f"{prompt} [y/N/a=allow this session] ", style="yellow") raw = console.input(line) action, reason = _parse_confirm_answer(raw) if action == "always": diff --git a/tests/plugins/coder/test_coder_ui_ux.py b/tests/plugins/coder/test_coder_ui_ux.py index 08d4e74ff..e42003560 100644 --- a/tests/plugins/coder/test_coder_ui_ux.py +++ b/tests/plugins/coder/test_coder_ui_ux.py @@ -151,20 +151,71 @@ def test_confirm_diff_f_expands_full_diff_then_approves(self, monkeypatch): assert "truncated" in printed_syntaxes[0] assert "line_79 = 79" in printed_syntaxes[1] - def test_confirm_always_enables_session_auto_approve(self, tmp_path, monkeypatch): + def test_session_allow_covers_only_that_tool(self, tmp_path, monkeypatch): agent = _make_agent(tmp_path, auto_approve=False) - monkeypatch.setattr(display.console, "input", lambda _prompt="": "a") + prompts = [] + answers = iter(["a", "n"]) + + def fake_input(prompt=""): + prompts.append(str(prompt)) + return next(answers) + + monkeypatch.setattr(display.console, "input", fake_input) + first = ToolCall(name="write_file", args={"path": "new.py", "content": "x = 1\n"}, + raw="", start=0, end=0) + assert agent._execute_tool(first, interactive=True).ok is True + assert agent.auto_approve is False + + second = ToolCall(name="write_file", args={"path": "other.py", "content": "y = 2\n"}, + raw="", start=0, end=0) + assert agent._execute_tool(second, interactive=True).ok is True + assert len(prompts) == 1 + + shell = ToolCall(name="run_shell", args={"command": "echo hi"}, + raw="", start=0, end=0) + res = agent._execute_tool(shell, interactive=True) + assert len(prompts) == 2 + assert res.ok is False - call = ToolCall( - name="write_file", - args={"path": "new.py", "content": "x = 1\n"}, - raw="", - start=0, - end=0, - ) - res = agent._execute_tool(call, interactive=True) - assert res.ok is True - assert agent.auto_approve is True + def test_session_allow_for_shell_is_exact_command_only(self, tmp_path, monkeypatch): + agent = _make_agent(tmp_path, auto_approve=False) + prompts = [] + answers = iter(["a", "n"]) + + def fake_input(prompt=""): + prompts.append(str(prompt)) + return next(answers) + + monkeypatch.setattr(display.console, "input", fake_input) + + def call(cmd): + return ToolCall(name="run_shell", args={"command": cmd}, raw="", start=0, end=0) + + agent._execute_tool(call("python --version"), interactive=True) + agent._execute_tool(call("python --version"), interactive=True) + assert len(prompts) == 1 + agent._execute_tool(call("python --version && echo x"), interactive=True) + assert len(prompts) == 2 + + def test_session_allow_never_waives_always_confirm_or_lenient(self, tmp_path, monkeypatch): + agent = _make_agent(tmp_path, auto_approve=False) + agent.always_confirm = {"write_file"} + prompts = [] + monkeypatch.setattr(display.console, "input", + lambda prompt="": prompts.append(str(prompt)) or "a") + for name in ("a.py", "b.py"): + c = ToolCall(name="write_file", args={"path": name, "content": "z = 0\n"}, + raw="", start=0, end=0) + assert agent._execute_tool(c, interactive=True).ok is True + assert len(prompts) == 2 + + agent.always_confirm = set() + prompts.clear() + for name in ("c.py", "d.py"): + c = ToolCall(name="write_file", args={"path": name, "content": "z = 0\n"}, + raw="", start=0, end=0, lenient=True) + assert agent._execute_tool(c, interactive=True).ok is True + assert len(prompts) == 2 def test_confirm_rejection_reason_is_passed_back_to_agent(self, tmp_path, monkeypatch): agent = _make_agent(tmp_path, auto_approve=False) @@ -242,6 +293,20 @@ def test_expand_at_mentions_blocks_outside_cwd(self, tmp_path): assert "top-secret" not in expanded assert " Date: Tue, 20 Oct 2026 14:59:17 +0200 Subject: [PATCH 3/5] Coder: free text at an approval prompt rejects and tells the agent why; plain /compact hint Typing anything other than y/a/f/n at an approval prompt used to count as a silent "no", so an instruction typed there was lost. It is now a rejection that carries the typed text back to the agent. The context-full warning printed literal "[bold]" tags because warnings are sanitised; it is plain text now. Changelog entries rewritten for users. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 22 +++++++++---------- localm/plugins/coder/agent/context.py | 2 +- localm/plugins/coder/display.py | 3 ++- tests/plugins/coder/test_coder_ui_ux.py | 28 +++++++++++++++++++++++++ 4 files changed, 41 insertions(+), 14 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b2110717c..75b4e2323 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,20 +12,18 @@ permanent public record of what shipped and are never rewritten; the in-progress ## [Unreleased] ### Added -- **CLI coder terminal UI and interactive approval upgrades.** `localm coder` now displays a - scrollback-safe inline spinner while waiting for the first token or running tools, shows live - line- and character-count progress when the model streams large file writes or patches inside - `` blocks, renders `set_todos` as a color-coded terminal checklist, supports inline - `@path` and `@path:start-end` context mentions in prompts, coalesces multi-line pastes in the - REPL, and expands write/edit/bash approval prompts to `[y/N/a=always/f=full]` (or - `[y/N/a=always]` for commands) with optional rejection feedback routed back to the agent. +- **`localm coder` shows what it is doing.** A live spinner runs while the model thinks, while + it writes a file (with the file name and line count) and while a tool runs. `set_todos` shows + the task list. Type `@path` or `@path:10-40` to attach a file to your message, and a + multi-line paste arrives as one message. +- **Approval prompts in `localm coder` take more answers.** `a` allows that tool for the rest of + the session (for a shell command, that exact command only), `f` shows the full diff, and + `n: ` tells the agent why you said no. ### Fixed -- **CLI coder console output during file creation, streaming, and tool failures.** New-file - previews now cap to a compact preview instead of dumping entire files line by line, streamed - assistant responses no longer print a second time as a Markdown panel after completion, and - failed tool calls now print a compact error excerpt in normal mode instead of hiding the - failure behind `--verbose`. +- **`localm coder` no longer floods the terminal.** A new file shows a short preview instead of + every line, a reply is no longer printed twice, and a failed tool shows the end of its error + output without needing `/verbose`. Warnings at the end of a task now appear in the terminal. - **Image prompts with letters such as ß work with the native image backend on Windows.** A prompt containing ß failed with "failed to encode prompt", and letters such as Γ€, ΓΆ and ΓΌ reached the diff --git a/localm/plugins/coder/agent/context.py b/localm/plugins/coder/agent/context.py index e9408aedc..7559675d2 100644 --- a/localm/plugins/coder/agent/context.py +++ b/localm/plugins/coder/agent/context.py @@ -632,7 +632,7 @@ def _maybe_compact(self, interactive: bool) -> None: if ratio >= _COMPACT_WARN_RATIO and not getattr(self, "_compact_warned", False): print_warning( f"Context is ~{ratio:.0%} full. " - "Use [bold]/compact[/bold] to summarise old turns." + "Use /compact to summarise old turns." ) self._compact_warned = True # warn once per session else: diff --git a/localm/plugins/coder/display.py b/localm/plugins/coder/display.py index 92e5b7c90..c9f3d192a 100644 --- a/localm/plugins/coder/display.py +++ b/localm/plugins/coder/display.py @@ -601,7 +601,8 @@ def _parse_confirm_answer(raw: str) -> tuple[str, str]: for prefix in ("n:", "no:", "n ", "no "): if lower.startswith(prefix): return "no", s[len(prefix):].strip() - return "no", "" + # Any other text is a rejection that tells the agent what to do instead. + return "no", s def confirm_diff(path_label: str) -> bool: diff --git a/tests/plugins/coder/test_coder_ui_ux.py b/tests/plugins/coder/test_coder_ui_ux.py index e42003560..f0e1c313c 100644 --- a/tests/plugins/coder/test_coder_ui_ux.py +++ b/tests/plugins/coder/test_coder_ui_ux.py @@ -235,6 +235,34 @@ def test_confirm_rejection_reason_is_passed_back_to_agent(self, tmp_path, monkey assert "Rejected by user: keep existing config" in res.output + def test_free_text_at_prompt_rejects_and_passes_the_text_back(self, tmp_path, monkeypatch): + agent = _make_agent(tmp_path, auto_approve=False) + monkeypatch.setattr(display.console, "input", + lambda _prompt="": "put it in src/ instead") + call = ToolCall(name="write_file", args={"path": "cfg.py", "content": "x = 2\n"}, + raw="", start=0, end=0) + res = agent._execute_tool(call, interactive=True) + assert res.ok is False + assert "Rejected by user: put it in src/ instead" in res.output + assert not (tmp_path / "cfg.py").exists() + + def test_bare_enter_and_n_reject_without_a_reason(self): + assert display._parse_confirm_answer("") == ("no", "") + assert display._parse_confirm_answer("N") == ("no", "") + assert display._parse_confirm_answer(" y ") == ("yes", "") + + +class TestContextWarningText: + def test_compact_warning_carries_no_literal_markup(self, tmp_path): + agent = _make_agent(tmp_path) + warnings = [] + with patch("localm.plugins.coder.agent.print_warning", side_effect=warnings.append), \ + patch.object(type(agent), "_fill_ratio", return_value=0.9): + agent._maybe_compact(interactive=True) + assert warnings and "/compact" in warnings[0] + assert "[bold]" not in warnings[0] + + class TestToolResultAndCallFormatting: def test_failed_tool_prints_error_excerpt_even_when_not_verbose(self, capsys): res = ToolResult( From e0d295476a3630a59a91badda41cd1a2525447ea Mon Sep 17 00:00:00 2001 From: Matlan Date: Tue, 20 Oct 2026 15:33:04 +0200 Subject: [PATCH 4/5] Coder: fixes from review of the UI overhaul - @file text no longer counts as the request: chat() and run_task() take attachments separately, so a question about an attached file is not re-prompted for a tool call, and the session title, episode task and reviewer task hold only what the user typed (goal mode too). - [f] at a prompt shows only this file's diff; the stored preview is forgotten before every tool prompt. - "a" is offered only where it is honoured: on tool prompts, including a child agent's prompt routed to the parent terminal (which now also keeps its rejection reason); not on "Keep going?", save or export prompts. - URL tools are allowed per host; allows end on /cd and /clear. - The GUI rejection event keeps its exact "rejected by user" summary. - No spinner over shell, git or test runs (a subprocess may prompt on the terminal), nor while a background child compacts its history; an ordinary JSON code block is no longer reported as a tool call in progress. - The end-of-task footer and sensitive-file warning cover this task's files only; a tool call that could not be read is now reported to the user. - POSIX paste coalescing reads the terminal file descriptor directly, so a single read holding several pasted lines loses none of them. - basedpyright baseline entries refreshed for the touched coder files. Co-Authored-By: Claude Opus 5.5 --- .github/basedpyright-baseline.json | 96 +++++++-- localm/plugins/coder/agent/context.py | 9 +- localm/plugins/coder/agent/execution.py | 67 ++++-- localm/plugins/coder/agent/loop.py | 43 +++- localm/plugins/coder/agent/session.py | 5 +- localm/plugins/coder/cli/goal.py | 8 +- localm/plugins/coder/cli/repl.py | 81 ++++++-- localm/plugins/coder/display.py | 76 ++++--- tests/plugins/coder/test_coder_ui_ux.py | 263 ++++++++++++++++++++++++ 9 files changed, 557 insertions(+), 91 deletions(-) diff --git a/.github/basedpyright-baseline.json b/.github/basedpyright-baseline.json index b67c01506..408b39978 100644 --- a/.github/basedpyright-baseline.json +++ b/.github/basedpyright-baseline.json @@ -5224,8 +5224,8 @@ { "code": "reportAttributeAccessIssue", "range": { - "startColumn": 21, - "endColumn": 28, + "startColumn": 26, + "endColumn": 33, "lineCount": 1 } }, @@ -5625,6 +5625,14 @@ "lineCount": 1 } }, + { + "code": "reportAttributeAccessIssue", + "range": { + "startColumn": 38, + "endColumn": 52, + "lineCount": 1 + } + }, { "code": "reportAttributeAccessIssue", "range": { @@ -5636,8 +5644,16 @@ { "code": "reportAttributeAccessIssue", "range": { - "startColumn": 47, - "endColumn": 62, + "startColumn": 28, + "endColumn": 43, + "lineCount": 1 + } + }, + { + "code": "reportAttributeAccessIssue", + "range": { + "startColumn": 55, + "endColumn": 70, "lineCount": 1 } }, @@ -5732,8 +5748,8 @@ { "code": "reportAttributeAccessIssue", "range": { - "startColumn": 62, - "endColumn": 69, + "startColumn": 48, + "endColumn": 55, "lineCount": 1 } }, @@ -6027,6 +6043,14 @@ "lineCount": 1 } }, + { + "code": "reportAttributeAccessIssue", + "range": { + "startColumn": 59, + "endColumn": 72, + "lineCount": 1 + } + }, { "code": "reportAttributeAccessIssue", "range": { @@ -6582,8 +6606,24 @@ { "code": "reportAttributeAccessIssue", "range": { - "startColumn": 36, - "endColumn": 44, + "startColumn": 32, + "endColumn": 40, + "lineCount": 1 + } + }, + { + "code": "reportAttributeAccessIssue", + "range": { + "startColumn": 35, + "endColumn": 46, + "lineCount": 1 + } + }, + { + "code": "reportAttributeAccessIssue", + "range": { + "startColumn": 38, + "endColumn": 46, "lineCount": 1 } }, @@ -6718,8 +6758,16 @@ { "code": "reportAttributeAccessIssue", "range": { - "startColumn": 69, - "endColumn": 82, + "startColumn": 32, + "endColumn": 45, + "lineCount": 1 + } + }, + { + "code": "reportAttributeAccessIssue", + "range": { + "startColumn": 21, + "endColumn": 34, "lineCount": 1 } }, @@ -6742,8 +6790,16 @@ { "code": "reportAttributeAccessIssue", "range": { - "startColumn": 22, - "endColumn": 31, + "startColumn": 30, + "endColumn": 39, + "lineCount": 1 + } + }, + { + "code": "reportAttributeAccessIssue", + "range": { + "startColumn": 26, + "endColumn": 35, "lineCount": 1 } }, @@ -6838,8 +6894,16 @@ { "code": "reportAttributeAccessIssue", "range": { - "startColumn": 47, - "endColumn": 50, + "startColumn": 55, + "endColumn": 58, + "lineCount": 1 + } + }, + { + "code": "reportAttributeAccessIssue", + "range": { + "startColumn": 51, + "endColumn": 54, "lineCount": 1 } }, @@ -7006,8 +7070,8 @@ { "code": "reportAttributeAccessIssue", "range": { - "startColumn": 78, - "endColumn": 85, + "startColumn": 64, + "endColumn": 71, "lineCount": 1 } } diff --git a/localm/plugins/coder/agent/context.py b/localm/plugins/coder/agent/context.py index 7559675d2..2f8de3908 100644 --- a/localm/plugins/coder/agent/context.py +++ b/localm/plugins/coder/agent/context.py @@ -556,7 +556,7 @@ def _pass(earlier, excerpt) -> str: except Exception: return raw - if getattr(self, "on_event", None) is None and getattr(self, "progress_sink", None) is None: + if getattr(self, "_interactive", False) and getattr(self, "on_event", None) is None: with ActivitySpinner("Compacting session history..."): summary, failure = summarise_in_passes( groups, _pass, reply_tokens=reply_tokens, measure=measure) @@ -1031,10 +1031,13 @@ def _tapped_stream(): yield chunk if on_tool_progress is not None and len(raw_seen) - len(full) >= 12: pending = raw_seen[len(full):] - if any(m in pending for m in (" ToolResult: if needs_confirm and session_allowable and allow_key in session_allowed: needs_confirm = False if needs_confirm: - if self.confirm_handler is not None: - approved = invoke_confirm(self.confirm_handler, call, - agent=self._confirm_agent_label()) - elif interactive: - approved = self._confirm_tool(call) - if consume_confirm_always() and approved: - if session_allowable: - session_allowed.add(allow_key) - print_info(" " + self._session_allow_notice(call)) + if self.confirm_handler is not None or interactive: + # The terminal prompt (this agent's, or an interactive parent's + # reached through confirm_handler) offers "a" only for a call + # that can be session-allowed. A GUI handler never sets the + # answer flags, so consuming them there reads False and "". + with offering_session_allow(session_allowable): + if self.confirm_handler is not None: + approved = invoke_confirm(self.confirm_handler, call, + agent=self._confirm_agent_label()) else: - print_info(" Approved once: this call cannot be " - "allowed for the whole session.") + approved = self._confirm_tool(call) + if consume_confirm_always() and approved and session_allowable: + session_allowed.add(allow_key) + print_info(" " + self._session_allow_notice(call)) else: # Fail CLOSED: this tool requires confirmation and there is no way # to obtain it (non-interactive run, no approval handler), so deny @@ -636,7 +639,7 @@ def _execute_tool(self, call: ToolCall, interactive: bool) -> ToolResult: summary="denied: confirmation required, none available") return result if not approved: - reject_reason = consume_reject_reason() if interactive else "" + reject_reason = consume_reject_reason() err_msg = ( f"Rejected by user: {reject_reason}" if reject_reason else "Rejected by user." @@ -644,8 +647,10 @@ def _execute_tool(self, call: ToolCall, interactive: bool) -> ToolResult: result = ToolResult.error(err_msg) if interactive: print_tool_result(call.name, result, verbose=False) + # The GUI matches this exact summary to drop its duplicate card. + # See test_rejection_event_summary_is_the_literal_the_gui_matches. self._emit("tool_result", tool=call.name, ok=False, - summary=err_msg.lower()) + summary="rejected by user") return result # Snapshot file content before undoable writes so /undo can restore it and @@ -721,7 +726,7 @@ def _execute_tool(self, call: ToolCall, interactive: bool) -> ToolResult: # how long the tool took - the number the GUI shows next to the card. t_start = time.monotonic() try: - if interactive and call.name not in _PARENT_AGENT_TOOLS: + if interactive and self._spins_while_running(call.name): with ActivitySpinner(f"Running {call.name}..."): result = tool_def.fn(self.cwd, **args) else: @@ -927,23 +932,44 @@ def _confirm_agent_label(self) -> Optional[str]: return None return self.name or "sub-agent" + @staticmethod + def _spins_while_running(name: str) -> bool: + """Whether the live spinner animates while tool *name* runs. Not for a + tool that may hand the terminal to a subprocess prompt (a shell, git + asking for credentials or a passphrase) or that prints its own progress + (sub-agents). See test_spinner_skips_tools_that_can_prompt_on_the_terminal.""" + return not (name in _PARENT_AGENT_TOOLS or name in _SHELL_GUARDED_TOOLS + or name == "run_tests" or name.startswith("git_")) + @staticmethod def _session_allow_key(call: ToolCall): """The key an "always" answer stores for *call*, or None if it cannot be session-allowed. Shell-executing tools are keyed by the exact - command string, every other tool by its name.""" + command string, URL tools by host, every other tool by its name.""" + args = call.args if isinstance(call.args, dict) else {} if call.name in _SHELL_GUARDED_TOOLS: - cmd = call.args.get("command") if isinstance(call.args, dict) else None + cmd = args.get("command") if not isinstance(cmd, str) or not cmd.strip(): return None return (call.name, cmd.strip()) + if "url" in args: + from urllib.parse import urlsplit + try: + host = urlsplit(str(args.get("url") or "")).hostname + except ValueError: + host = None + return (call.name, "host", host) if host else None return call.name - @staticmethod - def _session_allow_notice(call: ToolCall) -> str: + @classmethod + def _session_allow_notice(cls, call: ToolCall) -> str: + key = cls._session_allow_key(call) if call.name in _SHELL_GUARDED_TOOLS: return (f"Allowing this exact {call.name} command without asking " "for the rest of this session.") + if isinstance(key, tuple) and len(key) == 3: + return (f"Allowing {call.name} to {key[2]} without asking for the " + "rest of this session.") return f"Allowing {call.name} without asking for the rest of this session." def _confirm_tool(self, call: ToolCall) -> bool: @@ -959,6 +985,7 @@ def _confirm_tool(self, call: ToolCall) -> bool: confirm = _agent.confirm confirm_diff = _agent.confirm_diff print_diff_preview = _agent.print_diff_preview + reset_diff_preview() if call.name in ("write_file", "edit_file"): path_arg = call.args.get("path", "") old_content, readable = read_old_content_checked(self.cwd, path_arg) diff --git a/localm/plugins/coder/agent/loop.py b/localm/plugins/coder/agent/loop.py index 9ca498d14..fc7860a9c 100644 --- a/localm/plugins/coder/agent/loop.py +++ b/localm/plugins/coder/agent/loop.py @@ -251,10 +251,13 @@ def response_similarity(a: str, b: str) -> float: class _LoopMixin: - def run_task(self, task: str) -> str: + def run_task(self, task: str, attachments: str = "") -> str: """ Run a single task to completion (non-interactive). + *attachments* is appended to what the model receives only, as in + :meth:`chat`. + Returns the agent's final text response. Used by spawn_agent and the CLI `run` command. """ @@ -266,7 +269,8 @@ def run_task(self, task: str) -> str: self._session_title = task # resume-listing display title # The raw request, before _with_episodes can prepend a lessons preamble. self._last_user_request = task - self._add_user(self._with_episodes(task)) + message = self._with_episodes(task) + self._add_user(compose(message, attachments) if attachments else message) return self._loop(interactive=False) def continue_task(self, message: str) -> str: @@ -282,10 +286,15 @@ def continue_task(self, message: str) -> str: self._add_user(message) return self._loop(interactive=False) - def chat(self, user_input: str) -> str: + def chat(self, user_input: str, attachments: str = "") -> str: """ Send one user message in an ongoing interactive session. + *attachments* (the REPL's @file blocks) is appended to the message the + model receives but is not part of the request text the session title, + action classification, episode recall and reviewer read. + See test_attachment_text_does_not_turn_a_question_into_an_action. + History is preserved between calls. Returns the agent's final text response for this turn. """ @@ -300,7 +309,7 @@ def chat(self, user_input: str) -> str: user_input = self._with_episodes(user_input) if not self._review_task: self._review_task = user_input - self._add_user(user_input) + self._add_user(compose(user_input, attachments) if attachments else user_input) return self._loop(interactive=True) def _with_episodes(self, task: str) -> str: @@ -389,6 +398,8 @@ def _loop(self, interactive: bool) -> str: partial_notice_count=0, partial_notice_cap_announced=False, nocall_escalation=0, nocall_declined=False, tool_calls_made=0, writes_at_start=self._write_total(), + files_at_start={f["path"]: f["writes"] + for f in self.changed_files()}, start_turns=start_turns, unfounded_checked=False) @@ -581,6 +592,11 @@ def _loop(self, interactive: bool) -> str: if looks_like_tool_attempt(leftover, tool_names): if st.partial_notice_count < _MAX_TOOL_REPAIRS: st.partial_notice_count += 1 + if interactive: + _agent.print_warning( + " Part of the reply looked like a tool call but " + "could not be read, so it was not run. The model " + "was told.") result_blocks.append( "[tool-call format] Part of this response looked " "like another tool call, but it could not be " @@ -882,12 +898,14 @@ def _handle_no_tool_calls(self, response, interactive, st) -> tuple[bool, str]: print_warning(enforcement.strip()) if unverified.strip(): print_warning(unverified.strip()) - if sensitive.strip(): - print_warning(sensitive.strip()) + task_files = self._files_changed_this_task(st) + task_sensitive = self._sensitive_changes_notice(task_files) if task_files else "" + if task_sensitive.strip(): + print_warning(task_sensitive.strip()) if st.tool_calls_made > 0: print_task_footer( turns_used=max(1, self._turns - getattr(st, "start_turns", self._turns - 1)), - files_changed=[c["path"] for c in self.changed_files()], + files_changed=[c["path"] for c in task_files], verify_status=self._last_verify_state, ctx_ratio=self._fill_ratio(), ) @@ -1045,7 +1063,13 @@ def _grounding_footer(self) -> str: parts.append(f"verify: {self._last_verify_state}") return "\n\n[session record: " + "; ".join(parts) + "]" - def _sensitive_changes_notice(self) -> str: + def _files_changed_this_task(self, st) -> list[dict]: + """changed_files() entries written since this task started.""" + before = getattr(st, "files_at_start", None) or {} + return [f for f in self.changed_files() + if f.get("writes", 0) > before.get(f.get("path"), 0)] + + def _sensitive_changes_notice(self, files: list[dict] | None = None) -> str: """A footer noting files a passing check cannot vouch for on its own (a rewritten test, a loosened CI/lint gate), or ``""`` when none. @@ -1060,7 +1084,8 @@ def _sensitive_changes_notice(self) -> str: """ try: from ..review_guard import classify_sensitive_changes, render_warning - message = render_warning(classify_sensitive_changes(self.changed_files())) + message = render_warning(classify_sensitive_changes( + self.changed_files() if files is None else files)) except Exception as e: from localm.debuglog import logger logger.debug("sensitive-changes notice skipped: %r", e) # See test_classify_sensitive_changes_tolerates_malformed_entries. diff --git a/localm/plugins/coder/agent/session.py b/localm/plugins/coder/agent/session.py index 79a0a1d3c..c2614cb2d 100644 --- a/localm/plugins/coder/agent/session.py +++ b/localm/plugins/coder/agent/session.py @@ -27,8 +27,9 @@ class _SessionMixin: def reset(self) -> None: - """Clear conversation history.""" + """Clear conversation history. Allows answered with "a" end with it.""" self._messages = [] + self._session_allowed = set() self._turns = 0 self._total_tokens = 0 self._last_turn_tokens = 0 @@ -171,6 +172,8 @@ def set_cwd(self, cwd: Path) -> None: failure is logged, and the old checkpoint stays where it was. """ self._migrate_checkpoint(self.cwd, cwd) + # An "a" answer was given for the old project; the new one asks again. + self._session_allowed = set() load_memory = _agent.load_memory # live: honour a patched agent.load_memory load_custom_instructions = _agent.load_custom_instructions self.cwd = cwd diff --git a/localm/plugins/coder/cli/goal.py b/localm/plugins/coder/cli/goal.py index 5ecabed5a..069dcee44 100644 --- a/localm/plugins/coder/cli/goal.py +++ b/localm/plugins/coder/cli/goal.py @@ -38,7 +38,7 @@ def _goal_task_wrap(task: str, until_cmd: str) -> str: def _run_goal_loop(agent: Agent, task: str, until_cmd: str, max_iters: int, - work_dir: Path) -> tuple[bool, str]: + work_dir: Path, attachments: str = "") -> tuple[bool, str]: """Iterate: run the task, run the verify command, feed failures back, until it exits 0 or the iteration cap is hit. Returns (success, last_response). @@ -51,7 +51,11 @@ def _run_goal_loop(agent: Agent, task: str, until_cmd: str, max_iters: int, print_info( f"Goal mode: iterating on the task until `{until_cmd}` exits 0 " f"(max {max_iters} iteration(s)).") - response = agent.run_task(_goal_task_wrap(task, until_cmd)) + if attachments: + response = agent.run_task(_goal_task_wrap(task, until_cmd), + attachments=attachments) + else: + response = agent.run_task(_goal_task_wrap(task, until_cmd)) for attempt in range(1, max_iters + 1): outcome = _run_verify(until_cmd, work_dir) code, output = outcome diff --git a/localm/plugins/coder/cli/repl.py b/localm/plugins/coder/cli/repl.py index ea3a00cf3..bce76e20e 100644 --- a/localm/plugins/coder/cli/repl.py +++ b/localm/plugins/coder/cli/repl.py @@ -64,20 +64,54 @@ def _read_multiline() -> str: first = False else: lines.append(line) - pasted = 0 - while _stdin_has_buffered_data(): - extra = sys.stdin.readline() - if not extra: - break - lines.append(extra.rstrip("\r\n")) - pasted += 1 - if pasted >= 2: - print_info(f" (pasted {pasted + 1} lines)") + extra_lines = _drain_pasted_lines() + lines.extend(extra_lines) + if len(extra_lines) >= 2: + print_info(f" (pasted {len(extra_lines) + 1} lines)") break return "\n".join(lines) -def _expand_at_mentions(user_input: str, cwd: Path, scope_allows=None) -> str: +def _drain_with_readline() -> list[str]: + """Pasted lines still waiting in the console input buffer (Windows).""" + lines = [] + while _stdin_has_buffered_data(): + extra = sys.stdin.readline() + if not extra: + break + lines.append(extra.rstrip("\r\n")) + return lines + + +def _drain_fd() -> list[str]: + """Pasted lines still waiting on the terminal (POSIX), read straight from + the file descriptor: one read can return several pasted lines, and a + buffered readline would keep the rest where select() cannot see them. + See test_paste_into_a_real_pty_arrives_as_one_message.""" + try: + if not sys.stdin.isatty(): + return [] + import select + fd = sys.stdin.fileno() + chunks: list[bytes] = [] + while select.select([fd], [], [], 0.05 if chunks else 0.0)[0]: + data = os.read(fd, 65536) + if not data: + break + chunks.append(data) + except (OSError, ValueError): + return [] + text = b"".join(chunks).decode(sys.stdin.encoding or "utf-8", errors="replace") + return text.replace("\r\n", "\n").replace("\r", "\n").splitlines() + + +def _drain_pasted_lines() -> list[str]: + """Lines of a paste that arrived after the first one, read without waiting + for more typing.""" + return _drain_with_readline() if os.name == "nt" else _drain_fd() + + +def _at_attachments(user_input: str, cwd: Path, scope_allows=None) -> str: """Attach existing workspace files referenced as @path or @path:start-end. Each file is read through the read_file tool, so confinement, notebook @@ -85,7 +119,7 @@ def _expand_at_mentions(user_input: str, cwd: Path, scope_allows=None) -> str: read. A path outside *scope_allows* (the session's /scope) is not attached. Tokens that name no file (a decorator, an @handle) are left alone.""" if "@" not in user_input: - return user_input + return "" from ..tools.files import tool_read_file attachments: list[str] = [] @@ -123,8 +157,13 @@ def _expand_at_mentions(user_input: str, cwd: Path, scope_allows=None) -> str: f'\n{result.output}\n') if not attachments: - return user_input - return user_input + "\n\n" + "\n\n".join(attachments) + return "" + return "\n\n" + "\n\n".join(attachments) + + +def _expand_at_mentions(user_input: str, cwd: Path, scope_allows=None) -> str: + """*user_input* with its @file attachments appended.""" + return user_input + _at_attachments(user_input, cwd, scope_allows) # Every REPL slash command, for tab completion and /help @@ -248,7 +287,7 @@ def _repl(agent: Agent) -> None: break continue - expanded_input = _expand_at_mentions( + attachments = _at_attachments( user_input, agent.cwd, scope_allows=agent._scope_allows if agent.scope else None) @@ -259,12 +298,20 @@ def _repl(agent: Agent) -> None: if agent.goal_cmd is not None: before = {f["path"]: f["writes"] for f in agent.changed_files()} try: - _run_goal_loop(agent, expanded_input, agent.goal_cmd, - agent.goal_max_iters, agent.cwd) + if attachments: + _run_goal_loop(agent, user_input, agent.goal_cmd, + agent.goal_max_iters, agent.cwd, + attachments=attachments) + else: + _run_goal_loop(agent, user_input, agent.goal_cmd, + agent.goal_max_iters, agent.cwd) finally: _warn_goal_run_sensitive_changes(agent, before) else: - agent.chat(expanded_input) + if attachments: + agent.chat(user_input, attachments=attachments) + else: + agent.chat(user_input) except KeyboardInterrupt: # _loop has already dealt with the checkpoint; swallow the # interrupt here. diff --git a/localm/plugins/coder/display.py b/localm/plugins/coder/display.py index c9f3d192a..73639b44c 100644 --- a/localm/plugins/coder/display.py +++ b/localm/plugins/coder/display.py @@ -7,6 +7,8 @@ from __future__ import annotations +import contextlib +import contextvars import difflib import re import threading @@ -28,6 +30,8 @@ _last_diff_preview: tuple[str, str, str, bool] | None = None _last_confirm_always: bool = False _last_reject_reason: str = "" +_offer_session_allow: contextvars.ContextVar[bool] = contextvars.ContextVar( + "coder_offer_session_allow", default=False) def consume_confirm_always() -> bool: @@ -605,32 +609,64 @@ def _parse_confirm_answer(raw: str) -> tuple[str, str]: return "no", s +def reset_diff_preview() -> None: + """Forget the stored preview so a later prompt cannot offer [f]ull for it.""" + global _last_diff_preview + _last_diff_preview = None + + +@contextlib.contextmanager +def offering_session_allow(offer: bool): + """Within the block, approval prompts offer the "a" (allow this session) + answer only when *offer* is true. Prompts outside any block never offer it.""" + token = _offer_session_allow.set(bool(offer)) + try: + yield + finally: + _offer_session_allow.reset(token) + + +def _answer_hint(full: bool = False) -> str: + parts = ["y/N"] + if _offer_session_allow.get(): + parts.append("a=allow this session") + if full: + parts.append("f=full diff") + return "[" + "/".join(parts) + "] " + + +def _record_answer(action: str, reason: str) -> bool: + global _last_confirm_always, _last_reject_reason + if action == "always": + _last_confirm_always = _offer_session_allow.get() + return True + if action == "yes": + return True + _last_reject_reason = reason + return False + + def confirm_diff(path_label: str) -> bool: - """Prompt to apply changes to *path_label* with support for [y/N/a/f].""" + """Prompt to apply changes to *path_label*. [f] reprints the full diff, and + is offered only when the stored preview is this file's and was cut short.""" global _last_confirm_always, _last_reject_reason _last_confirm_always = False _last_reject_reason = "" try: while True: - has_full = bool(_last_diff_preview and _last_diff_preview[3]) - opts = ("[y/N/a=allow this session/f=full diff] " if has_full - else "[y/N/a=allow this session] ") + preview = _last_diff_preview + has_full = bool(preview and preview[3] and preview[2] == path_label) prompt = Text("Apply changes to ", style="yellow") prompt.append(_strip_esc(path_label), style="bold yellow") - prompt.append(f"? {opts}", style="yellow") + prompt.append(f"? {_answer_hint(full=has_full)}", style="yellow") raw = console.input(prompt) action, reason = _parse_confirm_answer(raw) - if action == "full" and _last_diff_preview is not None: - old_t, new_t, lbl, _ = _last_diff_preview - print_diff_preview(old_t, new_t, path_label=lbl, max_lines=100_000) + if action == "full": + if has_full and preview is not None: + old_t, new_t, lbl, _ = preview + print_diff_preview(old_t, new_t, path_label=lbl, max_lines=100_000) continue - if action == "always": - _last_confirm_always = True - return True - if action == "yes": - return True - _last_reject_reason = reason - return False + return _record_answer(action, reason) except (KeyboardInterrupt, EOFError): return False @@ -644,16 +680,10 @@ def confirm(prompt: str) -> bool: _last_confirm_always = False _last_reject_reason = "" try: - line = _sanitized_text(f"{prompt} [y/N/a=allow this session] ", style="yellow") + line = _sanitized_text(f"{prompt} {_answer_hint()}", style="yellow") raw = console.input(line) action, reason = _parse_confirm_answer(raw) - if action == "always": - _last_confirm_always = True - return True - if action == "yes": - return True - _last_reject_reason = reason - return False + return _record_answer(action, reason) except (KeyboardInterrupt, EOFError): return False diff --git a/tests/plugins/coder/test_coder_ui_ux.py b/tests/plugins/coder/test_coder_ui_ux.py index f0e1c313c..fefc47dd6 100644 --- a/tests/plugins/coder/test_coder_ui_ux.py +++ b/tests/plugins/coder/test_coder_ui_ux.py @@ -7,11 +7,18 @@ from __future__ import annotations import io +import os +import sys +import time from pathlib import Path +from types import SimpleNamespace from unittest.mock import MagicMock, patch +import pytest + from localm.console import SafeConsole from localm.plugins.coder import display +from localm.plugins.coder.cli import repl as repl_mod from localm.plugins.coder.cli.repl import _expand_at_mentions, _read_multiline from localm.plugins.coder.parser import ToolCall from localm.plugins.coder.tools import ToolResult @@ -252,6 +259,28 @@ def test_bare_enter_and_n_reject_without_a_reason(self): assert display._parse_confirm_answer(" y ") == ("yes", "") +class TestAttachmentsStayOutOfTheRequestText: + def test_attachment_text_does_not_turn_a_question_into_an_action(self, tmp_path): + agent = _make_agent(tmp_path, auto_approve=True) + (tmp_path / "notes.py").write_text("def add(a, b):\n return a + b\n", encoding="utf-8") + calls = [] + + def fake_stream(messages, on_reasoning=None, **kw): + calls.append(messages) + yield "It returns the sum of a and b." + + agent.backend.chat_stream.side_effect = fake_stream + from localm.plugins.coder.cli.repl import _at_attachments + question = "what does @notes.py do" + agent.chat(question, attachments=_at_attachments(question, tmp_path)) + + assert len(calls) == 1, "a prose answer to a question must not be re-prompted" + sent = calls[0][-1]["content"] + assert "return a + b" in sent + assert agent._last_user_request == question + assert agent._session_title == question + + class TestContextWarningText: def test_compact_warning_carries_no_literal_markup(self, tmp_path): agent = _make_agent(tmp_path) @@ -348,6 +377,240 @@ def has_data(): monkeypatch.setattr("localm.plugins.coder.cli.repl._stdin_has_buffered_data", has_data) monkeypatch.setattr("sys.stdin.readline", lambda: next(buffered)) + monkeypatch.setattr("localm.plugins.coder.cli.repl._drain_pasted_lines", + repl_mod._drain_with_readline) msg = _read_multiline() assert msg == "first line\nsecond line\nthird line" + + @pytest.mark.skipif(os.name == "nt", reason="needs a POSIX pty") + def test_paste_into_a_real_pty_arrives_as_one_message(self, tmp_path): + import pty + import select + root = str(Path(__file__).resolve().parents[3]) + script = ( + "import sys; sys.path.insert(0, %r)\n" + "import readline\n" + "from localm.plugins.coder.cli.repl import _read_multiline\n" + "print('GOT=' + repr(_read_multiline()), flush=True)\n" + "print('NEXT=' + repr(_read_multiline()), flush=True)\n" % root) + env = dict(os.environ, TERM="xterm", LOCALM_HOME=str(tmp_path / "home")) + pid, fd = pty.fork() + if pid == 0: + os.execve(sys.executable, [sys.executable, "-c", script], env) + out = b"" + + def read_until(token: bytes, timeout: float = 20.0) -> None: + nonlocal out + end = time.monotonic() + timeout + while token not in out and time.monotonic() < end: + if select.select([fd], [], [], 0.2)[0]: + try: + out += os.read(fd, 4096) + except OSError: + break + assert token in out, out[-2000:] + + try: + read_until(b"You") + os.write(fd, b"l1\nl2\nl3\nl4\n") + read_until(b"GOT=") + read_until(b"You", timeout=10.0) + os.write(fd, b"after\n") + read_until(b"NEXT=") + read_until(b"\n", timeout=5.0) + finally: + try: + os.kill(pid, 9) + except OSError: + pass + os.waitpid(pid, 0) + os.close(fd) + text = out.decode("utf-8", "replace") + got = text.split("GOT=", 1)[1].splitlines()[0] + nxt = text.split("NEXT=", 1)[1].splitlines()[0] + assert got.strip() == repr("l1\nl2\nl3\nl4") + assert nxt.strip() == repr("after") + + +class TestReviewFindings: + def test_full_diff_is_offered_only_for_this_files_preview(self, monkeypatch): + big = "".join(f"line_{i} = {i}\n" for i in range(80)) + prompts, printed = [], [] + answers = iter(["f", "n"]) + + def fake_input(prompt=""): + prompts.append(str(prompt)) + return next(answers) + + monkeypatch.setattr(display.console, "input", fake_input) + with patch("localm.plugins.coder.display.Syntax", + side_effect=lambda text, *a, **kw: printed.append(text) or text), \ + patch.object(display.console, "print"): + display.print_diff_preview("", big, "a.py") + assert display.confirm_diff("b.py") is False + assert len(printed) == 1, "f must not print another file's diff" + assert all("f=full diff" not in p for p in prompts) + + def test_confirm_tool_forgets_an_earlier_files_preview(self, tmp_path, monkeypatch): + agent = _make_agent(tmp_path, auto_approve=False) + big = "".join(f"line_{i} = {i}\n" for i in range(80)) + with patch.object(display.console, "print"): + display.print_diff_preview("", big, "a.py") + prompts = [] + monkeypatch.setattr(display.console, "input", + lambda prompt="": prompts.append(str(prompt)) or "n") + call = ToolCall(name="patch_file", args={"path": "b.py", "diff": "--- a\n+++ b\n"}, + raw="", start=0, end=0) + with patch.object(display.console, "print"): + agent._execute_tool(call, interactive=True) + assert prompts and "f=full diff" not in prompts[0] + assert display._last_diff_preview is None + + def test_rejection_event_summary_is_the_literal_the_gui_matches(self, tmp_path, monkeypatch): + agent = _make_agent(tmp_path, auto_approve=False) + events = [] + agent.on_event = events.append + agent.confirm_handler = lambda call: False + call = ToolCall(name="write_file", args={"path": "x.py", "content": "x = 1\n"}, + raw="", start=0, end=0) + agent._execute_tool(call, interactive=False) + summaries = [e.get("summary") for e in events if e.get("type") == "tool_result"] + assert summaries == ["rejected by user"] + + def test_child_prompt_through_parent_terminal_honours_a_and_reason(self, tmp_path, monkeypatch): + child = _make_agent(tmp_path, auto_approve=False) + prompts = [] + answers = iter(["a", "use src/ instead"]) + + def fake_input(prompt=""): + prompts.append(str(prompt)) + return next(answers) + + monkeypatch.setattr(display.console, "input", fake_input) + child.confirm_handler = lambda call: display.confirm(f" Allow {call.name}?") + w1 = ToolCall(name="write_file", args={"path": "a.py", "content": "a = 1\n"}, + raw="", start=0, end=0) + w2 = ToolCall(name="write_file", args={"path": "b.py", "content": "b = 1\n"}, + raw="", start=0, end=0) + assert child._execute_tool(w1, interactive=False).ok is True + assert "a=allow this session" in prompts[0] + assert child._execute_tool(w2, interactive=False).ok is True + assert len(prompts) == 1 + sh = ToolCall(name="run_shell", args={"command": "echo hi"}, raw="", start=0, end=0) + res = child._execute_tool(sh, interactive=False) + assert res.ok is False and "Rejected by user: use src/ instead" in res.output + + def test_plain_questions_do_not_offer_session_allow(self, monkeypatch): + prompts = [] + monkeypatch.setattr(display.console, "input", + lambda prompt="": prompts.append(str(prompt)) or "a") + assert display.confirm(" Keep going?") is True + assert "a=allow" not in prompts[0] + assert display.consume_confirm_always() is False + + def test_session_allows_end_on_cd_and_clear(self, tmp_path, monkeypatch): + agent = _make_agent(tmp_path, auto_approve=False) + agent._session_allowed.add("write_file") + other = tmp_path / "other" + other.mkdir() + with patch("localm.plugins.coder.agent.ProjectMap") as pm, \ + patch("localm.plugins.coder.agent.load_memory", return_value=""): + pm.build.return_value.file_count.return_value = 0 + agent.set_cwd(other) + assert agent._session_allowed == set() + agent._session_allowed.add("write_file") + agent.reset() + assert agent._session_allowed == set() + + def test_url_tools_are_allowed_per_host(self): + from localm.plugins.coder.agent.execution import _ExecutionMixin as M + + def key(url): + return M._session_allow_key( + ToolCall(name="fetch_url", args={"url": url}, raw="", start=0, end=0)) + + assert key("https://docs.python.org/3/") == key("https://docs.python.org/x") + assert key("https://docs.python.org/3/") != key("https://evil.example/") + assert key("not a url") is None + + def test_spinner_skips_tools_that_can_prompt_on_the_terminal(self): + from localm.plugins.coder.agent.execution import _ExecutionMixin as M + for name in ("run_shell", "run_shell_background", "git_push", "git_commit", + "run_tests", "spawn_agent", "dispatch_parallel"): + assert M._spins_while_running(name) is False, name + for name in ("read_file", "write_file", "search_files", "fetch_url"): + assert M._spins_while_running(name) is True, name + + def test_footer_and_warning_cover_only_this_tasks_files(self, tmp_path): + agent = _make_agent(tmp_path) + rows = [{"path": "tests/test_a.py", "writes": 2}, {"path": "src/b.py", "writes": 1}] + agent.changed_files = lambda: rows + st = SimpleNamespace(files_at_start={"tests/test_a.py": 2}) + assert [f["path"] for f in agent._files_changed_this_task(st)] == ["src/b.py"] + + def test_unreadable_tool_call_is_shown_to_the_user(self, tmp_path): + agent = _make_agent(tmp_path, auto_approve=True) + (tmp_path / "a.py").write_text("x = 1\n", encoding="utf-8") + replies = [ + 'Reading.\n\n{"name": "read_file", "args": {"path": "a.py"}}\n\n' + 'and then\n\n{"name": "write_file", "args": {"path": "b.py", "content": "oops\n\nDone.', + "Finished.", + ] + + def fake_stream(messages, on_reasoning=None, **kw): + yield replies.pop(0) if replies else "Finished." + + agent.backend.chat_stream.side_effect = fake_stream + warnings = [] + with patch("localm.plugins.coder.agent.print_warning", side_effect=warnings.append): + agent.chat("read a.py") + assert any("could not be read" in w for w in warnings) + + def test_ordinary_code_fence_is_not_reported_as_a_tool_call(self, tmp_path): + agent = _make_agent(tmp_path) + reply = ("Here is the config:\n```\n{\"settings\": {\n" + + "".join(f' "key_{i}": {i},\n' for i in range(30)) + + ' "last": 0}}\n```\nThat is all.') + chunks = [reply[i:i + 10] for i in range(0, len(reply), 10)] + agent.backend.chat_stream.side_effect = lambda messages, on_reasoning=None, **kw: iter(chunks) + from localm.plugins.coder.agent import context + progress = [] + with patch.object(context._TerminalReply, "tool_progress", + lambda self, label, detail="": progress.append(label)), \ + patch.object(context, "print_streaming_token"), \ + patch.object(context, "print_thinking"), \ + patch.object(context, "print_assistant_label"), \ + patch.object(context, "print_streaming_done"): + agent._call_llm([{"role": "user", "content": "show code"}], interactive=True) + assert progress == [] + + def test_compaction_spinner_only_in_the_interactive_session(self, tmp_path): + agent = _make_agent(tmp_path) + agent._messages = [{"role": "user" if i % 2 == 0 else "assistant", + "content": f"message {i} " * 20} for i in range(12)] + from localm.plugins.coder.agent import context + for interactive, expected in ((False, 0), (True, 1)): + agent._interactive = interactive + with patch.object(context, "ActivitySpinner") as spin, \ + patch("localm.inference.compact.summarise_in_passes", + return_value=("summary", None)), \ + patch.object(context, "print_info"): + agent._compact_history() + assert spin.call_count == expected, interactive + + def test_goal_mode_keeps_attachments_out_of_the_request(self, tmp_path, monkeypatch): + from localm.plugins.coder.cli import goal + agent = MagicMock() + seen = {} + + def run_task(task, attachments=""): + seen["task"], seen["attachments"] = task, attachments + return "done" + + agent.run_task.side_effect = run_task + monkeypatch.setattr(goal._cli, "_run_verify", lambda cmd, cwd: (0, "ok"), raising=False) + goal._run_goal_loop(agent, "what is @a.py", "true", 1, tmp_path, + attachments="\n\nbody") + assert "body" not in seen["task"] + assert "body" in seen["attachments"] From e4582e7e633a381276ba58efebf10990a97908b2 Mon Sep 17 00:00:00 2001 From: Matlan Date: Tue, 20 Oct 2026 15:40:13 +0200 Subject: [PATCH 5/5] Coder REPL: pasted lines join chat messages only, never a /command A paste that starts with "/export notes.md" used to append the pasted lines to the command's argument. The reader now returns the typed text and the pasted lines separately: a /command gets only what was typed (and says the pasted lines were not used), a chat message gets all of it, and @file mentions are read from the typed text only. Co-Authored-By: Claude Opus 5.5 --- localm/plugins/coder/cli/repl.py | 22 +++++++++--- tests/plugins/coder/test_coder_ui_ux.py | 34 +++++++++++++++++++ ...st_resume_keeps_checkpoint_until_a_turn.py | 2 +- 3 files changed, 53 insertions(+), 5 deletions(-) diff --git a/localm/plugins/coder/cli/repl.py b/localm/plugins/coder/cli/repl.py index bce76e20e..7113c0ac5 100644 --- a/localm/plugins/coder/cli/repl.py +++ b/localm/plugins/coder/cli/repl.py @@ -51,7 +51,18 @@ def _read_multiline() -> str: Read one user message, supporting backslash line continuation and coalescing multi-line terminal pastes into a single message. """ + typed, pasted = _read_input() + return "\n".join([typed, *pasted]) + + +def _read_input() -> tuple[str, list[str]]: + """Read one message as (typed text, pasted lines that followed it). + + The typed part is what the prompt returned, with backslash continuations + joined. Pasted lines arrive in the same read and are kept apart so a + /command is never built from them.""" lines: list[str] = [] + extra_lines: list[str] = [] first = True while True: prompt = "\n[bold green]You[/bold green]: " if first else "[dim]...[/dim] " @@ -65,11 +76,10 @@ def _read_multiline() -> str: else: lines.append(line) extra_lines = _drain_pasted_lines() - lines.extend(extra_lines) if len(extra_lines) >= 2: print_info(f" (pasted {len(extra_lines) + 1} lines)") break - return "\n".join(lines) + return "\n".join(lines), extra_lines def _drain_with_readline() -> list[str]: @@ -269,12 +279,13 @@ def _repl(agent: Agent) -> None: _setup_readline(agent) while True: try: - user_input = _read_multiline().strip() + typed, pasted = _read_input() except (KeyboardInterrupt, EOFError): console.print("\n[dim]Bye.[/dim]") break - if not user_input: + user_input = typed.strip() + if not user_input and not any(p.strip() for p in pasted): continue if user_input.lower() in ("exit", "quit", "q", "bye"): @@ -282,6 +293,8 @@ def _repl(agent: Agent) -> None: break if user_input.startswith("/"): + if pasted: + print_info(" Lines pasted after a /command were not used.") stop = _handle_command(user_input, agent) if stop: break @@ -290,6 +303,7 @@ def _repl(agent: Agent) -> None: attachments = _at_attachments( user_input, agent.cwd, scope_allows=agent._scope_allows if agent.scope else None) + user_input = "\n".join([typed, *pasted]).strip() try: # The saved checkpoint is not cleared before the turn; the turn diff --git a/tests/plugins/coder/test_coder_ui_ux.py b/tests/plugins/coder/test_coder_ui_ux.py index fefc47dd6..8bb098190 100644 --- a/tests/plugins/coder/test_coder_ui_ux.py +++ b/tests/plugins/coder/test_coder_ui_ux.py @@ -433,6 +433,40 @@ def read_until(token: bytes, timeout: float = 20.0) -> None: assert nxt.strip() == repr("after") +class TestPastedLinesNeverBuildACommand: + def _drive(self, monkeypatch, agent, reads): + feed = iter(reads) + + def fake_read_input(): + nxt = next(feed) + if isinstance(nxt, BaseException): + raise nxt + return nxt + + monkeypatch.setattr(repl_mod, "_read_input", fake_read_input) + monkeypatch.setattr(repl_mod, "_setup_readline", lambda agent: None) + repl_mod._repl(agent) + + def test_lines_pasted_after_a_command_are_not_passed_to_it(self, tmp_path, monkeypatch): + agent = _make_agent(tmp_path) + commands, infos = [], [] + monkeypatch.setattr(repl_mod, "_handle_command", + lambda text, agent: commands.append(text) or False) + monkeypatch.setattr(repl_mod, "print_info", infos.append) + self._drive(monkeypatch, agent, + [("/export notes.md", ["second line", "third line"]), EOFError()]) + assert commands == ["/export notes.md"] + assert any("not used" in i for i in infos) + + def test_pasted_lines_join_a_chat_message(self, tmp_path, monkeypatch): + agent = _make_agent(tmp_path) + sent = [] + monkeypatch.setattr(agent, "chat", lambda msg: sent.append(msg) or "ok") + self._drive(monkeypatch, agent, + [("Traceback (most recent call last):", [" File x", "Error"]), EOFError()]) + assert sent == ["Traceback (most recent call last):\n File x\nError"] + + class TestReviewFindings: def test_full_diff_is_offered_only_for_this_files_preview(self, monkeypatch): big = "".join(f"line_{i} = {i}\n" for i in range(80)) diff --git a/tests/plugins/coder/test_resume_keeps_checkpoint_until_a_turn.py b/tests/plugins/coder/test_resume_keeps_checkpoint_until_a_turn.py index f20bd915e..b74128e6d 100644 --- a/tests/plugins/coder/test_resume_keeps_checkpoint_until_a_turn.py +++ b/tests/plugins/coder/test_resume_keeps_checkpoint_until_a_turn.py @@ -149,7 +149,7 @@ def test_ctrl_c_at_the_first_prompt_after_resume_keeps_the_session( def _ctrl_c(): raise KeyboardInterrupt - monkeypatch.setattr(repl_mod, "_read_multiline", _ctrl_c) + monkeypatch.setattr(repl_mod, "_read_input", _ctrl_c) result, backend = _cli(proj, monkeypatch, "--resume")