Expand Python maintenance tasks

python-tools/app/runner.py

@@ -34,24 +34,30 @@ class RunManager:
         input_data: dict[str, str | int | list[str]] | None = None,
         uploads: Mapping[str, PendingUpload] | None = None,
     ) -> tuple[Run, bool]:
+        sensitive_names = {field.name for field in task.inputs if field.sensitive}
+        stored_input = self._redact_input(input_data or {}, sensitive_names)
         run, created = self.store.enqueue(
             target_kind="task",
             target_path=task.id,
             queue_group=task.group_id,
-            input_data=input_data,
+            input_data=stored_input,
         )
         if created:
             try:
                 uploaded_inputs = self._save_uploads(run.id, uploads or {})
+                execution_input = {**(input_data or {}), **uploaded_inputs}
+                stored_input = self._redact_input(execution_input, sensitive_names)
+                run = self.store.set_input(run.id, stored_input)
+                if sensitive_names:
+                    self._write_execution_input(run.id, execution_input)
             except (OSError, ValueError) as error:
                 self._delete_uploads(run.id)
+                self._delete_execution_input(run.id)
                 return self.store.finish(
                     run.id,
                     status="failed",
                     message=f"Could not save uploaded file: {error}",
                 ), True
-            if uploaded_inputs:
-                run = self.store.set_input(run.id, {**(input_data or {}), **uploaded_inputs})
             self._wake_worker(task.group_id)
         return run, created
 
@@ -77,6 +83,7 @@ class RunManager:
         completed = self.store.get(run.id)
         if completed is not None and completed.status == "cancelled":
             self._delete_uploads(run.id)
+            self._delete_execution_input(run.id)
         return completed
 
     def artifacts(self, run_id: int) -> list[str]:
@@ -128,6 +135,13 @@ class RunManager:
                     continue
                 if expired:
                     shutil.rmtree(path, ignore_errors=True)
+        for path in runs_root.glob("*.execution-input.json"):
+            run_id_text = path.name.removesuffix(".execution-input.json")
+            if not run_id_text.isdigit():
+                continue
+            run = self.store.get(int(run_id_text))
+            if run is None or run.status not in {"queued", "running"}:
+                path.unlink(missing_ok=True)
 
     def run_background_task(self, task: BackgroundTask) -> None:
         """Run a declared internal task without creating high-volume run history."""
@@ -239,12 +253,14 @@ class RunManager:
 
         log_path = self.settings.state_root / "runs" / f"{run.id}.log"
         log_path.parent.mkdir(parents=True, exist_ok=True)
-        input_path = self.settings.state_root / "runs" / f"{run.id}.input.json"
+        input_path = self._execution_input_path(run.id)
         artifact_root = self._artifact_root(run.id)
         artifact_root.mkdir(parents=True, exist_ok=True)
         try:
-            input_path.write_text(run.input_json + "\n", encoding="utf-8")
-            input_path.chmod(0o600)
+            if not input_path.exists():
+                input_path = self.settings.state_root / "runs" / f"{run.id}.input.json"
+                input_path.write_text(run.input_json + "\n", encoding="utf-8")
+                input_path.chmod(0o600)
         except OSError as error:
             self._delete_uploads(run.id)
             shutil.rmtree(artifact_root, ignore_errors=True)
@@ -284,6 +300,7 @@ class RunManager:
             with self._lock:
                 self._processes.pop(run.id, None)
             self._delete_uploads(run.id)
+            self._delete_execution_input(run.id)
 
         if self.store.is_cancel_requested(run.id) or (
             parent_run_id is not None and self.store.is_cancel_requested(parent_run_id)
@@ -330,6 +347,32 @@ class RunManager:
     def _artifact_root(self, run_id: int) -> Path:
         return self.settings.state_root / "runs" / f"{run_id}.artifacts"
 
+    def _execution_input_path(self, run_id: int) -> Path:
+        return self.settings.state_root / "runs" / f"{run_id}.execution-input.json"
+
+    def _write_execution_input(
+        self,
+        run_id: int,
+        input_data: Mapping[str, str | int | list[str]],
+    ) -> None:
+        path = self._execution_input_path(run_id)
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(json.dumps(input_data, sort_keys=True, separators=(",", ":")) + "\n", encoding="utf-8")
+        path.chmod(0o600)
+
+    def _delete_execution_input(self, run_id: int) -> None:
+        self._execution_input_path(run_id).unlink(missing_ok=True)
+
+    @staticmethod
+    def _redact_input(
+        input_data: Mapping[str, str | int | list[str]],
+        sensitive_names: set[str],
+    ) -> dict[str, str | int | list[str]]:
+        return {
+            name: "[redacted]" if name in sensitive_names else value
+            for name, value in input_data.items()
+        }
+
     def _save_uploads(
         self,
         run_id: int,