diff --git a/python-tools/app/discovery.py b/python-tools/app/discovery.py
index d8e5acf..8f2d3ce 100644
--- a/python-tools/app/discovery.py
+++ b/python-tools/app/discovery.py
@@ -61,6 +61,7 @@ class InputField:
     options: tuple[InputOption, ...]
     accept: tuple[str, ...]
     maximum_bytes: int | None
+    sensitive: bool = False
 
 
 @dataclass(frozen=True)
@@ -325,7 +326,7 @@ def parse_input_field(manifest_path: Path, filename: str, raw: Any) -> InputFiel
         raise ValueError(f"{manifest_path} task {filename} has a non-object input field.")
     allowed = {
         "name", "label", "type", "required", "default", "minimum", "maximum",
-        "step", "pattern", "options", "accept", "maximum_bytes",
+        "step", "pattern", "options", "accept", "maximum_bytes", "sensitive",
     }
     unknown = set(raw) - allowed
     if unknown:
@@ -344,6 +345,9 @@ def parse_input_field(manifest_path: Path, filename: str, raw: Any) -> InputFiel
     required = raw.get("required", False)
     if not isinstance(required, bool):
         raise ValueError(f"{manifest_path} task {filename} input {name} required must be true or false.")
+    sensitive = raw.get("sensitive", False)
+    if not isinstance(sensitive, bool):
+        raise ValueError(f"{manifest_path} task {filename} input {name} sensitive must be true or false.")
 
     numeric_keys = ("minimum", "maximum", "step")
     numeric_values = {key: raw.get(key) for key in numeric_keys}
@@ -407,6 +411,7 @@ def parse_input_field(manifest_path: Path, filename: str, raw: Any) -> InputFiel
         options=options,
         accept=accept,
         maximum_bytes=maximum_bytes,
+        sensitive=sensitive,
     )
 
 
diff --git a/python-tools/app/runner.py b/python-tools/app/runner.py
index 8c076b2..1f81aed 100644
--- a/python-tools/app/runner.py
+++ b/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,
diff --git a/python-tools/app/web.py b/python-tools/app/web.py
index 0a4257d..08ebdb4 100644
--- a/python-tools/app/web.py
+++ b/python-tools/app/web.py
@@ -102,7 +102,12 @@ def create_app(settings: Settings | None = None) -> Flask:
             if request.accept_mimetypes.best == "application/json":
                 return {"errors": errors}, 400
             return render_index(
-                form_values={task.id: submitted_values},
+                form_values={
+                    task.id: {
+                        field.name: [] if field.sensitive else submitted_values[field.name]
+                        for field in task.inputs
+                    }
+                },
                 input_errors={task.id: errors},
                 status=400,
             )
diff --git a/python-tools/templates/index.html b/python-tools/templates/index.html
index 765504e..d4cb05c 100644
--- a/python-tools/templates/index.html
+++ b/python-tools/templates/index.html
@@ -68,12 +68,13 @@
                       </select>
                     </label>
                   {% else %}
-                    {% set html_type = "number" if field.type == "integer" else "datetime-local" if field.type == "datetime" else field.type %}
+                    {% set html_type = "password" if field.sensitive else "number" if field.type == "integer" else "datetime-local" if field.type == "datetime" else field.type %}
                     <label>{{ field.label }}{% if field.required %} (required){% endif %}
                       <input
                         type="{{ html_type }}"
                         name="{{ field.name }}"
-                        value="{{ values.get(field.name, field.default if field.default is not none else "") }}"
+                        value="{{ "" if field.sensitive else values.get(field.name, field.default if field.default is not none else "") }}"
+                        {% if field.sensitive %}autocomplete="off"{% endif %}
                         {% if field.required %}required{% endif %}
                         {% if field.minimum is not none %}min="{{ field.minimum }}"{% endif %}
                         {% if field.maximum is not none %}max="{{ field.maximum }}"{% endif %}