Update Python tools

python-tools/app/discovery.py

@@ -27,6 +27,17 @@ class Task:
     display_name: str
     description: str | None
     inputs: tuple["InputField", ...]
+    wait_for_result: bool = False
+
+
+@dataclass(frozen=True)
+class BackgroundTask:
+    """A declared internal task that is never exposed through the web UI."""
+
+    id: str
+    group_id: str
+    path: Path
+    interval_seconds: int
 
 
 @dataclass(frozen=True)
@@ -56,6 +67,10 @@ class Group:
     display_name: str
     tasks: tuple[Task, ...]
 
+    @property
+    def can_run_as_group(self) -> bool:
+        return all(not task.inputs for task in self.tasks)
+
 
 class TaskCatalog:
     """Discovers only direct Python children of non-hidden group directories."""
@@ -96,6 +111,21 @@ class TaskCatalog:
                     return task
         return None
 
+    def background_tasks(self) -> tuple[BackgroundTask, ...]:
+        if not self.task_root.is_dir():
+            return ()
+
+        tasks = []
+        for group_path in sorted(self.task_root.iterdir(), key=lambda path: path.name):
+            if (
+                group_path.name.startswith((".", "_"))
+                or group_path.is_symlink()
+                or not group_path.is_dir()
+            ):
+                continue
+            tasks.extend(load_background_tasks(group_path))
+        return tuple(tasks)
+
     def _tasks_in_group(self, group_path: Path) -> list[Task]:
         task_paths = [
             path
@@ -107,12 +137,13 @@ class TaskCatalog:
                 or path.suffix != ".py"
             )
         ]
-        inputs_by_filename = load_input_manifest(
+        declarations_by_filename = load_input_manifest(
             group_path,
             {path.name for path in task_paths},
         )
         tasks = []
         for path in task_paths:
+            declaration = declarations_by_filename.get(path.name)
             tasks.append(
                 Task(
                     id=f"{group_path.name}/{path.name}",
@@ -121,12 +152,76 @@ class TaskCatalog:
                     path=path,
                     display_name=display_name(path.stem),
                     description=read_description(path),
-                    inputs=inputs_by_filename.get(path.name, ()),
+                    inputs=declaration.inputs if declaration else (),
+                    wait_for_result=declaration.wait_for_result if declaration else False,
                 )
             )
         return tasks
 
 
+def load_background_tasks(group_path: Path) -> list[BackgroundTask]:
+    """Read optional internal recurring-task declarations for one group."""
+
+    manifest_path = group_path / "background-tasks.json"
+    if not manifest_path.exists():
+        return []
+    if manifest_path.is_symlink():
+        raise ValueError(f"Background task manifest cannot be a symlink: {manifest_path}")
+    try:
+        data = json.loads(manifest_path.read_text(encoding="utf-8"))
+    except (OSError, UnicodeDecodeError, json.JSONDecodeError) as error:
+        raise ValueError(f"Could not read background task manifest {manifest_path}: {error}") from error
+    if not isinstance(data, dict) or set(data) != {"version", "tasks"}:
+        raise ValueError(
+            f"Background task manifest {manifest_path} must contain only version and tasks."
+        )
+    if data["version"] != 1 or not isinstance(data["tasks"], list):
+        raise ValueError(f"Background task manifest {manifest_path} has an unsupported structure.")
+
+    tasks = []
+    seen_ids = set()
+    for declaration in data["tasks"]:
+        if not isinstance(declaration, dict) or set(declaration) != {"id", "script", "interval_seconds"}:
+            raise ValueError(f"Background task manifest {manifest_path} has an invalid task declaration.")
+        task_id = declaration["id"]
+        script = declaration["script"]
+        interval_seconds = declaration["interval_seconds"]
+        if not isinstance(task_id, str) or not FIELD_NAME.fullmatch(task_id) or task_id in seen_ids:
+            raise ValueError(f"Background task manifest {manifest_path} has an invalid or duplicate task id.")
+        if (
+            not isinstance(script, str)
+            or not script.startswith("internal/")
+            or Path(script).suffix != ".py"
+            or Path(script).is_absolute()
+            or ".." in Path(script).parts
+        ):
+            raise ValueError(f"Background task manifest {manifest_path} task {task_id} has an invalid script.")
+        if not isinstance(interval_seconds, int) or isinstance(interval_seconds, bool) or interval_seconds < 1:
+            raise ValueError(f"Background task manifest {manifest_path} task {task_id} has an invalid interval.")
+        path = group_path / script
+        try:
+            path.relative_to(group_path)
+        except ValueError as error:
+            raise ValueError(f"Background task manifest {manifest_path} task {task_id} escapes its group.") from error
+        if path.is_symlink() or not path.is_file():
+            raise ValueError(f"Background task manifest {manifest_path} task {task_id} script is unavailable.")
+        parent = path.parent
+        while parent != group_path:
+            if parent.is_symlink():
+                raise ValueError(f"Background task manifest {manifest_path} task {task_id} script uses a symlink.")
+            parent = parent.parent
+        tasks.append(
+            BackgroundTask(
+                id=f"{group_path.name}/{task_id}",
+                group_id=group_path.name,
+                path=path,
+                interval_seconds=interval_seconds,
+            )
+        )
+        seen_ids.add(task_id)
+    return tasks
+
+
 def read_description(path: Path) -> str | None:
     """Return the first docstring line without importing or executing a task."""
 
@@ -146,7 +241,7 @@ def read_description(path: Path) -> str | None:
 def load_input_manifest(
     group_path: Path,
     task_filenames: set[str],
-) -> dict[str, tuple[InputField, ...]]:
+) -> dict[str, "TaskDeclaration"]:
     """Read the optional data-only input declaration for one task group."""
 
     manifest_path = group_path / "task-inputs.json"
@@ -171,18 +266,24 @@ def load_input_manifest(
         raise ValueError(f"Input manifest {manifest_path} names unknown tasks: {names}.")
 
     return {
-        filename: parse_task_inputs(manifest_path, filename, declaration)
+        filename: parse_task_declaration(manifest_path, filename, declaration)
         for filename, declaration in data["tasks"].items()
     }
 
 
-def parse_task_inputs(
+@dataclass(frozen=True)
+class TaskDeclaration:
+    inputs: tuple[InputField, ...]
+    wait_for_result: bool
+
+
+def parse_task_declaration(
     manifest_path: Path,
     filename: str,
     declaration: Any,
-) -> tuple[InputField, ...]:
-    if not isinstance(declaration, dict) or set(declaration) != {"inputs"}:
-        raise ValueError(f"{manifest_path} task {filename} must contain only inputs.")
+) -> TaskDeclaration:
+    if not isinstance(declaration, dict) or not {"inputs"} <= set(declaration) or set(declaration) - {"inputs", "wait_for_result"}:
+        raise ValueError(f"{manifest_path} task {filename} has an invalid declaration.")
     raw_inputs = declaration["inputs"]
     if not isinstance(raw_inputs, list):
         raise ValueError(f"{manifest_path} task {filename} inputs must be a list.")
@@ -190,7 +291,10 @@ def parse_task_inputs(
     names = [field.name for field in fields]
     if len(names) != len(set(names)):
         raise ValueError(f"{manifest_path} task {filename} has duplicate input names.")
-    return fields
+    wait_for_result = declaration.get("wait_for_result", False)
+    if not isinstance(wait_for_result, bool):
+        raise ValueError(f"{manifest_path} task {filename} wait_for_result must be true or false.")
+    return TaskDeclaration(inputs=fields, wait_for_result=wait_for_result)
 
 
 def parse_input_field(manifest_path: Path, filename: str, raw: Any) -> InputField: