fix: address CodeRabbit review findings on the security-remediation diff

Critical permission-classifier bypasses:
- sudo long value-options (`sudo --user alice rm -rf /`) were not consumed, so
  the wrapped destructive payload classified as the option's value. Add the long
  forms to _SUDO_VALUE_OPTS.
- uv global options before `run` (`uv --directory repo run rm -rf /`) hid the
  subcommand; the global flag's value was mistaken for it. Add
  _uv_strip_global_opts and apply it in both mutation and destructive paths.

Major:
- file_restore: treat missing (None) or malformed-base64 content as a corrupt
  restore point instead of silently writing an empty/garbage file.
- web/fetch: _ip_is_blocked now fails closed (blocks) on an unparseable address.
- scratchpad: stop unlinking the advisory lock file (split-inode race); keep it
  persistent and add *.scratchpad.lock to the written .gitignore patterns.
- soul shutdown: don't re-await an already-finished task in the cleanup loop —
  retrieve its exception without re-raising so the rest of shutdown still runs.
- pythinkersoul: on mid-tool interruption, keep the real results of calls that
  already completed (captured via on_tool_result) and only synthesize the
  interruption marker for still-pending calls.

Minor / nitpick:
- /import arg parsing now uses shlex via a shared parse_import_args helper
  (soul/slash + ui/shell/export_import), preserving quoted paths.
- web/runner: offload the blocking wire-file stat with asyncio.to_thread;
  document the intentional broad except at the per-message dispatch boundary.
- cli/vis: rename unused callback param to _ctx.
- tests: regression cases for both bypasses; strengthened import token-count
  assertions; lock-file persistence test; minor annotations.

Author: elkaix

src/pythinker_code/cli/vis.py

@@ -12,7 +12,7 @@
 
 @cli.callback(invoke_without_command=True)
 def vis(
-    ctx: typer.Context,
+    _ctx: typer.Context,
     host: Annotated[
         str | None,
         typer.Option("--host", "-H", help="Bind to specific IP address"),

src/pythinker_code/file_restore.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import base64
+import binascii
 import re
 import time
 import uuid
@@ -92,6 +93,16 @@ def restore_file_restore_point(session: Session, restore_id: str) -> FileRestore
     if not point.existed:
         point.path.unlink(missing_ok=True)
         return point
+    # An existed-file restore must carry its content. Missing content (None) or
+    # malformed base64 is a corrupt restore point — fail loudly instead of
+    # silently writing an empty/garbage file. An empty string is a legitimately
+    # empty file and decodes to b"".
+    if point.content_b64 is None:
+        raise FileNotFoundError(f"Corrupt restore point: {restore_id}")
+    try:
+        content = base64.b64decode(point.content_b64, validate=True)
+    except (binascii.Error, ValueError) as exc:
+        raise FileNotFoundError(f"Corrupt restore point: {restore_id}") from exc
     point.path.parent.mkdir(parents=True, exist_ok=True)
-    point.path.write_bytes(base64.b64decode(point.content_b64 or ""))
+    point.path.write_bytes(content)
     return point

src/pythinker_code/scratchpad.py

@@ -34,7 +34,14 @@
 
 # Patterns written to the project .gitignore when the agent starts in a git repo.
 # These directories are local-only agent state and must never be committed.
-_GITIGNORE_ENTRIES = (".pythinker/", ".pythinker-review/", ".pythinker-review-flow/")
+_GITIGNORE_ENTRIES = (
+    ".pythinker/",
+    ".pythinker-review/",
+    ".pythinker-review-flow/",
+    # The advisory lock file is kept on disk for correct flock coordination, so
+    # ignore it rather than letting it dirty the project's working tree.
+    "*.scratchpad.lock",
+)
 _GITIGNORE_SECTION_HEADER = "# pythinker — local agent state (do not commit)"
 
 StatusReason = Literal[
@@ -434,9 +441,11 @@ def _exclude_lock(path: Path) -> Generator[None]:
                 with contextlib.suppress(OSError):
                     fcntl.flock(fh.fileno(), fcntl.LOCK_UN)
     finally:
+        # Keep the lock file on disk: unlinking it lets a concurrent writer
+        # create a fresh inode and acquire a second, independent flock, so the
+        # two writers would no longer be serialized. Releasing the flock (above)
+        # and closing the handle is enough.
         fh.close()
-        with contextlib.suppress(OSError):
-            lock_file.unlink()
 
 
 async def _append_gitignore_entries(work_dir: HostPath) -> None:

src/pythinker_code/soul/__init__.py

@@ -238,10 +238,19 @@ async def run_soul(
             soul_task.result()  # this will raise if any exception was raised in the run task
     finally:
         for task in (soul_task, cancel_event_task, notification_task):
-            if task is not None:
-                task.cancel()
-                with contextlib.suppress(asyncio.CancelledError):
-                    await task
+            if task is None:
+                continue
+            if task.done():
+                # Already finished (e.g. soul_task raised and was surfaced above).
+                # Retrieve any exception so it isn't flagged "never retrieved", but
+                # do not re-await/re-raise — that would abort the rest of shutdown
+                # (notification flush, wire.shutdown/join) and leak UI resources.
+                if not task.cancelled():
+                    task.exception()
+                continue
+            task.cancel()
+            with contextlib.suppress(asyncio.CancelledError):
+                await task
         try:
             await _deliver_notifications_to_wire_once(runtime, wire)
         except Exception:

src/pythinker_code/soul/permission.py

@@ -212,7 +212,34 @@ class PermissionProfile:
 _GIT_NETWORK = {"clone", "fetch", "ls-remote"}
 _WRAPPER_COMMANDS = {"command", "env", "nohup", "sudo", "time"}
 # sudo options that consume a following separate word (the value is NOT the command).
-_SUDO_VALUE_OPTS = {"-u", "-g", "-U", "-C", "-p", "-r", "-t", "-T", "-h", "-R", "-D"}
+_SUDO_VALUE_OPTS = {
+    # Short value-taking options.
+    "-u",
+    "-g",
+    "-U",
+    "-C",
+    "-p",
+    "-r",
+    "-t",
+    "-T",
+    "-h",
+    "-R",
+    "-D",
+    # Long forms with a space-separated value (e.g. ``sudo --user alice rm``).
+    # The ``--opt=value`` form carries its value inline, so it is consumed as a
+    # single token and needs no entry here.
+    "--user",
+    "--group",
+    "--other-user",
+    "--close-from",
+    "--prompt",
+    "--role",
+    "--type",
+    "--command-timeout",
+    "--host",
+    "--chroot",
+    "--chdir",
+}
 # GNU time options that consume a following separate word.
 _TIME_VALUE_OPTS = {"-o", "-f", "--output", "--format"}
 
@@ -463,10 +490,11 @@ def _segment_mutation_reason(tokens: list[str]) -> str | None:
         if subcommand in _GIT_NETWORK:
             return f"network access via git {subcommand}"
     if base == "uv":
-        run_payload = _uv_run_payload(args)
+        uv_args = _uv_strip_global_opts(args)
+        run_payload = _uv_run_payload(uv_args)
         if run_payload and (r := _segment_mutation_reason(run_payload)):
             return f"uv run: {r}"
-        nonopts = [a for a in args if not a.startswith("-")]
+        nonopts = [a for a in uv_args if not a.startswith("-")]
         if nonopts:
             head = nonopts[0]
             sub = nonopts[1] if (head in _UV_SUBNAMESPACES and len(nonopts) > 1) else head
@@ -702,6 +730,28 @@ def _xargs_payload(args: list[str]) -> list[str]:
 }
 
 
+def _uv_strip_global_opts(args: list[str]) -> list[str]:
+    """Drop uv's *global* options (and the values of value-taking ones) that
+    precede the subcommand, so ``uv --directory repo run rm`` resolves to
+    ``run rm`` and the wrapped command is not hidden behind a global flag's
+    value (``uv --directory repo run rm -rf /`` must still classify as ``rm``).
+
+    Reuses ``_UV_RUN_VALUE_OPTS`` (a superset of uv's value-taking global options)
+    to decide which flags consume a following word; ``--opt=value`` carries its
+    value inline and is consumed as one token.
+    """
+    i = 0
+    while i < len(args) and args[i].startswith("-"):
+        if args[i] == "--":
+            i += 1
+            break
+        if "=" not in args[i] and args[i] in _UV_RUN_VALUE_OPTS and i + 1 < len(args):
+            i += 2
+        else:
+            i += 1
+    return args[i:]
+
+
 def _uv_run_payload(args: list[str]) -> list[str] | None:
     """For ``uv run [opts] <cmd> ...``, return the wrapped command tokens, else ``None``.
 
@@ -891,7 +941,7 @@ def _segment_destructive_reason(tokens: list[str]) -> str | None:
         if payload and (r := _segment_destructive_reason(payload)):
             return f"{base}: {r}"
     if base == "uv":
-        run_payload = _uv_run_payload(args)
+        run_payload = _uv_run_payload(_uv_strip_global_opts(args))
         if run_payload and (r := _segment_destructive_reason(run_payload)):
             return f"uv run: {r}"
     return None

src/pythinker_code/soul/pythinkersoul.py

@@ -1505,6 +1505,16 @@ async def _append_notification(view: NotificationView) -> None:
         # Normalize: merge adjacent user messages for clean API input
         effective_history = normalize_history(self._context.history)
 
+        # Capture tool results as they stream in. If the batch is interrupted
+        # mid-flight, already-completed calls must keep their real output rather
+        # than being overwritten with a synthetic "interrupted" marker; only the
+        # still-pending calls get the marker (see the CancelledError handler).
+        completed_tool_results: dict[str, ToolResult] = {}
+
+        def _on_tool_result(tool_result: ToolResult) -> None:
+            completed_tool_results[tool_result.tool_call_id] = tool_result
+            wire_send(tool_result)
+
         async def _run_step_once() -> StepResult:
             # run an LLM step (may be interrupted)
             from pythinker_code.telemetry import metrics as _m
@@ -1535,7 +1545,7 @@ async def _run_step_once() -> StepResult:
                                 self._agent.toolset,
                                 effective_history,
                                 on_message_part=wire_send,
-                                on_tool_result=wire_send,
+                                on_tool_result=_on_tool_result,
                             )
                     finally:
                         reset_step_permission_profile(profile_token)
@@ -1668,12 +1678,15 @@ async def _pythinker_core_step_with_retry() -> StepResult:
             try:
                 results = await result.tool_results()
             except asyncio.CancelledError:
-                # Interrupted mid-tool: persist the assistant message and a synthetic
-                # interruption marker for every tool_call so the next turn does not see
-                # unanswered tool_calls (which providers reject).  Shield the write from
-                # the same cancellation so it completes, then re-raise.
+                # Interrupted mid-tool: persist the assistant message plus a result
+                # for every tool_call so the next turn does not see unanswered
+                # tool_calls (which providers reject). Keep the real output of calls
+                # that already completed (streamed via on_tool_result); only the
+                # still-pending calls get a synthetic interruption marker. Shield the
+                # write from the same cancellation so it completes, then re-raise.
                 interrupted = [
-                    ToolResult(
+                    completed_tool_results.get(tc.id)
+                    or ToolResult(
                         tool_call_id=tc.id,
                         return_value=ToolRuntimeError(message="Tool call interrupted by user."),
                     )

src/pythinker_code/soul/slash.py

@@ -320,11 +320,9 @@ async def export(soul: PythinkerSoul, args: str):
 @registry.command(name="import")
 async def import_context(soul: PythinkerSoul, args: str):
     """Import context from a file or session ID"""
-    from pythinker_code.utils.export import perform_import
+    from pythinker_code.utils.export import parse_import_args, perform_import
 
-    tokens = args.split()
-    force = "--force" in tokens
-    target = sanitize_cli_path(" ".join(t for t in tokens if t != "--force"))
+    target, force = parse_import_args(args)
     if not target:
         wire_send(TextPart(text="Usage: /import <file_path or session_id>"))
         return

src/pythinker_code/tools/web/fetch.py

@@ -29,7 +29,7 @@ def _ip_is_blocked(address: str) -> bool:
     try:
         ip = ipaddress.ip_address(address)
     except ValueError:
-        return False
+        return True  # fail closed: block addresses we cannot parse/classify
     return (not ip.is_global) or ip.is_multicast  # W2 fail-closed shape
 
 

src/pythinker_code/ui/shell/export_import.py

@@ -9,7 +9,7 @@
 from pythinker_code.ui.shell.console import console
 from pythinker_code.ui.shell.slash import ensure_pythinker_soul, registry, shell_mode_registry
 from pythinker_code.ui.theme import get_tui_tokens as _get_tui_tokens
-from pythinker_code.utils.path import sanitize_cli_path, shorten_home
+from pythinker_code.utils.path import shorten_home
 from pythinker_code.wire.types import TurnBegin, TurnEnd
 
 if TYPE_CHECKING:
@@ -66,15 +66,13 @@ async def export(app: Shell, args: str):
 @shell_mode_registry.command(name="import")
 async def import_context(app: Shell, args: str):
     """Import context from a file or session ID"""
-    from pythinker_code.utils.export import perform_import
+    from pythinker_code.utils.export import parse_import_args, perform_import
 
     soul = ensure_pythinker_soul(app)
     if soul is None:
         return
 
-    tokens = args.split()
-    force = "--force" in tokens
-    target = sanitize_cli_path(" ".join(t for t in tokens if t != "--force"))
+    target, force = parse_import_args(args)
     _t = _get_tui_tokens()
     if not target:
         console.print(f"[{_t.warning}]Usage: /import <file_path or session_id>[/]")

src/pythinker_code/utils/export.py

@@ -39,6 +39,23 @@
 """Common tool-call argument keys whose values make good one-line hints."""
 
 
+def parse_import_args(args: str) -> tuple[str, bool]:
+    """Parse ``/import`` arguments into ``(sanitized_path, force)``.
+
+    Uses shell-style tokenization so quoted/escaped paths with spaces survive
+    instead of being collapsed by ``str.split``. ``--force`` anywhere in the
+    tokens sets the flag and is dropped from the path before sanitization.
+    """
+    try:
+        tokens = shlex.split(args)
+    except ValueError:
+        # Unbalanced quotes: fall back to a plain split rather than raising.
+        tokens = args.split()
+    force = "--force" in tokens
+    path = sanitize_cli_path(" ".join(t for t in tokens if t != "--force"))
+    return path, force
+
+
 def _is_checkpoint_message(msg: Message) -> bool:
     """Check if a message is an internal checkpoint marker."""
     if msg.role != "user" or len(msg.content) != 1: