diff --git a/python-tools/app/discovery.py b/python-tools/app/discovery.py
index b103e5d..2188301 100644
--- a/python-tools/app/discovery.py
+++ b/python-tools/app/discovery.py
@@ -3,8 +3,15 @@
 from __future__ import annotations
 
 import ast
+import json
+import re
 from dataclasses import dataclass
 from pathlib import Path
+from typing import Any
+
+
+FIELD_NAME = re.compile(r"[a-z][a-z0-9_]*\Z")
+FIELD_TYPES = frozenset({"text", "integer", "date", "datetime", "choice", "multi_choice"})
 
 
 def display_name(value: str) -> str:
@@ -19,6 +26,27 @@ class Task:
     path: Path
     display_name: str
     description: str | None
+    inputs: tuple["InputField", ...]
+
+
+@dataclass(frozen=True)
+class InputOption:
+    value: str
+    label: str
+
+
+@dataclass(frozen=True)
+class InputField:
+    name: str
+    label: str
+    type: str
+    required: bool
+    default: str | int | tuple[str, ...] | None
+    minimum: int | None
+    maximum: int | None
+    step: int | None
+    pattern: str | None
+    options: tuple[InputOption, ...]
 
 
 @dataclass(frozen=True)
@@ -69,15 +97,22 @@ class TaskCatalog:
         return None
 
     def _tasks_in_group(self, group_path: Path) -> list[Task]:
-        tasks = []
-        for path in sorted(group_path.iterdir(), key=lambda candidate: candidate.name):
-            if (
+        task_paths = [
+            path
+            for path in sorted(group_path.iterdir(), key=lambda candidate: candidate.name)
+            if not (
                 path.name.startswith((".", "_"))
                 or path.is_symlink()
                 or not path.is_file()
                 or path.suffix != ".py"
-            ):
-                continue
+            )
+        ]
+        inputs_by_filename = load_input_manifest(
+            group_path,
+            {path.name for path in task_paths},
+        )
+        tasks = []
+        for path in task_paths:
             tasks.append(
                 Task(
                     id=f"{group_path.name}/{path.name}",
@@ -86,6 +121,7 @@ class TaskCatalog:
                     path=path,
                     display_name=display_name(path.stem),
                     description=read_description(path),
+                    inputs=inputs_by_filename.get(path.name, ()),
                 )
             )
         return tasks
@@ -105,3 +141,214 @@ def read_description(path: Path) -> str | None:
         (line.strip() for line in docstring.splitlines() if line.strip()),
         None,
     )
+
+
+def load_input_manifest(
+    group_path: Path,
+    task_filenames: set[str],
+) -> dict[str, tuple[InputField, ...]]:
+    """Read the optional data-only input declaration for one task group."""
+
+    manifest_path = group_path / "task-inputs.json"
+    if not manifest_path.exists():
+        return {}
+    if manifest_path.is_symlink():
+        raise ValueError(f"Input 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 input manifest {manifest_path}: {error}") from error
+    if not isinstance(data, dict) or set(data) != {"version", "tasks"}:
+        raise ValueError(
+            f"Input manifest {manifest_path} must contain only version and tasks."
+        )
+    if data["version"] != 1 or not isinstance(data["tasks"], dict):
+        raise ValueError(f"Input manifest {manifest_path} has an unsupported structure.")
+
+    unknown_tasks = set(data["tasks"]) - task_filenames
+    if unknown_tasks:
+        names = ", ".join(sorted(unknown_tasks))
+        raise ValueError(f"Input manifest {manifest_path} names unknown tasks: {names}.")
+
+    return {
+        filename: parse_task_inputs(manifest_path, filename, declaration)
+        for filename, declaration in data["tasks"].items()
+    }
+
+
+def parse_task_inputs(
+    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.")
+    raw_inputs = declaration["inputs"]
+    if not isinstance(raw_inputs, list):
+        raise ValueError(f"{manifest_path} task {filename} inputs must be a list.")
+    fields = tuple(parse_input_field(manifest_path, filename, raw) for raw in raw_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
+
+
+def parse_input_field(manifest_path: Path, filename: str, raw: Any) -> InputField:
+    if not isinstance(raw, dict):
+        raise ValueError(f"{manifest_path} task {filename} has a non-object input field.")
+    allowed = {
+        "name", "label", "type", "required", "default", "minimum", "maximum",
+        "step", "pattern", "options",
+    }
+    unknown = set(raw) - allowed
+    if unknown:
+        raise ValueError(
+            f"{manifest_path} task {filename} input has unknown keys: {', '.join(sorted(unknown))}."
+        )
+    name = raw.get("name")
+    label = raw.get("label")
+    field_type = raw.get("type")
+    if not isinstance(name, str) or not FIELD_NAME.fullmatch(name):
+        raise ValueError(f"{manifest_path} task {filename} has an invalid input name.")
+    if not isinstance(label, str) or not label.strip():
+        raise ValueError(f"{manifest_path} task {filename} input {name} needs a label.")
+    if field_type not in FIELD_TYPES:
+        raise ValueError(f"{manifest_path} task {filename} input {name} has an invalid type.")
+    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.")
+
+    numeric_keys = ("minimum", "maximum", "step")
+    numeric_values = {key: raw.get(key) for key in numeric_keys}
+    if field_type == "integer":
+        for key, value in numeric_values.items():
+            if value is not None and (not isinstance(value, int) or isinstance(value, bool)):
+                raise ValueError(f"{manifest_path} task {filename} input {name} {key} must be an integer.")
+        if numeric_values["minimum"] is not None and numeric_values["maximum"] is not None and numeric_values["minimum"] > numeric_values["maximum"]:
+            raise ValueError(f"{manifest_path} task {filename} input {name} minimum exceeds maximum.")
+        if numeric_values["step"] is not None and numeric_values["step"] <= 0:
+            raise ValueError(f"{manifest_path} task {filename} input {name} step must be positive.")
+    elif any(value is not None for value in numeric_values.values()):
+        raise ValueError(f"{manifest_path} task {filename} input {name} only integer fields accept numeric limits.")
+
+    pattern = raw.get("pattern")
+    if pattern is not None:
+        if field_type != "text" or not isinstance(pattern, str):
+            raise ValueError(f"{manifest_path} task {filename} input {name} has an invalid pattern.")
+        try:
+            re.compile(pattern)
+        except re.error as error:
+            raise ValueError(f"{manifest_path} task {filename} input {name} has an invalid pattern: {error}.") from error
+
+    options = parse_options(manifest_path, filename, name, field_type, raw.get("options"))
+    default = parse_default(manifest_path, filename, name, field_type, options, raw.get("default"))
+    validate_default(
+        manifest_path,
+        filename,
+        name,
+        field_type,
+        default,
+        numeric_values["minimum"],
+        numeric_values["maximum"],
+        numeric_values["step"],
+        pattern,
+    )
+    return InputField(
+        name=name,
+        label=label,
+        type=field_type,
+        required=required,
+        default=default,
+        minimum=numeric_values["minimum"],
+        maximum=numeric_values["maximum"],
+        step=numeric_values["step"],
+        pattern=pattern,
+        options=options,
+    )
+
+
+def parse_options(
+    manifest_path: Path,
+    filename: str,
+    name: str,
+    field_type: str,
+    raw_options: Any,
+) -> tuple[InputOption, ...]:
+    if field_type not in {"choice", "multi_choice"}:
+        if raw_options is not None:
+            raise ValueError(f"{manifest_path} task {filename} input {name} only choice fields accept options.")
+        return ()
+    if not isinstance(raw_options, list) or not raw_options:
+        raise ValueError(f"{manifest_path} task {filename} input {name} needs non-empty options.")
+    options = []
+    for raw_option in raw_options:
+        if isinstance(raw_option, str) and raw_option:
+            options.append(InputOption(value=raw_option, label=raw_option))
+        elif (
+            isinstance(raw_option, dict)
+            and set(raw_option) == {"value", "label"}
+            and isinstance(raw_option["value"], str)
+            and raw_option["value"]
+            and isinstance(raw_option["label"], str)
+            and raw_option["label"].strip()
+        ):
+            options.append(InputOption(value=raw_option["value"], label=raw_option["label"]))
+        else:
+            raise ValueError(f"{manifest_path} task {filename} input {name} has an invalid option.")
+    if len({option.value for option in options}) != len(options):
+        raise ValueError(f"{manifest_path} task {filename} input {name} has duplicate option values.")
+    return tuple(options)
+
+
+def parse_default(
+    manifest_path: Path,
+    filename: str,
+    name: str,
+    field_type: str,
+    options: tuple[InputOption, ...],
+    default: Any,
+) -> str | int | tuple[str, ...] | None:
+    if default is None:
+        return None
+    if field_type == "integer":
+        if not isinstance(default, int) or isinstance(default, bool):
+            raise ValueError(f"{manifest_path} task {filename} input {name} default must be an integer.")
+        return default
+    if field_type == "multi_choice":
+        if not isinstance(default, list) or not all(isinstance(value, str) for value in default):
+            raise ValueError(f"{manifest_path} task {filename} input {name} default must be a list of options.")
+        values = tuple(default)
+        if len(values) != len(set(values)) or not set(values) <= {option.value for option in options}:
+            raise ValueError(f"{manifest_path} task {filename} input {name} default contains an unknown option.")
+        return values
+    if not isinstance(default, str):
+        raise ValueError(f"{manifest_path} task {filename} input {name} default must be text.")
+    if field_type == "choice" and default not in {option.value for option in options}:
+        raise ValueError(f"{manifest_path} task {filename} input {name} default is not an option.")
+    return default
+
+
+def validate_default(
+    manifest_path: Path,
+    filename: str,
+    name: str,
+    field_type: str,
+    default: str | int | tuple[str, ...] | None,
+    minimum: int | None,
+    maximum: int | None,
+    step: int | None,
+    pattern: str | None,
+) -> None:
+    if default is None or field_type not in {"integer", "text"}:
+        return
+    if field_type == "text":
+        if pattern and not re.fullmatch(pattern, default):
+            raise ValueError(f"{manifest_path} task {filename} input {name} default does not match its pattern.")
+        return
+    assert isinstance(default, int)
+    if minimum is not None and default < minimum:
+        raise ValueError(f"{manifest_path} task {filename} input {name} default is below minimum.")
+    if maximum is not None and default > maximum:
+        raise ValueError(f"{manifest_path} task {filename} input {name} default is above maximum.")
+    if step is not None and (default - (minimum or 0)) % step:
+        raise ValueError(f"{manifest_path} task {filename} input {name} default does not match step.")
diff --git a/python-tools/app/inputs.py b/python-tools/app/inputs.py
new file mode 100644
index 0000000..c34466c
--- /dev/null
+++ b/python-tools/app/inputs.py
@@ -0,0 +1,87 @@
+"""Validation for task-input manifest values."""
+
+from __future__ import annotations
+
+import re
+from collections.abc import Mapping, Sequence
+from datetime import date, datetime
+
+from .discovery import InputField
+
+
+def validate_inputs(
+    fields: Sequence[InputField],
+    values: Mapping[str, list[str]],
+) -> tuple[dict[str, str | int | list[str]], dict[str, str]]:
+    result: dict[str, str | int | list[str]] = {}
+    errors: dict[str, str] = {}
+    for field in fields:
+        raw_values = values.get(field.name, [])
+        value, error = validate_field(field, raw_values)
+        if error:
+            errors[field.name] = error
+        elif value is not None:
+            result[field.name] = value
+    return result, errors
+
+
+def validate_field(
+    field: InputField,
+    raw_values: list[str],
+) -> tuple[str | int | list[str] | None, str | None]:
+    if field.type == "multi_choice":
+        return validate_multi_choice(field, raw_values)
+    if len(raw_values) > 1:
+        return None, "Only one value is allowed."
+    raw = raw_values[0] if raw_values else ""
+    if not raw:
+        if field.default is not None:
+            return field.default, None
+        if field.required:
+            return None, "This field is required."
+        return None, None
+    if field.type == "integer":
+        try:
+            value = int(raw)
+        except ValueError:
+            return None, "Enter a whole number."
+        if field.minimum is not None and value < field.minimum:
+            return None, f"Enter a value of at least {field.minimum}."
+        if field.maximum is not None and value > field.maximum:
+            return None, f"Enter a value no greater than {field.maximum}."
+        if field.step is not None and (value - (field.minimum or 0)) % field.step:
+            return None, f"Enter a value in increments of {field.step}."
+        return value, None
+    if field.type == "date":
+        try:
+            date.fromisoformat(raw)
+        except ValueError:
+            return None, "Enter a valid date."
+    elif field.type == "datetime":
+        try:
+            datetime.fromisoformat(raw)
+        except ValueError:
+            return None, "Enter a valid date and time."
+    elif field.type == "choice" and raw not in {option.value for option in field.options}:
+        return None, "Choose one of the available options."
+    elif field.type == "text" and field.pattern and not re.fullmatch(field.pattern, raw):
+        return None, "Enter a value in the required format."
+    return raw, None
+
+
+def validate_multi_choice(
+    field: InputField,
+    raw_values: list[str],
+) -> tuple[list[str] | None, str | None]:
+    if not raw_values:
+        if field.default is not None:
+            return list(field.default), None
+        if field.required:
+            return None, "Choose at least one option."
+        return None, None
+    allowed = {option.value for option in field.options}
+    if any(value not in allowed for value in raw_values):
+        return None, "Choose only from the available options."
+    if len(raw_values) != len(set(raw_values)):
+        return None, "Each option can be selected only once."
+    return raw_values, None
diff --git a/python-tools/app/runner.py b/python-tools/app/runner.py
index 5f14a36..c7195de 100644
--- a/python-tools/app/runner.py
+++ b/python-tools/app/runner.py
@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import os
+import json
 import signal
 import subprocess
 import threading
@@ -23,11 +24,16 @@ class RunManager:
         self._processes: dict[int, subprocess.Popen[bytes]] = {}
         self._lock = threading.Lock()
 
-    def run_task(self, task: Task) -> tuple[Run, bool]:
+    def run_task(
+        self,
+        task: Task,
+        input_data: dict[str, str | int | list[str]] | None = None,
+    ) -> tuple[Run, bool]:
         run, created = self.store.enqueue(
             target_kind="task",
             target_path=task.id,
             queue_group=task.group_id,
+            input_data=input_data,
         )
         if created:
             self._wake_worker(task.group_id)
@@ -143,11 +149,22 @@ class RunManager:
         task_state.mkdir(parents=True, exist_ok=True)
         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"
+        try:
+            input_path.write_text(run.input_json + "\n", encoding="utf-8")
+            input_path.chmod(0o600)
+        except OSError as error:
+            return self.store.finish(
+                run.id,
+                status="failed",
+                message=f"Could not prepare task input: {error}",
+            )
         environment = {
             "PATH": "/usr/local/bin:/usr/bin:/bin",
             "PYTHONUNBUFFERED": "1",
             "SERVER_MAINTENANCE_CREDENTIALS": str(self.settings.credentials_root),
             "SERVER_MAINTENANCE_STATE": str(task_state),
+            "SERVER_MAINTENANCE_INPUT": str(input_path),
         }
         if self.settings.timezone_name:
             environment["TZ"] = self.settings.timezone_name
diff --git a/python-tools/app/store.py b/python-tools/app/store.py
index 3fed05b..5730aba 100644
--- a/python-tools/app/store.py
+++ b/python-tools/app/store.py
@@ -3,6 +3,8 @@
 from __future__ import annotations
 
 import sqlite3
+import json
+from collections.abc import Mapping
 from dataclasses import dataclass
 from datetime import UTC, datetime
 from pathlib import Path
@@ -27,6 +29,7 @@ class Run:
     parent_run_id: int | None
     log_path: str | None
     message: str | None
+    input_json: str
 
 
 def now() -> str:
@@ -56,10 +59,19 @@ class RunStore:
                     cancel_requested_at TEXT,
                     parent_run_id INTEGER REFERENCES runs(id),
                     log_path TEXT,
-                    message TEXT
+                    message TEXT,
+                    input_json TEXT NOT NULL DEFAULT '{}'
                 )
                 """
             )
+            columns = {
+                row["name"]
+                for row in connection.execute("PRAGMA table_info(runs)").fetchall()
+            }
+            if "input_json" not in columns:
+                connection.execute(
+                    "ALTER TABLE runs ADD COLUMN input_json TEXT NOT NULL DEFAULT '{}'"
+                )
             connection.execute(
                 """
                 CREATE INDEX IF NOT EXISTS runs_queue_index
@@ -93,7 +105,9 @@ class RunStore:
         target_path: str,
         queue_group: str,
         trigger: str = "manual",
+        input_data: Mapping[str, str | int | list[str]] | None = None,
     ) -> tuple[Run, bool]:
+        input_json = json.dumps(input_data or {}, sort_keys=True, separators=(",", ":"))
         with self._connect() as connection:
             connection.execute("BEGIN IMMEDIATE")
             existing = connection.execute(
@@ -112,10 +126,10 @@ class RunStore:
             cursor = connection.execute(
                 """
                 INSERT INTO runs (
-                    target_kind, target_path, queue_group, trigger, status, created_at
-                ) VALUES (?, ?, ?, ?, 'queued', ?)
+                    target_kind, target_path, queue_group, trigger, status, created_at, input_json
+                ) VALUES (?, ?, ?, ?, 'queued', ?, ?)
                 """,
-                (target_kind, target_path, queue_group, trigger, now()),
+                (target_kind, target_path, queue_group, trigger, now(), input_json),
             )
             return self.get(cursor.lastrowid, connection=connection), True
 
@@ -146,8 +160,8 @@ class RunStore:
                 """
                 INSERT INTO runs (
                     target_kind, target_path, queue_group, trigger, status, created_at,
-                    started_at, parent_run_id
-                ) VALUES ('task', ?, ?, ?, 'running', ?, ?, ?)
+                    started_at, parent_run_id, input_json
+                ) VALUES ('task', ?, ?, ?, 'running', ?, ?, ?, '{}')
                 """,
                 (
                     task_path,
diff --git a/python-tools/app/web.py b/python-tools/app/web.py
index 396ad54..e61b859 100644
--- a/python-tools/app/web.py
+++ b/python-tools/app/web.py
@@ -10,6 +10,7 @@ from flask import Flask, abort, redirect, render_template, request, url_for
 
 from .config import Settings
 from .discovery import TaskCatalog
+from .inputs import validate_inputs
 from .runner import RunManager
 from .store import RunStore
 
@@ -53,8 +54,12 @@ def create_app(settings: Settings | None = None) -> Flask:
             return "—"
         return datetime.fromisoformat(value).astimezone().strftime("%Y-%m-%d %H:%M:%S %Z")
 
-    @app.get("/")
-    def index():
+    def render_index(
+        *,
+        form_values: dict[str, dict[str, list[str]]] | None = None,
+        input_errors: dict[str, dict[str, str]] | None = None,
+        status: int = 200,
+    ):
         groups = catalog.groups()
         latest = {
             ("group", group.id): store.last_run("group", group.id) for group in groups
@@ -62,14 +67,37 @@ def create_app(settings: Settings | None = None) -> Flask:
         for group in groups:
             for task in group.tasks:
                 latest[("task", task.id)] = store.last_run("task", task.id)
-        return render_template("index.html", groups=groups, latest=latest)
+        return (
+            render_template(
+                "index.html",
+                groups=groups,
+                latest=latest,
+                form_values=form_values or {},
+                input_errors=input_errors or {},
+            ),
+            status,
+        )
+
+    @app.get("/")
+    def index():
+        return render_index()
 
     @app.post("/tasks/<path:task_id>/run")
     def run_task(task_id: str):
         task = catalog.task(task_id)
         if task is None:
             abort(404)
-        run, _ = manager.run_task(task)
+        submitted_values = {
+            field.name: request.form.getlist(field.name) for field in task.inputs
+        }
+        input_data, errors = validate_inputs(task.inputs, submitted_values)
+        if errors:
+            return render_index(
+                form_values={task.id: submitted_values},
+                input_errors={task.id: errors},
+                status=400,
+            )
+        run, _ = manager.run_task(task, input_data)
         return redirect(url_for("run_detail", run_id=run.id))
 
     @app.post("/groups/<path:group_id>/run")
diff --git a/python-tools/static/style.css b/python-tools/static/style.css
index 63f9af7..4bfc3be 100644
--- a/python-tools/static/style.css
+++ b/python-tools/static/style.css
@@ -55,6 +55,37 @@ a {
   margin: .2rem 0;
 }
 
+.tasks form {
+  min-width: 16rem;
+}
+
+.task-input {
+  margin: .5rem 0;
+}
+
+.task-input label,
+.task-input fieldset label {
+  display: block;
+}
+
+.task-input input,
+.task-input select {
+  box-sizing: border-box;
+  font: inherit;
+  max-width: 100%;
+}
+
+.task-input fieldset {
+  border: 0;
+  margin: 0;
+  padding: 0;
+}
+
+.input-error {
+  color: #b00020;
+  margin: .2rem 0;
+}
+
 .run-summary {
   font-size: .9rem;
   opacity: .75;
diff --git a/python-tools/templates/index.html b/python-tools/templates/index.html
index 5297593..9504605 100644
--- a/python-tools/templates/index.html
+++ b/python-tools/templates/index.html
@@ -36,6 +36,49 @@
               {% endif %}
             </div>
             <form method="post" action="{{ url_for('run_task', task_id=task.id) }}">
+              {% set values = form_values.get(task.id, {}) %}
+              {% set errors = input_errors.get(task.id, {}) %}
+              {% for field in task.inputs %}
+                <div class="task-input">
+                  {% if field.type == "multi_choice" %}
+                    <fieldset>
+                      <legend>{{ field.label }}{% if field.required %} (required){% endif %}</legend>
+                      {% set selected = values.get(field.name, field.default or ()) %}
+                      {% for option in field.options %}
+                        <label>
+                          <input type="checkbox" name="{{ field.name }}" value="{{ option.value }}" {% if option.value in selected %}checked{% endif %}>
+                          {{ option.label }}
+                        </label>
+                      {% endfor %}
+                    </fieldset>
+                  {% elif field.type == "choice" %}
+                    <label>{{ field.label }}{% if field.required %} (required){% endif %}
+                      <select name="{{ field.name }}" {% if field.required %}required{% endif %}>
+                        {% if not field.required %}<option value="">No selection</option>{% endif %}
+                        {% set selected = values.get(field.name, field.default or "") %}
+                        {% for option in field.options %}
+                          <option value="{{ option.value }}" {% if option.value == selected %}selected{% endif %}>{{ option.label }}</option>
+                        {% endfor %}
+                      </select>
+                    </label>
+                  {% else %}
+                    {% set html_type = "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 "") }}"
+                        {% 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 %}
+                        {% if field.step is not none %}step="{{ field.step }}"{% endif %}
+                        {% if field.pattern %}pattern="{{ field.pattern }}"{% endif %}
+                      >
+                    </label>
+                  {% endif %}
+                  {% if errors.get(field.name) %}<p class="input-error">{{ errors[field.name] }}</p>{% endif %}
+                </div>
+              {% endfor %}
               <button type="submit">Run</button>
             </form>
           </li>
diff --git a/python-tools/tests/test_store.py b/python-tools/tests/test_store.py
index c1e94a4..8956227 100644
--- a/python-tools/tests/test_store.py
+++ b/python-tools/tests/test_store.py
@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import sqlite3
+
 from app.store import RunStore
 
 
@@ -19,3 +21,39 @@ def test_restart_marks_unfinished_runs_interrupted(settings) -> None:
     assert interrupted is not None
     assert interrupted.status == "interrupted"
     assert interrupted.finished_at is not None
+
+
+def test_initialize_migrates_existing_database_with_input_column(settings) -> None:
+    settings.state_root.mkdir()
+    with sqlite3.connect(settings.database_path) as connection:
+        connection.execute(
+            """
+            CREATE TABLE runs (
+                id INTEGER PRIMARY KEY,
+                target_kind TEXT NOT NULL,
+                target_path TEXT NOT NULL,
+                queue_group TEXT NOT NULL,
+                trigger TEXT NOT NULL,
+                status TEXT NOT NULL,
+                created_at TEXT NOT NULL,
+                started_at TEXT,
+                finished_at TEXT,
+                exit_code INTEGER,
+                cancel_requested_at TEXT,
+                parent_run_id INTEGER,
+                log_path TEXT,
+                message TEXT
+            )
+            """
+        )
+
+    store = RunStore(settings.database_path)
+    store.initialize()
+    run, _ = store.enqueue(
+        target_kind="task",
+        target_path="checks/status.py",
+        queue_group="checks",
+        input_data={"attempts": 2},
+    )
+
+    assert run.input_json == '{"attempts":2}'