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/CHANGELOG.md b/CHANGELOG.md index 0606fc7e9..4a07cbe4c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,7 +12,13 @@ permanent public record of what shipped and are never rewritten; the in-progress ## [Unreleased] ### Added - +- **`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. - **Pause, resume and cancel a model download in the GUI.** The Add a model card has Pause (stops and keeps what is saved, Resume continues from it) and Cancel (stops and deletes the partial download), the download line shows the steady speed and time left, and the Activity list has a @@ -46,6 +52,9 @@ permanent public record of what shipped and are never rewritten; the in-progress the finished ones. ### Fixed +- **`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. - **The Hugging Face token saved in Settings is used for every download.** Model pulls, projector and embedding-model downloads, speech-model downloads and the model-type lookup now send it, so diff --git a/localm/plugins/coder/agent/context.py b/localm/plugins/coder/agent/context.py index 55f80cf83..ee86617e1 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, spinner_available, ) 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, "_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) + else: + summary, failure = summarise_in_passes( + groups, _pass, reply_tokens=reply_tokens, measure=measure) if summary: body = compose("[Session summary]\n", summary) else: @@ -586,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: @@ -948,7 +994,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 @@ -977,13 +1024,40 @@ 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):] + tagged = "= 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 @@ -1008,6 +1082,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: @@ -1023,25 +1098,34 @@ 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() 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/core.py b/localm/plugins/coder/agent/core.py index dcd0deffd..635dae338 100644 --- a/localm/plugins/coder/agent/core.py +++ b/localm/plugins/coder/agent/core.py @@ -354,6 +354,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 b82dea0f0..a38678291 100644 --- a/localm/plugins/coder/agent/execution.py +++ b/localm/plugins/coder/agent/execution.py @@ -16,8 +16,10 @@ import localm.plugins.coder.agent as _agent from ..display import ( - console, print_progress_tool_call, print_tool_call, print_tool_error, - print_tool_result, tool_call_text, + ActivitySpinner, console, consume_confirm_always, consume_reject_reason, + offering_session_allow, print_info, print_progress_tool_call, + print_tool_call, print_tool_error, print_tool_result, reset_diff_preview, + tool_call_text, ) from ..diffutil import ( compute_multifile_diff, compute_search_replace_diff, compute_tool_diff, @@ -579,12 +581,34 @@ 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 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: + 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 @@ -616,9 +640,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() + 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) + # 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="rejected by user") return result @@ -698,7 +729,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 self._spins_while_running(call.name): + 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: @@ -715,7 +750,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) @@ -897,6 +935,46 @@ 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, 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 = 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 + + @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: """ Ask the user to approve a destructive tool call. @@ -910,6 +988,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 7bc17ec70..bda18047b 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 @@ -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: @@ -387,13 +396,11 @@ 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). + files_at_start={f["path"]: f["writes"] + for f in self.changed_files()}, + start_turns=start_turns, unfounded_checked=False) try: @@ -548,9 +555,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())) @@ -585,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 " @@ -880,7 +892,24 @@ 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()) + 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 task_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) @@ -1034,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. @@ -1049,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. @@ -1083,7 +1119,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 +1174,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 +1394,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 +1407,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 +1437,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/agent/session.py b/localm/plugins/coder/agent/session.py index 7fe6f403e..bfb0f8ecb 100644 --- a/localm/plugins/coder/agent/session.py +++ b/localm/plugins/coder/agent/session.py @@ -30,8 +30,9 @@ class _SessionMixin: _activated_lock: threading.Lock 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 @@ -178,6 +179,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/_main.py b/localm/plugins/coder/cli/_main.py index 5f191d31d..ce70310ed 100644 --- a/localm/plugins/coder/cli/_main.py +++ b/localm/plugins/coder/cli/_main.py @@ -941,15 +941,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/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 2617b7621..f67e769d0 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,14 +26,43 @@ from ..review_guard import classify_sensitive_changes, render_warning from .goal import _run_goal_loop +_AT_MENTION_RE = re.compile( + r"(?:^|(?<=\s))@([A-Za-z0-9_./\\-]+(?::\d+(?:-\d+)?)?)(?=\s|$|[.,;!?])" +) + + +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. """ + 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] " @@ -44,8 +75,105 @@ def _read_multiline() -> str: first = False else: lines.append(line) + extra_lines = _drain_pasted_lines() + if len(extra_lines) >= 2: + print_info(f" (pasted {len(extra_lines) + 1} lines)") + break + return "\n".join(lines), extra_lines + + +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 - return "\n".join(lines) + 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 + 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 "" + 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).rstrip(".") + if not spec or spec in seen: + continue + seen.add(spec) + path_part, offset, limit = spec, 0, 0 + if ":" in spec: + base, rng = spec.rsplit(":", 1) + m = re.fullmatch(r"(\d+)(?:-(\d+))?", rng) + if m: + path_part = base + 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: + if not candidate.is_file(): + continue + except OSError: + continue + 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 + result = tool_read_file(cwd, path_part, offset, limit) + if not result.ok: + print_info(f" @{spec} not attached: {result.output}") + continue + 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 "" + 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 @@ -85,16 +213,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 +266,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) @@ -148,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"): @@ -161,11 +293,18 @@ 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 continue + 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 # itself clears or rewrites it. See @@ -173,12 +312,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, user_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(user_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 517edb398..33cb221af 100644 --- a/localm/plugins/coder/display.py +++ b/localm/plugins/coder/display.py @@ -7,7 +7,12 @@ from __future__ import annotations +import contextlib +import contextvars import difflib +import re +import threading +import time from pathlib import Path from localm.console import SafeConsole as Console, strip_controls @@ -20,6 +25,146 @@ 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 = "" +_offer_session_allow: contextvars.ContextVar[bool] = contextvars.ContextVar( + "coder_offer_session_allow", default=False) + + +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 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 @@ -72,7 +217,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 +258,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 +384,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 +394,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 +439,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 +457,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 +512,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 +534,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 +550,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 +562,111 @@ 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() + # Any other text is a rejection that tells the agent what to do instead. + 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 'Apply changes to ? [y/N]' and return True on yes.""" + """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: - # 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: + 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"? {_answer_hint(full=has_full)}", style="yellow") + raw = console.input(prompt) + action, reason = _parse_confirm_answer(raw) + 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 + return _record_answer(action, reason) except (KeyboardInterrupt, EOFError): return False @@ -339,10 +676,14 @@ 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} {_answer_hint()}", style="yellow") + raw = console.input(line) + action, reason = _parse_confirm_answer(raw) + 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 new file mode 100644 index 000000000..8bb098190 --- /dev/null +++ b/tests/plugins/coder/test_coder_ui_ux.py @@ -0,0 +1,650 @@ +# 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 +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 +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_session_allow_covers_only_that_tool(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) + 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 + + 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) + 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 + + + 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 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) + 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( + 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)) + 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 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)) + 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"] 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")