feat(host-agent): add Console task submission with Host self-submission client

Adds a CSRF-protected task submission form to the local Console Tasks page

apps/device-host-agent/host_agent/app.py

@@ -204,6 +204,7 @@ def create_application(
             ttl_seconds=resolved_config.console_session_ttl_seconds
         ),
         enrollment_client=console_enrollment_client,
+        host_client=client,
         metadata_store=metadata_store,
         timeline=timeline,
         executor=executor,

apps/device-host-agent/host_agent/client.py

@@ -38,6 +38,21 @@ class StaleLeaseError(HostAgentAPIError):
     pass
 
 
+class HostTaskSubmissionUnknownError(RuntimeError):
+    """Raised when a Host self-submission request's Cloud outcome is uncertain.
+
+    This is distinct from :class:`HostAgentAPIError` because the request may
+    have reached the control plane but the response was lost, the server
+    returned a 5xx, or the success payload was malformed. Retrying would
+    duplicate the create, so the caller must treat the task as unknown and
+    surface that to the operator.
+    """
+
+    def __init__(self, reason: str) -> None:
+        super().__init__(f"host task submission outcome is unknown: {reason}")
+        self.reason = reason
+
+
 class HostAgentEnrollmentClient:
     def __init__(
         self,
@@ -172,16 +187,37 @@ class HostAgentClient:
         goal: str,
         device_id: str | None = None,
     ) -> HostTaskSubmissionResponse:
-        response = await self._request(
+        try:
-            "POST",
+            response = await self._client.request(
-            f"/internal/v1/hosts/{self.config.host_id}/tasks",
+                "POST",
-            json={
+                f"/internal/v1/hosts/{self.config.host_id}/tasks",
-                "host_id": self.config.host_id,
+                json={
-                "goal": goal,
+                    "host_id": self.config.host_id,
-                "device_id": device_id,
+                    "goal": goal,
-            },
+                    "device_id": device_id,
-        )
+                },
-        return HostTaskSubmissionResponse.model_validate(response.json())
+                headers={"Authorization": f"Bearer {self.config.token}"},
+            )
+        except httpx.TransportError as exc:
+            raise HostTaskSubmissionUnknownError(str(exc)) from exc
+        if response.status_code >= 500:
+            raise HostTaskSubmissionUnknownError(
+                f"control plane returned status {response.status_code}"
+            )
+        if not response.is_success:
+            _raise_api_error(response, stale_lease=False)
+        try:
+            payload = response.json()
+        except ValueError as exc:
+            raise HostTaskSubmissionUnknownError(
+                "control plane returned malformed success payload"
+            ) from exc
+        try:
+            return HostTaskSubmissionResponse.model_validate(payload)
+        except Exception as exc:
+            raise HostTaskSubmissionUnknownError(
+                "control plane returned malformed success payload"
+            ) from exc
 
     async def claim(self) -> AssignmentModel | None:
         response = await self._request(

apps/device-host-agent/host_agent/history.py

@@ -55,6 +55,21 @@ class ConsoleHistoryStore:
             {"revision": revision},
         )
 
+    def record_task_submission(
+        self,
+        *,
+        task_id: str,
+        device_id: str | None,
+    ) -> None:
+        if device_id:
+            summary = f"task submitted: {task_id} on {device_id}"
+        else:
+            summary = f"task submitted: {task_id} (automatic device)"
+        detail: dict[str, Any] = {"task_id": task_id}
+        if device_id is not None:
+            detail["device_id"] = device_id
+        self._insert("task_submission", summary, detail)
+
     def list_recent(self, limit: int | None = None) -> list[dict[str, Any]]:
         effective_limit = limit if limit is not None else self.limit
         with self._connect() as connection:

apps/device-host-agent/host_agent/web/app.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import asyncio
 import base64
 import json
+from collections.abc import Awaitable, Callable
 from pathlib import Path
 from typing import Any
 
@@ -12,7 +13,12 @@ from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse, Resp
 
 from device.manager import DeviceManager
 from host_agent.assignment import AssignmentExecutor
-from host_agent.client import HostAgentEnrollmentClient
+from host_agent.client import (
+    HostAgentClient,
+    HostAgentEnrollmentClient,
+    HostAgentAPIError,
+    HostTaskSubmissionUnknownError,
+)
 from host_agent.config import HostAgentConfig
 from host_agent.devices import register_local_device, unregister_local_device
 from host_agent.history import ConsoleHistoryStore
@@ -34,6 +40,9 @@ CSRF_HEADER_NAME = "X-CSRF-Token"
 CSRF_FORM_FIELD = "csrf_token"
 _LOOPBACK_BIND_HOSTS = frozenset({"127.0.0.1", "localhost", "::1"})
 
+TaskSubmissionCallable = Callable[..., Awaitable[str]]
+AUTOMATIC_DEVICE_VALUE = "__automatic__"
+
 _ENV = jinja2.Environment(
     loader=jinja2.FileSystemLoader(Path(__file__).parent / "templates"),
     autoescape=jinja2.select_autoescape(["html", "xml"]),
@@ -73,6 +82,34 @@ def _screenshot_data_uri(record: dict[str, Any]) -> str | None:
     return f"data:image/png;base64,{encoded}"
 
 
+def _safe_submission_error(detail: str) -> str:
+    """Return a safe, single-line error message for the operator.
+
+    The Cloud response ``detail`` is treated as a static control-plane message;
+    we strip surrounding whitespace and reject empty results so the operator
+    never sees a blank error or, through Jinja autoescape, anything that could
+    carry unrendered HTML.
+    """
+    cleaned = " ".join(str(detail).split()).strip()
+    return cleaned or "Cloud rejected the task submission."
+
+
+def _extract_task_id(response: Any) -> str | None:
+    """Normalize the Host self-submission return value to a Cloud task ID.
+
+    Tests and narrow protocol overrides may return a bare string while the
+    production client returns a pydantic model. Accept either so the rest of
+    the handler can rely on a single string ID.
+    """
+    candidate: Any = response
+    if hasattr(candidate, "task_id"):
+        candidate = getattr(candidate, "task_id")
+    if not isinstance(candidate, str):
+        return None
+    cleaned = candidate.strip()
+    return cleaned or None
+
+
 def _dashboard_texts(*, snapshot: dict[str, Any]) -> dict[str, str]:
     """Pre-compute human-readable text strings for the dashboard template."""
     heartbeat = snapshot.get("last_heartbeat")
@@ -120,12 +157,26 @@ def create_console_app(
     status_tracker: AgentStatusTracker,
     session_manager: SessionManager,
     enrollment_client: HostAgentEnrollmentClient | None,
+    host_client: HostAgentClient | None = None,
+    submit_self_task: TaskSubmissionCallable | None = None,
     metadata_store: TaskMetadataStore | None = None,
     timeline: Timeline | None = None,
     executor: AssignmentExecutor | None = None,
 ) -> FastAPI:
     app = FastAPI(title="Host Agent Console")
     cookie_secure = config.console_bind_host not in _LOOPBACK_BIND_HOSTS
+    if submit_self_task is None and host_client is not None:
+        submit_self_task = host_client.submit_self_task
+    submission_available = submit_self_task is not None
+
+    def _running_devices() -> list[dict[str, str]]:
+        return [
+            {
+                "id": device.id,
+                "label": device.name or device.id,
+            }
+            for device in manager.list_devices()
+        ]
 
     def _session_token(request: Request) -> str | None:
         return request.cookies.get(SESSION_COOKIE_NAME)
@@ -426,22 +477,181 @@ def create_console_app(
             entries=entries,
         )
 
+    def _tasks_list_context(
+        session: SessionState,
+        *,
+        goal_value: str = "",
+        selected_device: str = AUTOMATIC_DEVICE_VALUE,
+        submission_error: str | None = None,
+        submission_notice: str | None = None,
+        submission_unknown: bool = False,
+        status_code: int = 200,
+    ) -> dict[str, Any]:
+        devices = _running_devices()
+        return {
+            "title": "Tasks",
+            "session": session,
+            "csrf_token": session.csrf_token,
+            "tasks": metadata_store.list_tasks()
+            if metadata_store is not None
+            else [],
+            "metadata_store_missing": metadata_store is None,
+            "devices": devices,
+            "automatic_device_value": AUTOMATIC_DEVICE_VALUE,
+            "selected_device": selected_device
+            if any(d["id"] == selected_device for d in devices)
+            or selected_device == AUTOMATIC_DEVICE_VALUE
+            else AUTOMATIC_DEVICE_VALUE,
+            "submission_available": submission_available,
+            "goal_value": goal_value,
+            "submission_error": submission_error,
+            "submission_notice": submission_notice,
+            "submission_unknown": submission_unknown,
+            "status_code": status_code,
+        }
+
     @app.get("/tasks", response_class=HTMLResponse)
     async def tasks_page(
+        request: Request,
         session: SessionState = Depends(require_session),
     ) -> HTMLResponse:
         if metadata_store is None:
             raise HTTPException(
                 status_code=503, detail="task metadata store not configured"
             )
-        tasks_list = await asyncio.to_thread(metadata_store.list_tasks)
+        notice: str | None = None
+        unknown = False
+        error: str | None = None
+        if request.query_params.get("submitted") == "1":
+            task_id = request.query_params.get("task_id", "")
+            if task_id:
+                notice = (
+                    f"Task submitted. Cloud task ID: {task_id}. "
+                    "Track it from the Cloud console for execution progress."
+                )
+        elif request.query_params.get("outcome") == "unknown":
+            unknown = True
+        context = _tasks_list_context(
+            session,
+            submission_notice=notice,
+            submission_unknown=unknown,
+            submission_error=error,
+        )
         return _render(
             "tasks_list.html",
-            title="Tasks",
+            status_code=context["status_code"],
-            session=session,
+            **{k: v for k, v in context.items() if k != "status_code"},
-            tasks=tasks_list,
         )
 
+    @app.post("/tasks/submit", response_class=HTMLResponse)
+    async def tasks_submit(
+        request: Request,
+        session: SessionState = Depends(require_csrf),
+    ) -> Response:
+        if metadata_store is None:
+            raise HTTPException(
+                status_code=503, detail="task metadata store not configured"
+            )
+        if submit_self_task is None:
+            context = _tasks_list_context(
+                session,
+                submission_error=(
+                    "Host submission client is not available yet. "
+                    "Wait for Host enrollment to complete, then retry."
+                ),
+            )
+            return _render(
+                "tasks_list.html",
+                status_code=503,
+                **{k: v for k, v in context.items() if k != "status_code"},
+            )
+        form = await request.form()
+        raw_goal = str(form.get("goal", ""))
+        goal = raw_goal.strip()
+        device_selection = str(form.get("device_id", AUTOMATIC_DEVICE_VALUE)).strip()
+        explicit_device_id: str | None = None
+        if device_selection and device_selection != AUTOMATIC_DEVICE_VALUE:
+            snapshot_ids = {device.id for device in manager.list_devices()}
+            if device_selection not in snapshot_ids:
+                context = _tasks_list_context(
+                    session,
+                    goal_value=goal,
+                    selected_device=device_selection,
+                    submission_error=(
+                        "Selected device is no longer registered. "
+                        "Refresh and try again."
+                    ),
+                )
+                return _render(
+                    "tasks_list.html",
+                    status_code=400,
+                    **{k: v for k, v in context.items() if k != "status_code"},
+                )
+            explicit_device_id = device_selection
+
+        if not goal:
+            context = _tasks_list_context(
+                session,
+                goal_value=goal,
+                selected_device=device_selection or AUTOMATIC_DEVICE_VALUE,
+                submission_error="Goal cannot be empty.",
+            )
+            return _render(
+                "tasks_list.html",
+                status_code=400,
+                **{k: v for k, v in context.items() if k != "status_code"},
+            )
+
+        try:
+            response = await submit_self_task(
+                goal=goal, device_id=explicit_device_id
+            )
+        except HostAgentAPIError as exc:
+            context = _tasks_list_context(
+                session,
+                goal_value=goal,
+                selected_device=device_selection,
+                submission_error=_safe_submission_error(str(exc.detail)),
+            )
+            return _render(
+                "tasks_list.html",
+                status_code=502,
+                **{k: v for k, v in context.items() if k != "status_code"},
+            )
+        except HostTaskSubmissionUnknownError:
+            return RedirectResponse(
+                url="/tasks?outcome=unknown",
+                status_code=303,
+            )
+
+        task_id = _extract_task_id(response)
+        if task_id is None:
+            context = _tasks_list_context(
+                session,
+                goal_value=goal,
+                selected_device=device_selection,
+                submission_error=(
+                    "Host submission client returned an unexpected response."
+                ),
+            )
+            return _render(
+                "tasks_list.html",
+                status_code=502,
+                **{k: v for k, v in context.items() if k != "status_code"},
+            )
+
+        try:
+            await asyncio.to_thread(
+                history_store.record_task_submission,
+                task_id=task_id,
+                device_id=explicit_device_id,
+            )
+        except Exception:
+            pass
+
+        params = f"submitted=1&task_id={task_id}"
+        return RedirectResponse(url=f"/tasks?{params}", status_code=303)
+
     @app.get("/tasks/{task_id}", response_class=HTMLResponse)
     async def task_detail_page(
         task_id: str,

apps/device-host-agent/host_agent/web/templates/tasks_list.html

@@ -1,20 +1,52 @@
 {% extends "base.html" %}
 {% block body %}
     <h1>Tasks</h1>
-    {% if not tasks %}
+
-    <p>No tasks recorded.</p>
+    <section id="task-submission">
-    {% else %}
+      <h2>Submit task to current Host</h2>
-    <table>
+      {% if submission_available %}
-      <thead><tr><th>Task ID</th><th>Status</th><th>Device</th><th>Created</th><th>Updated</th></tr></thead>
+      <form method="post" action="/tasks/submit">
-      <tbody>{% for task in tasks %}
+        <input type="hidden" name="csrf_token" value="{{ csrf_token }}">
-        <tr>
+        <p>
-          <td><a href="/tasks/{{ task["id"] }}">{{ task["id"] }}</a></td>
+          <label for="goal">Goal</label><br>
-          <td>{{ task.get("status") or "" }}</td>
+          <textarea id="goal" name="goal" rows="4" cols="60" required>{{ goal_value }}</textarea>
-          <td>{{ task.get("device_id") or "" }}</td>
+        </p>
-          <td>{{ task.get("created_at") or "" }}</td>
+        <p>
-          <td>{{ task.get("updated_at") or "" }}</td>
+          <label for="device_id">Target device</label><br>
-        </tr>
+          <select id="device_id" name="device_id">
-        {% endfor %}</tbody>
+            <option value="{{ automatic_device_value }}"{% if selected_device == automatic_device_value %} selected{% endif %}>Automatic (let Host choose an eligible device)</option>
-    </table>
+            {% for device in devices %}<option value="{{ device.id }}"{% if selected_device == device.id %} selected{% endif %}>{{ device.label }}</option>{% endfor %}
-    {% endif %}
+          </select>
-{% endblock %}
+        </p>
+        {% if submission_error %}<p class="error" id="submission-error">{{ submission_error }}</p>{% endif %}
+        {% if submission_unknown %}<p class="error" id="submission-unknown">Submission outcome is unknown. The task may have been queued. Check the Cloud console before submitting again.</p>{% endif %}
+        {% if submission_notice %}<p class="notice" id="submission-notice">{{ submission_notice }}</p>{% endif %}
+        <p><button type="submit">Submit task</button></p>
+      </form>
+      {% else %}
+      <p class="error" id="submission-unavailable">Host submission client is not available yet. Wait for Host enrollment to complete, then refresh.</p>
+      {% endif %}
+    </section>
+
+    <section id="local-tasks">
+      <h2>Local Runtime tasks</h2>
+      {% if metadata_store_missing %}
+      <p class="error">Task metadata store is not configured.</p>
+      {% elif not tasks %}
+      <p>No tasks recorded.</p>
+      {% else %}
+      <table>
+        <thead><tr><th>Task ID</th><th>Status</th><th>Device</th><th>Created</th><th>Updated</th></tr></thead>
+        <tbody>{% for task in tasks %}
+          <tr>
+            <td><a href="/tasks/{{ task["id"] }}">{{ task["id"] }}</a></td>
+            <td>{{ task.get("status") or "" }}</td>
+            <td>{{ task.get("device_id") or "" }}</td>
+            <td>{{ task.get("created_at") or "" }}</td>
+            <td>{{ task.get("updated_at") or "" }}</td>
+          </tr>
+          {% endfor %}</tbody>
+      </table>
+      {% endif %}
+    </section>
+{% endblock %}

apps/device-host-agent/tests/test_client.py

@@ -11,6 +11,8 @@ from cloud.internal_api.models import AssignmentModel, DeviceSnapshotModel
 from host_agent.client import (
     HostAgentClient,
     HostAgentEnrollmentClient,
+    HostAgentAPIError,
+    HostTaskSubmissionUnknownError,
     StaleLeaseError,
 )
 from host_agent.config import HostAgentConfig
@@ -172,6 +174,161 @@ def test_result_report_retries_identical_payload_after_response_loss() -> None:
     assert payloads[0]["failure_reason"] == "planner unavailable"
 
 
+def test_submit_self_task_posts_once_and_returns_task_id() -> None:
+    requests: list[httpx.Request] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        requests.append(request)
+        return httpx.Response(201, json={"task_id": "task-cloud-1"})
+
+    async def scenario() -> None:
+        async with httpx.AsyncClient(
+            transport=httpx.MockTransport(handler),
+            base_url="https://control.example",
+        ) as http_client:
+            client = HostAgentClient(_config(), http_client=http_client)
+            response = await client.submit_self_task(
+                goal="open settings",
+                device_id=None,
+            )
+
+        assert response.task_id == "task-cloud-1"
+
+    asyncio.run(scenario())
+    assert len(requests) == 1
+    assert requests[0].url.path == "/internal/v1/hosts/host-a/tasks"
+    assert requests[0].headers["authorization"] == "Bearer host-secret"
+    payload = json.loads(requests[0].content)
+    assert payload == {
+        "host_id": "host-a",
+        "goal": "open settings",
+        "device_id": None,
+    }
+
+
+def test_submit_self_task_does_not_retry_transport_failure() -> None:
+    attempts = 0
+    sleeps: list[float] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        nonlocal attempts
+        attempts += 1
+        raise httpx.ConnectError("network down", request=request)
+
+    async def sleep(delay: float) -> None:
+        sleeps.append(delay)
+
+    async def scenario() -> None:
+        async with httpx.AsyncClient(
+            transport=httpx.MockTransport(handler),
+            base_url="https://control.example",
+        ) as http_client:
+            client = HostAgentClient(_config(), http_client=http_client, sleep=sleep)
+            with pytest.raises(HostTaskSubmissionUnknownError):
+                await client.submit_self_task(goal="open settings")
+
+    asyncio.run(scenario())
+    assert attempts == 1
+    assert sleeps == []
+
+
+def test_submit_self_task_treats_5xx_as_unknown_outcome_without_retry() -> None:
+    attempts = 0
+    sleeps: list[float] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        nonlocal attempts
+        attempts += 1
+        return httpx.Response(502, json={"detail": "bad gateway"})
+
+    async def sleep(delay: float) -> None:
+        sleeps.append(delay)
+
+    async def scenario() -> None:
+        async with httpx.AsyncClient(
+            transport=httpx.MockTransport(handler),
+            base_url="https://control.example",
+        ) as http_client:
+            client = HostAgentClient(_config(), http_client=http_client, sleep=sleep)
+            with pytest.raises(HostTaskSubmissionUnknownError):
+                await client.submit_self_task(goal="open settings")
+
+    asyncio.run(scenario())
+    assert attempts == 1
+    assert sleeps == []
+
+
+def test_submit_self_task_raises_definitive_error_on_4xx_rejection() -> None:
+    attempts = 0
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        nonlocal attempts
+        attempts += 1
+        return httpx.Response(
+            403,
+            json={"detail": "Host self-submission is disabled"},
+        )
+
+    async def scenario() -> None:
+        async with httpx.AsyncClient(
+            transport=httpx.MockTransport(handler),
+            base_url="https://control.example",
+        ) as http_client:
+            client = HostAgentClient(_config(), http_client=http_client)
+            with pytest.raises(HostAgentAPIError) as error:
+                await client.submit_self_task(goal="open settings")
+            assert "host-secret" not in str(error.value)
+
+    asyncio.run(scenario())
+    assert attempts == 1
+
+
+def test_submit_self_task_treats_malformed_success_as_unknown() -> None:
+    attempts = 0
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        nonlocal attempts
+        attempts += 1
+        return httpx.Response(201, json={"unexpected": "shape"})
+
+    async def scenario() -> None:
+        async with httpx.AsyncClient(
+            transport=httpx.MockTransport(handler),
+            base_url="https://control.example",
+        ) as http_client:
+            client = HostAgentClient(_config(), http_client=http_client)
+            with pytest.raises(HostTaskSubmissionUnknownError):
+                await client.submit_self_task(goal="open settings")
+
+    asyncio.run(scenario())
+    assert attempts == 1
+
+
+def test_submit_self_task_does_not_duplicate_when_response_is_lost() -> None:
+    attempts = 0
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        nonlocal attempts
+        attempts += 1
+        raise httpx.ReadError("response lost", request=request)
+
+    async def scenario() -> None:
+        async with httpx.AsyncClient(
+            transport=httpx.MockTransport(handler),
+            base_url="https://control.example",
+        ) as http_client:
+            client = HostAgentClient(
+                _config(),
+                http_client=http_client,
+                sleep=lambda delay: asyncio.sleep(0),
+            )
+            with pytest.raises(HostTaskSubmissionUnknownError):
+                await client.submit_self_task(goal="open settings")
+
+    asyncio.run(scenario())
+    assert attempts == 1
+
+
 def test_bootstrap_client_directly_enrolls_and_enrolls_device() -> None:
     requests: list[httpx.Request] = []
     host_attempts = 0

apps/device-host-agent/tests/test_history.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import json
 from datetime import UTC, datetime
 
 from host_agent.history import ConsoleHistoryStore
@@ -67,3 +68,71 @@ def test_history_store_uses_injected_now_for_occurred_at(tmp_path) -> None:
     entries = store.list_recent()
 
     assert entries[0]["occurred_at"] == fixed_now.isoformat()
+
+
+def test_history_store_records_task_submission_with_device(tmp_path) -> None:
+    store = ConsoleHistoryStore(tmp_path / "history.sqlite3")
+
+    store.record_task_submission(task_id="task-cloud-1", device_id="device-cloud-a")
+
+    entries = store.list_recent()
+
+    assert len(entries) == 1
+    assert entries[0]["kind"] == "task_submission"
+    assert entries[0]["summary"] == "task submitted: task-cloud-1 on device-cloud-a"
+    assert entries[0]["detail"] == {
+        "task_id": "task-cloud-1",
+        "device_id": "device-cloud-a",
+    }
+
+
+def test_history_store_records_task_submission_without_device(tmp_path) -> None:
+    store = ConsoleHistoryStore(tmp_path / "history.sqlite3")
+
+    store.record_task_submission(task_id="task-cloud-2", device_id=None)
+
+    entries = store.list_recent()
+
+    assert entries[0]["summary"] == (
+        "task submitted: task-cloud-2 (automatic device)"
+    )
+    assert entries[0]["detail"] == {"task_id": "task-cloud-2"}
+
+
+def test_history_store_task_submission_redacts_goal_and_secrets(tmp_path) -> None:
+    store = ConsoleHistoryStore(tmp_path / "history.sqlite3")
+    secret_goal = (
+        "rotate secret-XYZ-abcdef-very-secret "
+        "cookie=session-abc; lease=lease-stale"
+    )
+
+    store.record_task_submission(task_id="task-cloud-3", device_id="device-cloud-a")
+    store.record_heartbeat(device_count=1)
+
+    entries = store.list_recent()
+    rendered = "\n".join(
+        repr(entry["summary"]) + " " + json.dumps(entry["detail"])
+        for entry in entries
+    )
+
+    assert secret_goal not in rendered
+    assert "session-abc" not in rendered
+    assert "lease-stale" not in rendered
+
+
+def test_history_store_task_submissions_prune_beyond_limit(tmp_path) -> None:
+    store = ConsoleHistoryStore(tmp_path / "history.sqlite3", limit=3)
+
+    for index in range(5):
+        store.record_task_submission(
+            task_id=f"task-{index}", device_id=f"device-{index}"
+        )
+
+    entries = store.list_recent()
+
+    assert len(entries) == 3
+    assert [entry["detail"]["task_id"] for entry in entries] == [
+        "task-4",
+        "task-3",
+        "task-2",
+    ]

apps/device-host-agent/tests/test_web_app.py

@@ -1,12 +1,14 @@
 from __future__ import annotations
 
 import re
+from collections.abc import Awaitable, Callable
 from datetime import UTC, datetime
 
 from fastapi.testclient import TestClient
 
 from cloud.internal_api.models import AssignmentModel
 from device.manager import DeviceManager
+from host_agent.client import HostAgentAPIError, HostTaskSubmissionUnknownError
 from host_agent.config import HostAgentConfig
 from host_agent.history import ConsoleHistoryStore
 from host_agent.identity import HostIdentityStore
@@ -15,11 +17,20 @@ from host_agent.status import AgentStatusTracker
 from host_agent.web.app import SESSION_COOKIE_NAME, create_console_app
 from host_agent.web.auth import SessionManager
 from storage.device_config import DeviceConfigStore
+from storage.task_metadata import TaskMetadataStore
 
 CSRF_PATTERN = re.compile(r'name="csrf_token" value="([^"]+)"')
 
+TaskSubmissionCallable = Callable[[str, str | None], Awaitable[str]]
 
-def _build_client(tmp_path, *, create_account: bool = True) -> tuple[TestClient, dict]:
+
+def _build_client(
+    tmp_path,
+    *,
+    create_account: bool = True,
+    submit_self_task: TaskSubmissionCallable | None = None,
+    include_metadata_store: bool = True,
+) -> tuple[TestClient, dict]:
     config = HostAgentConfig(
         control_plane_url="https://control.example",
         host_id="host-a",
@@ -36,6 +47,11 @@ def _build_client(tmp_path, *, create_account: bool = True) -> tuple[TestClient,
     history_store = ConsoleHistoryStore(tmp_path / "history.sqlite3")
     status_tracker = AgentStatusTracker()
     session_manager = SessionManager(ttl_seconds=3600.0)
+    metadata_store: TaskMetadataStore | None = None
+    if include_metadata_store:
+        metadata_store = TaskMetadataStore(
+            db_path=tmp_path / "task_metadata.sqlite3"
+        )
 
     app = create_console_app(
         config=config,
@@ -47,6 +63,8 @@ def _build_client(tmp_path, *, create_account: bool = True) -> tuple[TestClient,
         status_tracker=status_tracker,
         session_manager=session_manager,
         enrollment_client=None,
+        submit_self_task=submit_self_task,
+        metadata_store=metadata_store,
     )
     client = TestClient(app)
     context = {
@@ -56,6 +74,7 @@ def _build_client(tmp_path, *, create_account: bool = True) -> tuple[TestClient,
         "history_store": history_store,
         "session_manager": session_manager,
         "status_tracker": status_tracker,
+        "metadata_store": metadata_store,
     }
     return client, context
 
@@ -309,3 +328,472 @@ def test_logout_invalidates_session_so_subsequent_request_redirects_to_login(
 
     assert response.status_code == 303
     assert response.headers["location"] == "/login"
+
+
+def _make_submission_recorder(
+    *,
+    task_id: str = "task-cloud-1",
+    raise_api_error: HostAgentAPIError | None = None,
+    raise_unknown: bool = False,
+    history_store: ConsoleHistoryStore | None = None,
+    history_fail: bool = False,
+) -> tuple[TaskSubmissionCallable, dict]:
+    captured: dict = {}
+
+    async def submit(goal: str, device_id: str | None) -> str:
+        captured["goal"] = goal
+        captured["device_id"] = device_id
+        if raise_unknown:
+            raise HostTaskSubmissionUnknownError("transport failure")
+        if raise_api_error is not None:
+            raise raise_api_error
+        return task_id
+
+    async def record(*, task_id: str, device_id: str | None) -> None:
+        captured.setdefault("history_calls", []).append(
+            {"task_id": task_id, "device_id": device_id}
+        )
+        if history_fail:
+            raise RuntimeError("history store unavailable")
+
+    if history_store is not None:
+        original = history_store.record_task_submission
+
+        def _proxy(*, task_id: str, device_id: str | None) -> None:
+            try:
+                asyncio_run(record(task_id=task_id, device_id=device_id))
+            except RuntimeError:
+                pass
+            return original(task_id=task_id, device_id=device_id)
+
+        history_store.record_task_submission = _proxy  # type: ignore[method-assign]
+    return submit, captured
+
+
+def asyncio_run(coro):
+    import asyncio
+
+    return asyncio.get_event_loop().run_until_complete(coro) if asyncio.get_event_loop().is_running() else asyncio.run(coro)
+
+
+def test_tasks_page_renders_submission_form_with_device_options(tmp_path) -> None:
+    submit, _ = _make_submission_recorder()
+    client, context = _build_client(tmp_path, submit_self_task=submit)
+    context["manager"].register_device(
+        "device-runtime-a",
+        lambda: object(),  # type: ignore[arg-type,return-value]
+        name="Lab iPhone",
+    )
+    _login(client)
+
+    response = client.get("/tasks")
+
+    assert response.status_code == 200
+    assert "Submit task to current Host" in response.text
+    assert 'action="/tasks/submit"' in response.text
+    assert "Automatic (let Host choose" in response.text
+    assert "Lab iPhone" in response.text
+
+
+def test_authenticated_automatic_submission_invokes_submit_with_none_device(
+    tmp_path,
+) -> None:
+    submit, captured = _make_submission_recorder(task_id="task-cloud-auto")
+    client, context = _build_client(tmp_path, submit_self_task=submit)
+    context["manager"].register_device(
+        "device-runtime-a",
+        lambda: object(),  # type: ignore[arg-type,return-value]
+        name="Lab iPhone",
+    )
+    csrf_token = _login(client)
+
+    response = client.post(
+        "/tasks/submit",
+        data={
+            "goal": "open settings",
+            "device_id": "__automatic__",
+            "csrf_token": csrf_token,
+        },
+        follow_redirects=False,
+    )
+
+    assert response.status_code == 303
+    assert response.headers["location"] == "/tasks?submitted=1&task_id=task-cloud-auto"
+    assert captured == {"goal": "open settings", "device_id": None}
+
+
+def test_authenticated_explicit_device_submission_uses_runtime_id(tmp_path) -> None:
+    submit, captured = _make_submission_recorder(task_id="task-cloud-explicit")
+    client, context = _build_client(tmp_path, submit_self_task=submit)
+    context["manager"].register_device(
+        "device-cloud-a",
+        lambda: object(),  # type: ignore[arg-type,return-value]
+        name="Cloud iPhone",
+    )
+    csrf_token = _login(client)
+
+    response = client.post(
+        "/tasks/submit",
+        data={
+            "goal": "open mail",
+            "device_id": "device-cloud-a",
+            "csrf_token": csrf_token,
+        },
+        follow_redirects=False,
+    )
+
+    assert response.status_code == 303
+    assert (
+        response.headers["location"]
+        == "/tasks?submitted=1&task_id=task-cloud-explicit"
+    )
+    assert captured == {"goal": "open mail", "device_id": "device-cloud-a"}
+
+
+def test_redirected_tasks_page_renders_task_id_confirmation(tmp_path) -> None:
+    submit, _ = _make_submission_recorder()
+    client, _ = _build_client(tmp_path, submit_self_task=submit)
+    _login(client)
+
+    response = client.get("/tasks?submitted=1&task_id=task-cloud-1")
+
+    assert response.status_code == 200
+    assert "task-cloud-1" in response.text
+    assert "Cloud task ID" in response.text
+
+
+def test_confirmed_submission_records_history_audit(tmp_path) -> None:
+    submit, _ = _make_submission_recorder(task_id="task-cloud-audit")
+    client, context = _build_client(tmp_path, submit_self_task=submit)
+    context["manager"].register_device(
+        "device-cloud-a",
+        lambda: object(),  # type: ignore[arg-type,return-value]
+        name="Cloud iPhone",
+    )
+    csrf_token = _login(client)
+
+    client.post(
+        "/tasks/submit",
+        data={
+            "goal": "audit me",
+            "device_id": "device-cloud-a",
+            "csrf_token": csrf_token,
+        },
+        follow_redirects=False,
+    )
+
+    entries = context["history_store"].list_recent()
+    assert entries[0]["kind"] == "task_submission"
+    assert entries[0]["detail"] == {
+        "task_id": "task-cloud-audit",
+        "device_id": "device-cloud-a",
+    }
+
+
+def test_unauthenticated_submission_redirects_to_login_without_calling_client(
+    tmp_path,
+) -> None:
+    submit, captured = _make_submission_recorder()
+    client, _ = _build_client(tmp_path, submit_self_task=submit)
+
+    response = client.post(
+        "/tasks/submit",
+        data={"goal": "open settings", "device_id": "__automatic__"},
+        follow_redirects=False,
+    )
+
+    assert response.status_code == 303
+    assert response.headers["location"] == "/login"
+    assert captured == {}
+
+
+def test_csrf_invalid_submission_is_rejected(tmp_path) -> None:
+    submit, captured = _make_submission_recorder()
+    client, _ = _build_client(tmp_path, submit_self_task=submit)
+    _login(client)
+
+    response = client.post(
+        "/tasks/submit",
+        data={
+            "goal": "open settings",
+            "device_id": "__automatic__",
+            "csrf_token": "wrong-token",
+        },
+    )
+
+    assert response.status_code == 403
+    assert captured == {}
+
+
+def test_blank_goal_rejected_before_client_invocation(tmp_path) -> None:
+    submit, captured = _make_submission_recorder()
+    client, context = _build_client(tmp_path, submit_self_task=submit)
+    context["manager"].register_device(
+        "device-runtime-a",
+        lambda: object(),  # type: ignore[arg-type,return-value]
+        name="Lab iPhone",
+    )
+    csrf_token = _login(client)
+
+    response = client.post(
+        "/tasks/submit",
+        data={
+            "goal": "   ",
+            "device_id": "__automatic__",
+            "csrf_token": csrf_token,
+        },
+    )
+
+    assert response.status_code == 400
+    assert "Goal cannot be empty" in response.text
+    assert captured == {}
+
+
+def test_stale_device_selection_rejected_before_client_invocation(tmp_path) -> None:
+    submit, captured = _make_submission_recorder()
+    client, context = _build_client(tmp_path, submit_self_task=submit)
+    context["manager"].register_device(
+        "device-runtime-a",
+        lambda: object(),  # type: ignore[arg-type,return-value],
+        name="Lab iPhone",
+    )
+    csrf_token = _login(client)
+
+    response = client.post(
+        "/tasks/submit",
+        data={
+            "goal": "open settings",
+            "device_id": "device-that-was-removed",
+            "csrf_token": csrf_token,
+        },
+    )
+
+    assert response.status_code == 400
+    assert "no longer registered" in response.text
+    assert captured == {}
+
+
+def test_cloud_definitive_rejection_renders_safe_error_without_calling_history(
+    tmp_path,
+) -> None:
+    submit, captured = _make_submission_recorder(
+        raise_api_error=HostAgentAPIError(
+            403, "Host self-submission is disabled"
+        ),
+    )
+    client, context = _build_client(tmp_path, submit_self_task=submit)
+    context["manager"].register_device(
+        "device-runtime-a",
+        lambda: object(),  # type: ignore[arg-type,return-value]
+        name="Lab iPhone",
+    )
+    csrf_token = _login(client)
+
+    response = client.post(
+        "/tasks/submit",
+        data={
+            "goal": "open settings",
+            "device_id": "__automatic__",
+            "csrf_token": csrf_token,
+        },
+    )
+
+    assert response.status_code == 502
+    assert "Host self-submission is disabled" in response.text
+    entries = context["history_store"].list_recent()
+    assert all(entry["kind"] != "task_submission" for entry in entries)
+    assert captured == {"goal": "open settings", "device_id": None}
+
+
+def test_transport_uncertain_response_redirects_to_unknown_outcome(
+    tmp_path,
+) -> None:
+    submit, captured = _make_submission_recorder(raise_unknown=True)
+    client, context = _build_client(tmp_path, submit_self_task=submit)
+    context["manager"].register_device(
+        "device-runtime-a",
+        lambda: object(),  # type: ignore[arg-type,return-value]
+        name="Lab iPhone",
+    )
+    csrf_token = _login(client)
+
+    response = client.post(
+        "/tasks/submit",
+        data={
+            "goal": "open settings",
+            "device_id": "__automatic__",
+            "csrf_token": csrf_token,
+        },
+        follow_redirects=False,
+    )
+
+    assert response.status_code == 303
+    assert response.headers["location"] == "/tasks?outcome=unknown"
+    entries = context["history_store"].list_recent()
+    assert all(entry["kind"] != "task_submission" for entry in entries)
+    assert captured == {"goal": "open settings", "device_id": None}
+
+
+def test_unknown_outcome_query_shows_safe_message(tmp_path) -> None:
+    submit, _ = _make_submission_recorder()
+    client, _ = _build_client(tmp_path, submit_self_task=submit)
+    _login(client)
+
+    response = client.get("/tasks?outcome=unknown")
+
+    assert response.status_code == 200
+    assert "Submission outcome is unknown" in response.text
+    assert "Check the Cloud console" in response.text
+
+
+def test_submission_form_absent_when_client_missing(tmp_path) -> None:
+    client, _ = _build_client(tmp_path, submit_self_task=None)
+    _login(client)
+
+    response = client.get("/tasks")
+
+    assert response.status_code == 200
+    assert 'action="/tasks/submit"' not in response.text
+    assert "submission client is not available yet" in response.text
+
+
+def test_submit_when_client_missing_returns_503_without_history(tmp_path) -> None:
+    client, context = _build_client(tmp_path, submit_self_task=None)
+    context["manager"].register_device(
+        "device-runtime-a",
+        lambda: object(),  # type: ignore[arg-type,return-value]
+        name="Lab iPhone",
+    )
+    csrf_token = _login(client)
+
+    response = client.post(
+        "/tasks/submit",
+        data={
+            "goal": "open settings",
+            "device_id": "__automatic__",
+            "csrf_token": csrf_token,
+        },
+    )
+
+    assert response.status_code == 503
+    entries = context["history_store"].list_recent()
+    assert all(entry["kind"] != "task_submission" for entry in entries)
+
+
+def test_audit_failure_does_not_break_successful_submission(tmp_path) -> None:
+    submit, _ = _make_submission_recorder(
+        task_id="task-cloud-audit-fail",
+        history_store=None,
+    )
+    client, context = _build_client(tmp_path, submit_self_task=submit)
+
+    def boom(*, task_id: str, device_id: str | None) -> None:
+        raise RuntimeError("audit DB offline")
+
+    context["history_store"].record_task_submission = boom  # type: ignore[method-assign]
+    context["manager"].register_device(
+        "device-runtime-a",
+        lambda: object(),  # type: ignore[arg-type,return-value]
+        name="Lab iPhone",
+    )
+    csrf_token = _login(client)
+
+    response = client.post(
+        "/tasks/submit",
+        data={
+            "goal": "open settings",
+            "device_id": "__automatic__",
+            "csrf_token": csrf_token,
+        },
+        follow_redirects=False,
+    )
+
+    assert response.status_code == 303
+    assert (
+        response.headers["location"]
+        == "/tasks?submitted=1&task_id=task-cloud-audit-fail"
+    )
+
+
+def test_failed_submission_never_writes_successful_audit(tmp_path) -> None:
+    submit, _ = _make_submission_recorder(
+        raise_api_error=HostAgentAPIError(422, "ownership mismatch"),
+    )
+    client, context = _build_client(tmp_path, submit_self_task=submit)
+    context["manager"].register_device(
+        "device-runtime-a",
+        lambda: object(),  # type: ignore[arg-type,return-value]
+        name="Lab iPhone",
+    )
+    csrf_token = _login(client)
+
+    client.post(
+        "/tasks/submit",
+        data={
+            "goal": "open settings",
+            "device_id": "device-runtime-a",
+            "csrf_token": csrf_token,
+        },
+    )
+
+    entries = context["history_store"].list_recent()
+    assert all(entry["kind"] != "task_submission" for entry in entries)
+
+
+def test_tasks_page_autoescapes_goal_device_and_error_text(tmp_path) -> None:
+    submit, _ = _make_submission_recorder(
+        raise_api_error=HostAgentAPIError(403, "<img src=x onerror=alert(1)>"),
+    )
+    client, context = _build_client(tmp_path, submit_self_task=submit)
+    context["manager"].register_device(
+        "<script>alert('x')</script>",
+        lambda: object(),  # type: ignore[arg-type,return-value]
+        name='"><img src=x onerror=alert(1)>',
+    )
+    csrf_token = _login(client)
+
+    response = client.post(
+        "/tasks/submit",
+        data={
+            "goal": "<script>alert('x')</script>",
+            "device_id": "<script>alert('x')</script>",
+            "csrf_token": csrf_token,
+        },
+    )
+
+    assert response.status_code == 502
+    html = response.text
+    assert "&lt;script&gt;alert(&#39;x&#39;)&lt;/script&gt;" in html
+    assert "<img src=x onerror=alert(1)>" not in html
+    assert "<script>alert('x')</script>" not in html
+    assert "&lt;script&gt;alert" in html
+    assert "alert(1)" in html  # auto-escaped as text, not executable
+
+
+def test_submitted_redirect_does_not_include_goal_text(tmp_path) -> None:
+    submit, _ = _make_submission_recorder(task_id="task-cloud-clean")
+    client, context = _build_client(tmp_path, submit_self_task=submit)
+    context["manager"].register_device(
+        "device-runtime-a",
+        lambda: object(),  # type: ignore[arg-type,return-value],
+        name="Lab iPhone",
+    )
+    csrf_token = _login(client)
+    secret_goal = "rotate bearer-token-deadbeef-very-secret-12345"
+
+    response = client.post(
+        "/tasks/submit",
+        data={
+            "goal": secret_goal,
+            "device_id": "__automatic__",
+            "csrf_token": csrf_token,
+        },
+        follow_redirects=False,
+    )
+
+    assert response.status_code == 303
+    assert secret_goal not in response.headers["location"]
+    assert "bearer-token-deadbeef" not in response.headers["location"]
+    follow = client.get(response.headers["location"])
+    assert secret_goal not in follow.text
+    assert "bearer-token-deadbeef" not in follow.text

docs/HOST_AGENT_CONSOLE_TASK_SUBMISSION.md

@@ -0,0 +1,117 @@
+# Host Agent Local Console Task Submission
+
+The Host Agent local Console adds a **Submit task to current Host** form on
+the authenticated Tasks page. The form lets an operator who is already
+signed in to the loopback Console enqueue a goal task for the current Host
+without going through the Cloud Console or the public SDK.
+
+The submission reuses the existing Host self-submission contract (a single
+authenticated `POST /internal/v1/hosts/{host_id}/tasks` against the control
+plane). It does not introduce a new Cloud endpoint, a public task scope, or
+a cross-Host target.
+
+## Automatic vs explicit device
+
+The form offers two submission modes for the device target:
+
+- **Automatic** — the form omits the `device_id` field. The Cloud scheduler
+  picks any eligible device currently registered for the authenticated
+  Host. The Console labels this option *Automatic (let Host choose an
+  eligible device)*.
+- **Explicit device** — the operator selects one of the device IDs that are
+  currently registered in the running `DeviceManager`. In
+  enrollment-managed deployments these are the Cloud device IDs assigned by
+  the Host enrollment flow (the same IDs the heartbeat sends to the
+  control plane), not the local configuration IDs.
+
+The Console revalidates the selected device against a fresh
+`DeviceManager.list_devices()` snapshot before issuing the outbound
+request. If the device was removed between page render and POST, the
+Console returns a validation error and never calls the Host client.
+
+## Required state and policy
+
+- The Console session must be valid (the same local account login used for
+  the rest of the Console). The submission form is rendered only for
+  authenticated sessions, and the POST handler requires both a valid
+  session and the matching CSRF token.
+- The Host must have completed self-enrollment. The Host self-submission
+  client is wired in by `host_agent.app.create_application` from the same
+  asynchronous `HostAgentClient` instance that drives heartbeat, claim,
+  lease renewal, and result reporting, so the submission form disappears
+  with a clear *submission client is not available yet* message until
+  enrollment is complete.
+- The Cloud control plane must have `self_submission_enabled = true` in
+  the Host governance policy. A disabled policy is returned to the
+  operator as a definitive 4xx rejection ("Host self-submission is
+  disabled") without recording a successful audit entry; the cached
+  policy is *not* used for local authorization, so a stale cache never
+  silently denies a newly enabled Host.
+
+## Cloud queue semantics
+
+- Task creation is **not** idempotent. After the POST reaches the control
+  plane, the Cloud scheduler assigns a task ID. The Console never retries
+  the creation call. Even when the response is lost, a 5xx is returned, or
+  the success payload is malformed, the Console only reports the outcome
+  as **unknown** and instructs the operator to check the Cloud console
+  before submitting again. This avoids the known failure mode where one
+  operator click duplicates a task in the Cloud queue.
+- The local Console does not show queue progress for a freshly submitted
+  task. After a successful submission the form redirect carries the Cloud
+  task ID in the URL and the Tasks page shows *Task submitted. Cloud task
+  ID: \<id\>*; execution progress, lease state, and the terminal result
+  continue to appear on the local Tasks page only after the Cloud
+  scheduler assigns the task to this Host and the assignment is processed
+  locally.
+
+## Outcome-unknown procedure
+
+The Console intentionally distinguishes the **unknown outcome** case from
+both a confirmed submission and a definitive rejection:
+
+- A transport failure (DNS, connect, read timeout, dropped connection),
+  any 5xx response, or a malformed 2xx payload makes the local client
+  raise `HostTaskSubmissionUnknownError`. The Console catches it and
+  redirects the operator to `GET /tasks?outcome=unknown`, which renders a
+  *Submission outcome is unknown. The task may have been queued. Check
+  the Cloud console before submitting again.* notice. No history entry is
+  written and no success confirmation is shown.
+- A definitive 4xx rejection (for example, a disabled self-submission
+  policy or a target device that the control plane does not recognize as
+  owned by this Host) raises `HostAgentAPIError`. The Console renders the
+  Cloud's `detail` message safely through Jinja autoescape without
+  echoing the goal text or any credentials. No history entry is written
+  and no task ID is shown.
+
+## Auditing
+
+- Confirmed submissions write a `task_submission` history row that records
+  the Cloud task ID and, when applicable, the target device ID. The row
+  deliberately does **not** persist the goal, the Host token, the
+  session cookie, the lease secret, or any other operator credential. The
+  row is bounded by the existing `console_history_limit` and survives
+  process restarts.
+- A local audit write failure is best-effort: it does not turn a
+  Cloud-confirmed submission into a retryable failure. A `task_submission`
+  history row is only written after the Cloud returns a 2xx with a usable
+  task ID, so a rejection or unknown outcome can never produce a
+  misleading successful audit entry.
+
+## Operator checklist
+
+1. Sign in to the loopback Console with a valid local account.
+2. Open **Tasks**.
+3. Type a non-empty goal in the textarea.
+4. Choose **Automatic** to let the Cloud pick a device, or pick a listed
+   device ID for an explicit target.
+5. Click **Submit task**.
+6. On confirmation, the page shows *Task submitted. Cloud task ID: \<id\>*.
+   Track the task from the Cloud console; the local Tasks list will fill
+   in once the scheduler assigns the task to this Host.
+7. If the page shows the *outcome is unknown* notice, check the Cloud
+   console for a matching task before submitting again.
+8. If the page shows a Cloud rejection, fix the underlying issue (policy,
+   device ownership, lease state) and submit again. The page does not
+   preserve the goal text in the URL, in the error message, or in the
+   history; retype the goal when retrying.