core dabbit

Author: NSPC911

docs/src/content/docs/dev/features/ipc.mdx

@@ -44,6 +44,8 @@ Every command prints one JSON object. Successful commands return `"ok": true` an
 IPC only listens on `127.0.0.1`, and each rovr instance generates a private token. The target instance still checks `settings.ipc.permissions` for every action.<br/>
 Permissions can be `allow`, `deny`, or `prompt`; unconfigured actions are denied. Subcommands use keys such as `clipboard.paste` and `tab.close`.
 
+`prompt` is unavailable for `notify`, `ask`, and `input` commands, which always require user interaction.
+
 ## Commands
 
 ### `cd`
@@ -85,6 +87,8 @@ rovr --ipc clipboard list
 }
 ```
 
+if the user intervenes due to errors, returns with `"ok": false`
+
 `list` returns every clipboard entry with its type and whether it is selected:
 
 ```json

src/rovr/__main__.py

@@ -300,7 +300,7 @@ def cli(argv: list[str] | None = None) -> None:
                 args.ipc
                 if args.ipc is not None
                 else args.ipc_to[1:]
-                if args.ipc_to[0].isdigit()
+                if args.ipc_to[0].isdecimal()
                 else args.ipc_to
             )
             IPC_PARSER.parse_args(ipc_args)

src/rovr/assets/config.toml

@@ -153,6 +153,9 @@ cd = "allow"
 quit = "prompt"
 suspend = "allow"
 
+ask = "allow"
+input = "allow"
+notify = "allow"
 
 [metadata]
 fields = [

src/rovr/footer/process_container.py

@@ -963,13 +963,17 @@ def extract_archive(self, archive_path: str, destination_path: str) -> None:
     @work(thread=True)
     def paste_items(
         self, copied: list[str], has_cut: list[str], dest: str = ""
-    ) -> None:
+    ) -> None | Literal[True]:
         """
         Paste copied or cut files to the current directory
         Args:
             copied (list[str]): A list of items to be copied to the location
             has_cut (list[str]): A list of items to be cut to the location
             dest (str): The directory to copy to.
+
+        Returns:
+            True: If the operation was successful.
+            None: If the operation was cancelled or failed.
         """
         if dest == "":
             dest = getcwd()
@@ -1447,6 +1451,7 @@ def paste_items(
             with suppress(OSError):
                 os.rmdir(folder)
         bar.ok()
+        return True
 
     @work(thread=True)
     def remote_download(self, uris: list[str], paths: list[str]) -> None:

src/rovr/functions/ipc_instances.py

@@ -46,7 +46,7 @@ def publish_instance(port: int, token: str) -> Path:
         os.replace(temporary, destination)
         # safety first
         destination.chmod(0o600)
-    except BaseException:
+    except Exception:
         temporary.unlink(missing_ok=True)
         raise
     return destination
@@ -107,9 +107,11 @@ async def _probe_instance(path: Path, descriptor: Instance) -> InstanceInfo | No
         )
     except TimeoutError:
         return info
-    except OSError:
+    except ConnectionRefusedError:
         path.unlink(missing_ok=True)
         return
+    except OSError:
+        return info
 
     try:
         writer.write(

src/rovr/functions/ipc_receiver.py

@@ -8,6 +8,7 @@
 from typing import Any, Callable, Literal, TypedDict, cast
 
 from textual import work
+from textual.worker import Worker
 
 from rovr.app import Application
 from rovr.functions.cwd import getcwd
@@ -149,11 +150,14 @@ async def conn(
                         # does because we are assuming that ipc paste is
                         # a prompt (default) else the person kniws what
                         # they want from rovr
-                        worker = self.app.query_one("ProcessContainer").paste_items(
-                            to_copy, to_cut, getcwd()
-                        )
+                        worker: Worker[None | Literal[True]] = self.app.query_one(
+                            "ProcessContainer"
+                        ).paste_items(to_copy, to_cut, getcwd())
                         await worker.wait()
-                        out = {"copy": len(to_copy), "cut": len(to_cut)}
+                        if ok := bool(worker.result):
+                            out = {"copy": len(to_copy), "cut": len(to_cut)}
+                        else:
+                            err = "user intervened"
                 case "copy" | "cut":
                     if not await check_permission(self, action, args):
                         ok = False
@@ -168,8 +172,12 @@ async def conn(
                             "keep",
                         )
                         if selection not in ("keep", "add", "replace"):
-                            ok = False
-                            err = "selection must be keep, add, or replace"
+                            return assemble_and_write(
+                                writer,
+                                False,
+                                None,
+                                "invalid selection argument, must be one of keep, add, replace",
+                            )
                         avail = [
                             p(path)
                             for path in args[1:]
@@ -181,7 +189,8 @@ async def conn(
                             if not (path in avail or path.startswith("--"))
                         ]
                         func: Callable[
-                            [list[str], Literal["reselect", "select", "no"]], None
+                            [list[str], Literal["reselect", "select", "no"]],
+                            Worker[None],
                         ] = (
                             self.Clipboard.copy_to_clipboard
                             if args[0] == "copy"
@@ -192,7 +201,8 @@ async def conn(
                             select = "select"
                         elif selection == "replace":
                             select = "reselect"
-                        func(avail, select)
+                        worker = func(avail, select)
+                        await worker.wait()
                 case _:
                     ok = False
                     err = "clipboard action is not valid"
@@ -344,6 +354,16 @@ async def conn(
                 ok = False
                 err = "invalid notification arguments"
             else:
+                if notification.markup:
+                    from textual.content import Content
+
+                    try:
+                        Content.from_markup(notification.message)
+                    except Exception as exc:
+                        return assemble_and_write(
+                            writer, False, None, f"invalid markup: {exc}"
+                        )
+
                 self.notify(
                     notification.message,
                     title=notification.title or "",
@@ -374,23 +394,26 @@ async def conn(
                 else:
                     self.action_suspend_process()
         case "ask":
-            if not await check_permission(self, action, args):
-                ok = False
-                err = "denied"
-            elif len(args) != 1:
+            if len(args) != 1:
                 ok = False
                 err = "too many arguments" if args else "question not provided"
+            elif not await check_permission(self, action, args):
+                ok = False
+                err = "denied"
             else:
                 from rovr.screens import YesOrNo
 
                 out: bool = await self.push_screen_wait(YesOrNo(args[0]))
         case "input":
-            if not await check_permission(self, action, args):
+            if len(args) > 2 or len(args) == 0:
                 ok = False
-                err = "denied"
-            elif len(args) != 1:
+                err = "too many arguments" if len(args) > 2 else "prompt not provided"
+            elif len(args) == 1 and args[0] == "--is-path":
                 ok = False
-                err = "too many arguments" if args else "prompt not provided"
+                err = "prompt not provided"
+            elif not await check_permission(self, action, args):
+                ok = False
+                err = "denied"
             else:
                 from rovr.screens import ModalInput