From ec261d57c2c00d8c3ceb3c629eb73e91f81e2ab1 Mon Sep 17 00:00:00 2001 From: Jerry Yan <792602257@qq.com> Date: Tue, 14 Jul 2026 12:47:49 +0800 Subject: [PATCH] feat: surface task execution progress across Host Agent and Cloud Host Agent now persists step-level execution detail locally (via a real TaskMetadataStore/Timeline wired into TaskRunner) and reports a bounded in-progress snapshot piggybacked on lease renewal. Cloud persists that snapshot per active assignment and exposes it through the existing task list/detail query path; Cloud Console renders it as a live badge. Host Agent's local console gains authenticated, read-only task list and detail/timeline pages (same-origin, server-rendered) with inlined screenshots. Also fixes a pre-existing gap in the shared Timeline: the actual per-step LLM prompt is now recorded instead of the task goal, benefiting both Runtime and Host Agent consoles. When a host uses the cloud planner transport, each decide call's prompt and resulting tool decision are durably logged in a new planner_decision_log table (with bounded retention) and browsable from Cloud Console; direct-transport hosts explicitly surface a "not reported" state. Includes Alembic migrations 0008 (progress columns on scheduled_tasks) and 0009 (planner_decision_log), bounded Host-Agent-local retention, dual-backend repository parity, and Vitest + pytest coverage. Task 6.5 (manual end-to-end device verification) remains. Co-Authored-By: Claude Opus 4.6 --- apps/cloud-api/cloud_api/app.py | 38 +++ apps/device-host-agent/host_agent/app.py | 55 ++- .../host_agent/assignment.py | 8 + apps/device-host-agent/host_agent/client.py | 30 +- apps/device-host-agent/host_agent/config.py | 21 ++ apps/device-host-agent/host_agent/lease.py | 7 +- apps/device-host-agent/host_agent/progress.py | 52 +++ .../device-host-agent/host_agent/retention.py | 63 ++++ apps/device-host-agent/host_agent/status.py | 18 + apps/device-host-agent/host_agent/web/app.py | 146 +++++++- cloud-console/src/api.ts | 11 + cloud-console/src/plannerHistory.test.ts | 51 +++ cloud-console/src/plannerHistory.ts | 36 ++ cloud-console/src/taskProgress.test.ts | 131 ++++++++ cloud-console/src/taskProgress.ts | 52 +++ cloud-console/src/types.ts | 18 + cloud-console/src/views/TasksView.vue | 176 +++++++++- docs/CLOUD_DEPLOYMENT.md | 33 ++ docs/MACOS_IPHONE_SETUP.md | 23 ++ .../.openspec.yaml | 2 + .../design.md | 129 +++++++ .../proposal.md | 34 ++ .../specs/cloud-planner-proxy/spec.md | 34 ++ .../cloud-task-progress-visibility/spec.md | 38 +++ .../host-agent-console-task-pages/spec.md | 34 ++ .../specs/host-agent-protocol/spec.md | 23 ++ .../specs/host-agent-task-progress/spec.md | 30 ++ .../tasks.md | 68 ++++ .../cloud-platform/cloud/control_config.py | 16 +- packages/cloud-platform/cloud/db_models.py | 37 +- .../cloud-platform/cloud/internal_api/api.py | 44 ++- .../cloud/internal_api/models.py | 7 + .../versions/0008_task_progress_columns.py | 38 +++ .../versions/0009_planner_decision_log.py | 49 +++ packages/cloud-platform/cloud/repository.py | 133 +++++++- packages/cloud-platform/cloud/scheduler.py | 9 +- packages/cloud-platform/cloud/schema.py | 2 +- packages/cloud-platform/cloud/sdk/api.py | 60 +++- packages/cloud-platform/cloud/sdk/models.py | 22 ++ .../cloud-platform/cloud/sql_repository.py | 108 +++++- runtime/ai_planner.py | 12 +- runtime/planner.py | 3 + runtime/task.py | 33 +- runtime/tool_calling_client.py | 35 +- storage/artifact_store.py | 6 +- storage/task_metadata.py | 16 +- storage/timeline.py | 5 + tests/test_ai_planner.py | 115 ++++++- tests/test_ai_planner_task_runner.py | 150 ++++++++- tests/test_cloud_migrations.py | 51 +++ tests/test_cloud_planner_decision_endpoint.py | 172 +++++++++- tests/test_cloud_repository_contract.py | 198 +++++++++++ tests/test_cloud_sdk_api.py | 1 + tests/test_cloud_task_progress_visibility.py | 316 ++++++++++++++++++ tests/test_host_agent_console_tasks.py | 210 ++++++++++++ tests/test_host_agent_progress_reporting.py | 235 +++++++++++++ tests/test_host_agent_task_storage.py | 290 ++++++++++++++++ tests/test_timeline.py | 21 ++ tests/test_tool_calling_client.py | 168 ++++++++-- 59 files changed, 3801 insertions(+), 122 deletions(-) create mode 100644 apps/device-host-agent/host_agent/progress.py create mode 100644 apps/device-host-agent/host_agent/retention.py create mode 100644 cloud-console/src/plannerHistory.test.ts create mode 100644 cloud-console/src/plannerHistory.ts create mode 100644 cloud-console/src/taskProgress.test.ts create mode 100644 cloud-console/src/taskProgress.ts create mode 100644 openspec/changes/task-execution-progress-visibility/.openspec.yaml create mode 100644 openspec/changes/task-execution-progress-visibility/design.md create mode 100644 openspec/changes/task-execution-progress-visibility/proposal.md create mode 100644 openspec/changes/task-execution-progress-visibility/specs/cloud-planner-proxy/spec.md create mode 100644 openspec/changes/task-execution-progress-visibility/specs/cloud-task-progress-visibility/spec.md create mode 100644 openspec/changes/task-execution-progress-visibility/specs/host-agent-console-task-pages/spec.md create mode 100644 openspec/changes/task-execution-progress-visibility/specs/host-agent-protocol/spec.md create mode 100644 openspec/changes/task-execution-progress-visibility/specs/host-agent-task-progress/spec.md create mode 100644 openspec/changes/task-execution-progress-visibility/tasks.md create mode 100644 packages/cloud-platform/cloud/migrations/versions/0008_task_progress_columns.py create mode 100644 packages/cloud-platform/cloud/migrations/versions/0009_planner_decision_log.py create mode 100644 tests/test_cloud_task_progress_visibility.py create mode 100644 tests/test_host_agent_console_tasks.py create mode 100644 tests/test_host_agent_progress_reporting.py create mode 100644 tests/test_host_agent_task_storage.py diff --git a/apps/cloud-api/cloud_api/app.py b/apps/cloud-api/cloud_api/app.py index ad9ae0f..d98e310 100644 --- a/apps/cloud-api/cloud_api/app.py +++ b/apps/cloud-api/cloud_api/app.py @@ -187,6 +187,19 @@ def create_app( ), name="cloud-lease-reaper", ), + asyncio.create_task( + _run_planner_decision_log_pruner_loop( + services, + stop_workers, + interval_seconds=( + control_config.planner_decision_log_prune_interval_seconds + ), + retention_days=( + control_config.planner_decision_log_retention_days + ), + ), + name="cloud-planner-decision-log-pruner", + ), ] app.state.worker_tasks = tuple(worker_tasks) app.state.startup_complete = True @@ -442,3 +455,28 @@ async def _wait_for_stop(stop: asyncio.Event, interval_seconds: float) -> bool: except TimeoutError: return False return True + + +async def _run_planner_decision_log_pruner_loop( + services: CloudApplicationServices, + stop: asyncio.Event, + *, + interval_seconds: float, + retention_days: int, +) -> None: + while not stop.is_set(): + correlation_token = bind_correlation_id(new_correlation_id()) + try: + services.repository.prune_planner_decision_log( + now=utc_now(), + prune_after_terminal_seconds=retention_days * 86_400, + ) + except Exception: + logger.exception( + "planner decision log prune failed", + extra={"worker": "planner_decision_log_pruner"}, + ) + finally: + reset_correlation_id(correlation_token) + if await _wait_for_stop(stop, interval_seconds): + return diff --git a/apps/device-host-agent/host_agent/app.py b/apps/device-host-agent/host_agent/app.py index d07849c..f7990d3 100644 --- a/apps/device-host-agent/host_agent/app.py +++ b/apps/device-host-agent/host_agent/app.py @@ -22,10 +22,14 @@ from host_agent.lease import ActiveAssignmentRunner from host_agent.local_account import LocalAccountStore from host_agent.policy_cache import HostPolicyCacheStore from host_agent.processor import AssignmentProcessingResult, AssignmentProcessor +from host_agent.retention import prune_task_history from host_agent.status import AgentStatusTracker from host_agent.web.app import create_console_app from host_agent.web.auth import SessionManager +from storage.artifact_store import ArtifactStore from storage.device_config import DeviceConfigStore +from storage.task_metadata import TaskMetadataStore +from storage.timeline import Timeline @dataclass @@ -178,6 +182,16 @@ def create_application( console_enrollment_client: HostAgentEnrollmentClient | None = None if resolved_config.enrollment_managed: console_enrollment_client = HostAgentEnrollmentClient(resolved_config) + metadata_store = TaskMetadataStore(db_path=resolved_config.task_progress_db_path) + timeline = Timeline(ArtifactStore(root=resolved_config.task_artifact_dir)) + executor = AssignmentExecutor( + create_execution_factories( + resolved_manager, + metadata_store=metadata_store, + timeline=timeline, + host_agent_config=resolved_config, + ) + ) console_app = create_console_app( config=resolved_config, manager=resolved_manager, @@ -190,6 +204,9 @@ def create_application( ttl_seconds=resolved_config.console_session_ttl_seconds ), enrollment_client=console_enrollment_client, + metadata_store=metadata_store, + timeline=timeline, + executor=executor, ) console_server = _EmbeddedConsoleServer( uvicorn.Config( @@ -215,19 +232,18 @@ def create_application( revision=revision ), ) - executor = AssignmentExecutor( - create_execution_factories( - resolved_manager, - host_agent_config=resolved_config, - ) - ) active_runner = ActiveAssignmentRunner(client, executor) processor = AssignmentProcessor( client, active_runner, status_tracker=status_tracker, - on_result=lambda assignment, result: _record_assignment_history( - history_store, assignment, result + on_result=lambda assignment, result: _on_assignment_finished( + history_store, + metadata_store, + timeline, + resolved_config, + assignment, + result, ), ) dependency_supervisor: DependencySupervisor | None = None @@ -245,6 +261,18 @@ def create_application( ) +def _on_assignment_finished( + history_store: ConsoleHistoryStore, + metadata_store: TaskMetadataStore, + timeline: Timeline, + config: HostAgentConfig, + assignment: AssignmentModel, + result: AssignmentProcessingResult, +) -> None: + _record_assignment_history(history_store, assignment, result) + _prune_task_history(metadata_store, timeline, config) + + def _record_assignment_history( history_store: ConsoleHistoryStore, assignment: AssignmentModel, @@ -260,6 +288,17 @@ def _record_assignment_history( ) +def _prune_task_history( + metadata_store: TaskMetadataStore, + timeline: Timeline, + config: HostAgentConfig, +) -> None: + try: + prune_task_history(metadata_store, timeline, config=config) + except Exception: + pass + + def _configured_device_manager( config_store: DeviceConfigStore, *, diff --git a/apps/device-host-agent/host_agent/assignment.py b/apps/device-host-agent/host_agent/assignment.py index df5156e..d78c1e0 100644 --- a/apps/device-host-agent/host_agent/assignment.py +++ b/apps/device-host-agent/host_agent/assignment.py @@ -8,6 +8,7 @@ from cloud.internal_api.models import AssignmentModel from core.models import Task from host_agent.execution import ExecutionFactories from host_agent.planner_context import bind_planner_execution_context +from host_agent.progress import TaskProgressHolder, TaskProgressSnapshot @dataclass(frozen=True) @@ -20,6 +21,11 @@ class AssignmentExecutionResult: class AssignmentExecutor: def __init__(self, factories: ExecutionFactories) -> None: self.factories = factories + self._progress = TaskProgressHolder() + + def latest_progress(self) -> TaskProgressSnapshot | None: + """Latest step progress reported by the currently-running assignment.""" + return self._progress.snapshot() def execute( self, @@ -27,6 +33,7 @@ class AssignmentExecutor: *, should_stop: Callable[[], bool] | None = None, ) -> AssignmentExecutionResult: + self._progress.clear() with bind_planner_execution_context(assignment): if should_stop is not None and should_stop(): return AssignmentExecutionResult( @@ -50,6 +57,7 @@ class AssignmentExecutor: ) -> AssignmentExecutionResult: task = Task(goal=assignment.goal or "", device_id=assignment.device_id) runner = self.factories.task_runner_factory() + runner.on_step_progress = self._progress.update if should_stop is None: completed = runner.run(task) else: diff --git a/apps/device-host-agent/host_agent/client.py b/apps/device-host-agent/host_agent/client.py index 2214c80..4cc37cf 100644 --- a/apps/device-host-agent/host_agent/client.py +++ b/apps/device-host-agent/host_agent/client.py @@ -16,9 +16,13 @@ from cloud.internal_api.models import ( HostEnrollmentResponse, HostTaskSubmissionResponse, LeaseRenewalResponse, + TaskProgressModel, TerminalResultResponse, ) from host_agent.config import HostAgentConfig +from host_agent.progress import TaskProgressSnapshot + +_VALID_STEP_STATUSES = frozenset({"running", "completed", "failed"}) class HostAgentAPIError(RuntimeError): @@ -194,19 +198,33 @@ class HostAgentClient: async def renew( self, assignment: AssignmentModel, + *, + progress: TaskProgressSnapshot | None = None, ) -> LeaseRenewalResponse: + payload: dict[str, Any] = { + "host_id": self.config.host_id, + "task_id": assignment.task_id, + "attempt": assignment.attempt, + "lease_id": assignment.lease_id, + } + if progress is not None: + step_status = ( + progress.step_status + if progress.step_status in _VALID_STEP_STATUSES + else "running" + ) + payload["progress"] = TaskProgressModel( + step_index=max(progress.step_index, 0), + step_status=step_status, + summary=progress.summary[:500], + ).model_dump(mode="json") response = await self._request( "POST", ( f"/internal/v1/hosts/{self.config.host_id}/assignments/" f"{assignment.task_id}/renew" ), - json={ - "host_id": self.config.host_id, - "task_id": assignment.task_id, - "attempt": assignment.attempt, - "lease_id": assignment.lease_id, - }, + json=payload, ) return LeaseRenewalResponse.model_validate(response.json()) diff --git a/apps/device-host-agent/host_agent/config.py b/apps/device-host-agent/host_agent/config.py index 6ffde69..8868668 100644 --- a/apps/device-host-agent/host_agent/config.py +++ b/apps/device-host-agent/host_agent/config.py @@ -43,6 +43,10 @@ class HostAgentConfig: runtime_host: str = "127.0.0.1" runtime_port: int = 8000 dependency_restart_max_attempts: int = 5 + task_progress_db_path: Path = Path("host_agent_data/task_progress.sqlite3") + task_artifact_dir: Path = Path("host_agent_data/history") + task_retention_max_count: int = 50 + task_retention_max_age_days: int = 7 def load_host_agent_config( @@ -135,6 +139,23 @@ def load_host_agent_config( dependency_restart_max_attempts=_positive_int( values, "HOST_AGENT_DEPENDENCY_RESTART_MAX_ATTEMPTS", 5 ), + task_progress_db_path=Path( + values.get( + "HOST_AGENT_TASK_PROGRESS_DB_PATH", + "host_agent_data/task_progress.sqlite3", + ).strip() + ), + task_artifact_dir=Path( + values.get( + "HOST_AGENT_TASK_ARTIFACT_DIR", "host_agent_data/history" + ).strip() + ), + task_retention_max_count=_positive_int( + values, "HOST_AGENT_TASK_RETENTION_MAX_COUNT", 50 + ), + task_retention_max_age_days=_positive_int( + values, "HOST_AGENT_TASK_RETENTION_MAX_AGE_DAYS", 7 + ), ) if config.max_retry_backoff_seconds < config.retry_backoff_seconds: raise HostAgentConfigurationError( diff --git a/apps/device-host-agent/host_agent/lease.py b/apps/device-host-agent/host_agent/lease.py index 8dc4e8b..29e9e1c 100644 --- a/apps/device-host-agent/host_agent/lease.py +++ b/apps/device-host-agent/host_agent/lease.py @@ -11,6 +11,7 @@ import httpx from cloud.internal_api.models import AssignmentModel from host_agent.assignment import AssignmentExecutionResult from host_agent.client import HostAgentAPIError, HostAgentClient, StaleLeaseError +from host_agent.progress import TaskProgressSnapshot class InterruptibleAssignmentExecutor(Protocol): @@ -21,6 +22,8 @@ class InterruptibleAssignmentExecutor(Protocol): should_stop: Callable[[], bool] | None = None, ) -> AssignmentExecutionResult: ... + def latest_progress(self) -> TaskProgressSnapshot | None: ... + class LeaseGuard: def __init__(self) -> None: @@ -92,7 +95,9 @@ class ActiveAssignmentRunner: if done: return try: - response = await self.client.renew(assignment) + response = await self.client.renew( + assignment, progress=self.executor.latest_progress() + ) except StaleLeaseError: guard.mark_lost("lease rejected by control plane") return diff --git a/apps/device-host-agent/host_agent/progress.py b/apps/device-host-agent/host_agent/progress.py new file mode 100644 index 0000000..27559cb --- /dev/null +++ b/apps/device-host-agent/host_agent/progress.py @@ -0,0 +1,52 @@ +from __future__ import annotations + +import threading +from dataclasses import dataclass +from datetime import UTC, datetime +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from collections.abc import Callable + +_MAX_SUMMARY_LENGTH = 500 + + +@dataclass(frozen=True) +class TaskProgressSnapshot: + step_index: int + step_status: str + summary: str + updated_at: datetime + + +class TaskProgressHolder: + """Thread-safe latest-step-progress holder for one in-flight assignment. + + Written by the execution thread (via ``update``, wired as the + ``TaskRunner.on_step_progress`` callback) and read by the asyncio + lease-renewal loop (via ``snapshot``) just before each renewal call. + """ + + def __init__(self, *, now: Callable[[], datetime] | None = None) -> None: + self._now = now or (lambda: datetime.now(UTC)) + self._lock = threading.Lock() + self._snapshot: TaskProgressSnapshot | None = None + + def update(self, step_index: int, step_status: str, summary: str) -> None: + if len(summary) > _MAX_SUMMARY_LENGTH: + summary = summary[:_MAX_SUMMARY_LENGTH] + with self._lock: + self._snapshot = TaskProgressSnapshot( + step_index=step_index, + step_status=step_status, + summary=summary, + updated_at=self._now(), + ) + + def clear(self) -> None: + with self._lock: + self._snapshot = None + + def snapshot(self) -> TaskProgressSnapshot | None: + with self._lock: + return self._snapshot diff --git a/apps/device-host-agent/host_agent/retention.py b/apps/device-host-agent/host_agent/retention.py new file mode 100644 index 0000000..a431c6f --- /dev/null +++ b/apps/device-host-agent/host_agent/retention.py @@ -0,0 +1,63 @@ +from __future__ import annotations + +import logging +from datetime import UTC, datetime, timedelta + +from host_agent.config import HostAgentConfig +from storage.task_metadata import TaskMetadataStore +from storage.timeline import Timeline + +logger = logging.getLogger(__name__) + + +def prune_task_history( + metadata_store: TaskMetadataStore, + timeline: Timeline, + *, + config: HostAgentConfig, + now: datetime | None = None, +) -> int: + """Delete tasks beyond the configured retention window/count. + + Computes two candidate retained sets -- the last ``max_count`` tasks + and tasks younger than ``max_age_days`` -- and keeps whichever set + is smaller (i.e. the more restrictive bound always wins). + + Returns the number of tasks pruned. + """ + reference = now or datetime.now(UTC) + tasks = metadata_store.list_tasks() + + keep_by_count = {task["id"] for task in tasks[: config.task_retention_max_count]} + cutoff = reference - timedelta(days=config.task_retention_max_age_days) + keep_by_age = { + task["id"] for task in tasks if _parse_timestamp(task["created_at"]) >= cutoff + } + keep_ids = keep_by_count if len(keep_by_count) <= len(keep_by_age) else keep_by_age + + pruned = 0 + for task in tasks: + if task["id"] not in keep_ids: + _safe_delete(timeline, metadata_store, task["id"]) + pruned += 1 + return pruned + + +def _safe_delete( + timeline: Timeline, metadata_store: TaskMetadataStore, task_id: str +) -> None: + try: + timeline.delete_task(task_id) + except Exception: + logger.debug("timeline delete failed for task %s", task_id, exc_info=True) + try: + metadata_store.delete_task(task_id) + except Exception: + logger.debug("metadata delete failed for task %s", task_id, exc_info=True) + + +def _parse_timestamp(value: str) -> datetime: + parsed = datetime.fromisoformat(value) + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=UTC) + return parsed diff --git a/apps/device-host-agent/host_agent/status.py b/apps/device-host-agent/host_agent/status.py index 87bcc7b..cacf955 100644 --- a/apps/device-host-agent/host_agent/status.py +++ b/apps/device-host-agent/host_agent/status.py @@ -6,6 +6,7 @@ from datetime import UTC, datetime from typing import TYPE_CHECKING, Any from cloud.internal_api.models import AssignmentModel +from host_agent.progress import TaskProgressSnapshot if TYPE_CHECKING: from collections.abc import Callable @@ -34,6 +35,7 @@ class AgentStatusTracker: self._current_assignment: _CurrentAssignment | None = None self._last_heartbeat: _LastHeartbeat | None = None self._host_policy: dict[str, Any] | None = None + self._latest_progress: TaskProgressSnapshot | None = None def mark_assignment_started(self, assignment: AssignmentModel) -> None: with self._lock: @@ -48,6 +50,11 @@ class AgentStatusTracker: def mark_assignment_finished(self) -> None: with self._lock: self._current_assignment = None + self._latest_progress = None + + def set_latest_progress(self, snapshot: TaskProgressSnapshot | None) -> None: + with self._lock: + self._latest_progress = snapshot def mark_heartbeat(self, *, ok: bool, device_count: int) -> None: with self._lock: @@ -74,6 +81,7 @@ class AgentStatusTracker: with self._lock: current_assignment = self._current_assignment last_heartbeat = self._last_heartbeat + progress = self._latest_progress return { "current_assignment": ( { @@ -96,4 +104,14 @@ class AgentStatusTracker: else None ), "host_policy": self._host_policy.copy() if self._host_policy else None, + "progress": ( + { + "step_index": progress.step_index, + "step_status": progress.step_status, + "summary": progress.summary, + "updated_at": progress.updated_at.isoformat(), + } + if progress is not None + else None + ), } diff --git a/apps/device-host-agent/host_agent/web/app.py b/apps/device-host-agent/host_agent/web/app.py index 2749936..05ebded 100644 --- a/apps/device-host-agent/host_agent/web/app.py +++ b/apps/device-host-agent/host_agent/web/app.py @@ -1,14 +1,17 @@ from __future__ import annotations import asyncio +import base64 import json from html import escape as _escape +from pathlib import Path from typing import Any from fastapi import Depends, FastAPI, HTTPException, Request from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse, Response from device.manager import DeviceManager +from host_agent.assignment import AssignmentExecutor from host_agent.client import HostAgentEnrollmentClient from host_agent.config import HostAgentConfig from host_agent.devices import register_local_device, unregister_local_device @@ -23,6 +26,8 @@ from host_agent.web.auth import ( change_password, ) from storage.device_config import DeviceConfigStore +from storage.task_metadata import TaskMetadataStore +from storage.timeline import Timeline SESSION_COOKIE_NAME = "host_console_session" CSRF_HEADER_NAME = "X-CSRF-Token" @@ -58,6 +63,7 @@ def _chrome(title: str, body_html: str, *, session: SessionState | None) -> str: