diff --git a/apps/cloud-api/cloud_api/app.py b/apps/cloud-api/cloud_api/app.py index 6b2f74a..c76a38a 100644 --- a/apps/cloud-api/cloud_api/app.py +++ b/apps/cloud-api/cloud_api/app.py @@ -8,6 +8,7 @@ from dataclasses import dataclass from typing import Any from fastapi import FastAPI, status +from fastapi import Request from fastapi.responses import JSONResponse from cloud.auth import create_auth_provider @@ -20,6 +21,13 @@ from cloud.control_config import ( from cloud.database import CloudDatabase from cloud.internal_api.api import create_internal_router from cloud.plugins import PluginRegistry +from cloud.observability import ( + CORRELATION_HEADER, + bind_correlation_id, + new_correlation_id, + normalize_correlation_id, + reset_correlation_id, +) from cloud.pool import DevicePool from cloud.scheduler import TaskScheduler from cloud.schema import require_current_schema @@ -136,6 +144,28 @@ def create_app( app = FastAPI(title="Device Cloud API", lifespan=lifespan) + @app.middleware("http") + async def correlation_logging(request: Request, call_next): + correlation_id = normalize_correlation_id( + request.headers.get(CORRELATION_HEADER) + ) + correlation_token = bind_correlation_id(correlation_id) + try: + response = await call_next(request) + logger.info( + "cloud request completed", + extra={ + "correlation_id": correlation_id, + "method": request.method, + "path": request.url.path, + "status_code": response.status_code, + }, + ) + response.headers[CORRELATION_HEADER] = correlation_id + return response + finally: + reset_correlation_id(correlation_token) + @app.get("/health/live") def health_live() -> dict[str, str]: return {"status": "live"} @@ -198,6 +228,7 @@ async def _run_scheduler_loop( interval_seconds: float, ) -> None: while not stop.is_set(): + correlation_token = bind_correlation_id(new_correlation_id()) try: services.scheduler.assign() except Exception: @@ -205,6 +236,8 @@ async def _run_scheduler_loop( "cloud lifecycle iteration failed", extra={"worker": "scheduler"}, ) + finally: + reset_correlation_id(correlation_token) if await _wait_for_stop(stop, interval_seconds): return @@ -217,6 +250,7 @@ async def _run_lease_reaper_loop( max_attempts: int, ) -> None: while not stop.is_set(): + correlation_token = bind_correlation_id(new_correlation_id()) try: services.repository.reap_expired_leases( now=utc_now(), @@ -227,6 +261,8 @@ async def _run_lease_reaper_loop( "cloud lifecycle iteration failed", extra={"worker": "lease_reaper"}, ) + finally: + reset_correlation_id(correlation_token) if await _wait_for_stop(stop, interval_seconds): return diff --git a/apps/cloud-api/tests/test_app.py b/apps/cloud-api/tests/test_app.py index 479f324..1c5db6c 100644 --- a/apps/cloud-api/tests/test_app.py +++ b/apps/cloud-api/tests/test_app.py @@ -255,6 +255,43 @@ def test_readiness_reports_stopped_worker() -> None: assert response.json()["checks"]["workers"] is False +def test_request_correlation_id_is_propagated_without_sensitive_headers( + monkeypatch, + tmp_path, +) -> None: + request_logs: list[dict[str, object]] = [] + + def record_info(_message: str, *, extra: dict[str, object]) -> None: + request_logs.append(extra) + + monkeypatch.setattr(app_module.logger, "info", record_info) + app = create_app( + config=CloudControlConfig( + database_url=f"sqlite:///{(tmp_path / 'correlation.sqlite3').as_posix()}" + ) + ) + + with TestClient(app) as client: + response = client.get( + "/health/live", + headers={ + "X-Correlation-ID": "request-123", + "Authorization": "Bearer never-log-this", + }, + ) + + assert response.headers["X-Correlation-ID"] == "request-123" + assert request_logs == [ + { + "correlation_id": "request-123", + "method": "GET", + "path": "/health/live", + "status_code": 200, + } + ] + assert "never-log-this" not in repr(request_logs) + + def test_production_app_rejects_missing_credentials() -> None: with pytest.raises(CloudConfigurationError, match="credential"): create_app( diff --git a/openspec/changes/cloud-control-plane-integration/tasks.md b/openspec/changes/cloud-control-plane-integration/tasks.md index 02158c8..9194c38 100644 --- a/openspec/changes/cloud-control-plane-integration/tasks.md +++ b/openspec/changes/cloud-control-plane-integration/tasks.md @@ -47,7 +47,7 @@ - [x] 6.2 Implement FastAPI lifespan startup/shutdown for configuration validation, database checks, scheduler loop, and lease-reaper loop. - [x] 6.3 Ensure lifecycle iteration failures are logged and retried without terminating later iterations. - [x] 6.4 Add `/health/live` and `/health/ready` with separate process, database/schema, and worker-state semantics. -- [ ] 6.5 Add structured correlation-aware logging for requests and task lifecycle events with sensitive payload redaction. +- [x] 6.5 Add structured correlation-aware logging for requests and task lifecycle events with sensitive payload redaction. - [ ] 6.6 Add app-level tests for startup failures, readiness transitions, graceful shutdown, persisted queue recovery, and expired-lease recovery. ## 7. Device Host Agent Execution Loop diff --git a/packages/cloud-platform/cloud/observability.py b/packages/cloud-platform/cloud/observability.py new file mode 100644 index 0000000..55795db --- /dev/null +++ b/packages/cloud-platform/cloud/observability.py @@ -0,0 +1,63 @@ +from __future__ import annotations + +from contextvars import ContextVar, Token +from typing import Any +from uuid import uuid4 + + +CORRELATION_HEADER = "X-Correlation-ID" +_correlation_id: ContextVar[str | None] = ContextVar( + "cloud_correlation_id", + default=None, +) +_SENSITIVE_KEYS = { + "authorization", + "token", + "bearer_token", + "password", + "screenshot", + "ui_tree", + "typed_text", + "text_input", +} + + +def new_correlation_id() -> str: + return uuid4().hex + + +def normalize_correlation_id(value: str | None) -> str: + if value is None: + return new_correlation_id() + normalized = value.strip() + if not normalized or len(normalized) > 128: + return new_correlation_id() + return normalized + + +def bind_correlation_id(correlation_id: str) -> Token[str | None]: + return _correlation_id.set(correlation_id) + + +def reset_correlation_id(token: Token[str | None]) -> None: + _correlation_id.reset(token) + + +def current_correlation_id() -> str: + correlation_id = _correlation_id.get() + return correlation_id or new_correlation_id() + + +def redact_sensitive_fields(value: Any) -> Any: + if isinstance(value, dict): + return { + key: "[REDACTED]" + if key.lower() in _SENSITIVE_KEYS + else redact_sensitive_fields(item) + for key, item in value.items() + } + if isinstance(value, list): + return [redact_sensitive_fields(item) for item in value] + if isinstance(value, tuple): + return tuple(redact_sensitive_fields(item) for item in value) + return value diff --git a/packages/cloud-platform/cloud/sql_repository.py b/packages/cloud-platform/cloud/sql_repository.py index 5c7218c..1a0820a 100644 --- a/packages/cloud-platform/cloud/sql_repository.py +++ b/packages/cloud-platform/cloud/sql_repository.py @@ -1,6 +1,7 @@ from __future__ import annotations import json +import logging from dataclasses import asdict from datetime import datetime from typing import Any @@ -16,9 +17,13 @@ from cloud.db_models import ( ScheduledTaskRow, TaskAttemptRow, ) +from cloud.observability import current_correlation_id from core.models import utc_now +logger = logging.getLogger(__name__) + + class SQLAlchemyCloudRepository: """SQLAlchemy adapter preserving the existing CloudStore CRUD surface.""" @@ -288,6 +293,7 @@ class SQLAlchemyCloudRepository: ) ) session.flush() + _log_task_lifecycle("assigned", task) return _leased_assignment_from_row(task) def claim_assignment( @@ -331,6 +337,7 @@ class SQLAlchemyCloudRepository: task.updated_at = _iso(now) attempt.status = "dispatched" session.flush() + _log_task_lifecycle("claimed", task) return _leased_assignment_from_row(task) def renew_lease( @@ -381,6 +388,7 @@ class SQLAlchemyCloudRepository: task.lease_expires_at = renewed_until task.updated_at = _iso(now) attempt_row.lease_expires_at = renewed_until + _log_task_lifecycle("renewed", task) return "renewed" def record_task_result( @@ -449,6 +457,7 @@ class SQLAlchemyCloudRepository: attempt_row.completed_at = completed_at_iso attempt_row.failure_reason = failure_reason attempt_row.result_json = result_json + _log_task_lifecycle("completed" if status == "done" else "failed", task) return "recorded" def reap_expired_leases( @@ -501,6 +510,10 @@ class SQLAlchemyCloudRepository: task.failure_reason = ( f"lease expired after {task.attempt_count} attempts" ) + _log_task_lifecycle( + "retried" if task.status == "queued" else "failed", + task, + ) reaped_task_ids.append(task.id) return reaped_task_ids @@ -641,6 +654,21 @@ def _parse_json_object(value: str | None) -> dict[str, Any] | None: return parsed if isinstance(parsed, dict) else None +def _log_task_lifecycle(event: str, task: ScheduledTaskRow) -> None: + logger.info( + "cloud task lifecycle", + extra={ + "event": event, + "correlation_id": current_correlation_id(), + "task_id": task.id, + "host_id": task.assigned_host_id, + "device_id": task.assigned_device_id, + "attempt": task.attempt_count, + "lease_id": task.lease_id, + }, + ) + + def _plugin_from_row(row: PluginRow) -> tuple[Any, bool]: from cloud.plugins import PluginManifest diff --git a/tests/test_cloud_observability.py b/tests/test_cloud_observability.py new file mode 100644 index 0000000..efdef67 --- /dev/null +++ b/tests/test_cloud_observability.py @@ -0,0 +1,25 @@ +from __future__ import annotations + +from cloud.observability import redact_sensitive_fields + + +def test_sensitive_payload_fields_are_redacted_recursively() -> None: + payload = { + "authorization": "Bearer secret", + "task_id": "task-a", + "result": { + "screenshot": "base64-data", + "ui_tree": {"typed_text": "private input"}, + }, + } + + redacted = redact_sensitive_fields(payload) + + assert redacted == { + "authorization": "[REDACTED]", + "task_id": "task-a", + "result": { + "screenshot": "[REDACTED]", + "ui_tree": "[REDACTED]", + }, + } diff --git a/tests/test_cloud_repository_contract.py b/tests/test_cloud_repository_contract.py index 3198d1c..abda2cd 100644 --- a/tests/test_cloud_repository_contract.py +++ b/tests/test_cloud_repository_contract.py @@ -1145,3 +1145,69 @@ def test_unexpired_lease_is_not_reaped(database_url: str) -> None: assert device_id in database.repository.list_reserved_device_ids(now=now) finally: database.close() + + +def test_task_lifecycle_logs_structured_identifiers( + database_url: str, monkeypatch +) -> None: + import cloud.sql_repository as repository_module + + database = CloudDatabase(database_url) + now = datetime(2026, 7, 12, 20, 0, tzinfo=UTC) + host_id = _unique_id("log-host") + device_id = _unique_id("log-device") + task_id = _unique_id("log-task") + events: list[dict[str, object]] = [] + + def record_info(_message: str, *, extra: dict[str, object]) -> None: + events.append(extra) + + monkeypatch.setattr(repository_module.logger, "info", record_info) + try: + database.repository.upsert_host(host_id, address=None, last_seen_at=now) + database.repository.replace_host_devices( + host_id, + [_device(device_id, host_id)], + ) + database.repository.enqueue_task( + ScheduledTask( + id=task_id, + goal="sensitive typed text", + workflow_definition_id=None, + constraints=TaskConstraints(), + created_at=now, + ) + ) + database.repository.assign_task( + task_id=task_id, + host_id=host_id, + device_id=device_id, + lease_id="log-lease", + lease_expires_at=now + timedelta(minutes=1), + now=now, + ) + database.repository.claim_assignment(host_id=host_id, now=now) + database.repository.record_task_result( + task_id=task_id, + attempt=1, + lease_id="log-lease", + host_id=host_id, + status="done", + failure_reason=None, + terminal_result={"screenshot": "secret-image"}, + completed_at=now + timedelta(seconds=1), + ) + + assert [event["event"] for event in events] == [ + "assigned", + "claimed", + "completed", + ] + assert all(event["task_id"] == task_id for event in events) + assert all(event["host_id"] == host_id for event in events) + assert all(event["device_id"] == device_id for event in events) + assert all(event["correlation_id"] for event in events) + assert "sensitive typed text" not in repr(events) + assert "secret-image" not in repr(events) + finally: + database.close()