feat(cloud): add targeted task governance foundation

packages/cloud-platform/cloud/auth.py

@@ -16,6 +16,8 @@ POOL_READ_SCOPE = "pool:read"
 PLUGINS_READ_SCOPE = "plugins:read"
 PLUGINS_ADMIN_SCOPE = "plugins:admin"
 USERS_ADMIN_SCOPE = "users:admin"
+GOVERNANCE_READ_SCOPE = "governance:read"
+GOVERNANCE_ADMIN_SCOPE = "governance:admin"
 
 
 @dataclass(frozen=True)

packages/cloud-platform/cloud/db_models.py

@@ -215,3 +215,29 @@ class AuthAuditRow(Base):
     outcome: Mapped[str] = mapped_column(String, nullable=False)
     correlation_id: Mapped[str | None] = mapped_column(String, nullable=True)
     metadata_json: Mapped[str] = mapped_column(Text, nullable=False, default="{}")
+
+
+class UserSubmissionPolicyRow(Base):
+    __tablename__ = "cloud_user_submission_policies"
+
+    user_id: Mapped[str] = mapped_column(
+        ForeignKey("cloud_users.id", ondelete="CASCADE"), primary_key=True
+    )
+    revision: Mapped[int] = mapped_column(Integer, nullable=False)
+    submission_enabled: Mapped[int] = mapped_column(Integer, nullable=False)
+    allowed_host_ids_json: Mapped[str | None] = mapped_column(Text, nullable=True)
+    allowed_device_targets_json: Mapped[str | None] = mapped_column(Text, nullable=True)
+    updated_at: Mapped[str] = mapped_column(String, nullable=False)
+
+
+class HostGovernancePolicyRow(Base):
+    __tablename__ = "cloud_host_governance_policies"
+
+    host_id: Mapped[str] = mapped_column(
+        ForeignKey("host_registrations.host_id", ondelete="CASCADE"), primary_key=True
+    )
+    revision: Mapped[int] = mapped_column(Integer, nullable=False)
+    self_submission_enabled: Mapped[int] = mapped_column(Integer, nullable=False)
+    max_active_tasks: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    daily_token_budget: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    updated_at: Mapped[str] = mapped_column(String, nullable=False)

packages/cloud-platform/cloud/governance.py

@@ -0,0 +1,61 @@
+"""Durable Cloud-side submission and Host governance policies."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime
+
+
+class TaskSubmissionPolicyError(PermissionError):
+    """Raised when a human user's policy disallows a task submission."""
+
+
+@dataclass(frozen=True)
+class UserSubmissionPolicy:
+    user_id: str
+    revision: int
+    submission_enabled: bool
+    allowed_host_ids: tuple[str, ...] | None
+    allowed_device_targets: tuple[tuple[str, str], ...] | None
+    updated_at: datetime
+
+
+@dataclass(frozen=True)
+class HostGovernancePolicy:
+    host_id: str
+    revision: int
+    self_submission_enabled: bool
+    max_active_tasks: int | None
+    daily_token_budget: int | None
+    updated_at: datetime
+
+
+def enforce_user_submission_policy(
+    policy: UserSubmissionPolicy | None,
+    *,
+    target_host_id: str | None,
+    target_device_id: str | None,
+) -> None:
+    if policy is None:
+        return
+    if not policy.submission_enabled:
+        raise TaskSubmissionPolicyError("task submission is disabled for this user")
+    restricted = (
+        policy.allowed_host_ids is not None
+        or policy.allowed_device_targets is not None
+    )
+    if not restricted:
+        return
+    if target_host_id is None:
+        raise TaskSubmissionPolicyError("an explicit permitted target is required")
+    if (
+        policy.allowed_host_ids is not None
+        and target_host_id not in policy.allowed_host_ids
+    ):
+        raise TaskSubmissionPolicyError("target host is not permitted")
+    if policy.allowed_device_targets is not None:
+        if target_device_id is None or (
+            target_host_id,
+            target_device_id,
+        ) not in policy.allowed_device_targets:
+            raise TaskSubmissionPolicyError("target device is not permitted")

packages/cloud-platform/cloud/internal_api/api.py

@@ -25,8 +25,11 @@ from cloud.internal_api.models import (
     DeviceEnrollmentResponse,
     HeartbeatRequest,
     HeartbeatResponse,
+    HostGovernancePolicyModel,
     HostEnrollmentRequest,
     HostEnrollmentResponse,
+    HostTaskSubmissionRequest,
+    HostTaskSubmissionResponse,
     LeaseRenewalRequest,
     LeaseRenewalResponse,
     PlannerDecisionError,
@@ -47,6 +50,7 @@ from runtime.tool_specs import ToolSpec
 
 if TYPE_CHECKING:
     from cloud.pool import DevicePool
+    from cloud.scheduler import TaskScheduler
 
 logger = logging.getLogger(__name__)
 
@@ -60,6 +64,7 @@ def create_internal_router(
     lease_duration_seconds: float = 60.0,
     sleep: Callable[[float], Awaitable[None]] = asyncio.sleep,
     planner_client_factory: Callable[[], ToolCallingClient] | None = None,
+    scheduler: TaskScheduler | None = None,
 ) -> APIRouter:
     if claim_poll_interval_seconds <= 0:
         raise ValueError("claim_poll_interval_seconds must be greater than zero")
@@ -166,12 +171,71 @@ def create_internal_router(
             address=payload.address,
             allow_device_takeover=allow_device_takeover,
         )
+        policy = pool.store.get_host_governance_policy(host_id)
+        policy_revision = policy.revision if policy is not None else 0
         return HeartbeatResponse(
             host_id=host_id,
             accepted_devices=len(devices),
             received_at=utc_now(),
+            policy_revision=policy_revision,
+            policy=(
+                HostGovernancePolicyModel(
+                    revision=policy.revision,
+                    self_submission_enabled=policy.self_submission_enabled,
+                    max_active_tasks=policy.max_active_tasks,
+                    daily_token_budget=policy.daily_token_budget,
+                )
+                if policy is not None and payload.policy_revision != policy.revision
+                else None
+            ),
         )
 
+    @router.post(
+        "/hosts/{host_id}/tasks",
+        response_model=HostTaskSubmissionResponse,
+        status_code=status.HTTP_201_CREATED,
+    )
+    def submit_host_task(
+        host_id: str,
+        payload: HostTaskSubmissionRequest,
+        request: Request,
+    ) -> HostTaskSubmissionResponse:
+        authorize_host(request, host_id)
+        if payload.host_id != host_id:
+            raise HTTPException(
+                status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
+                detail="task host_id must match the request path",
+            )
+        if scheduler is None:
+            raise HTTPException(
+                status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+                detail="task submission is unavailable",
+            )
+        policy = pool.store.get_host_governance_policy(host_id)
+        if policy is not None and not policy.self_submission_enabled:
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail="Host self-submission is disabled",
+            )
+        if payload.device_id is not None and not any(
+            device.host_id == host_id and device.device_id == payload.device_id
+            for device in pool.list_devices()
+        ):
+            raise HTTPException(
+                status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
+                detail="target device is not owned by this Host",
+            )
+        from cloud.scheduler import TaskConstraints
+
+        task_id = scheduler.submit(
+            goal=payload.goal,
+            constraints=TaskConstraints(
+                target_host_id=host_id,
+                target_device_id=payload.device_id,
+            ),
+        )
+        return HostTaskSubmissionResponse(task_id=task_id)
+
     @router.post(
         "/hosts/{host_id}/assignments/claim",
         response_model=ClaimResponse,
@@ -285,6 +349,7 @@ def create_internal_router(
     @router.post(
         "/hosts/{host_id}/planner/decide",
         response_model=PlannerDecisionResponse,
+        response_model_exclude_none=True,
         responses={
             status.HTTP_502_BAD_GATEWAY: {"model": PlannerDecisionError},
         },
@@ -354,6 +419,9 @@ def create_internal_router(
         return PlannerDecisionResponse(
             tool_name=decision.tool_name,
             arguments=dict(decision.arguments),
+            input_tokens=(decision.usage.input_tokens if decision.usage else None),
+            output_tokens=(decision.usage.output_tokens if decision.usage else None),
+            total_tokens=(decision.usage.total_tokens if decision.usage else None),
         )
 
     return router

packages/cloud-platform/cloud/internal_api/models.py

@@ -38,12 +38,22 @@ class HeartbeatRequest(BaseModel):
     host_id: str = Field(min_length=1)
     address: str | None = None
     devices: list[DeviceSnapshotModel] = Field(default_factory=list)
+    policy_revision: int = Field(default=0, ge=0)
+
+
+class HostGovernancePolicyModel(BaseModel):
+    revision: int = Field(ge=1)
+    self_submission_enabled: bool
+    max_active_tasks: int | None = None
+    daily_token_budget: int | None = None
 
 
 class HeartbeatResponse(BaseModel):
     host_id: str
     accepted_devices: int
     received_at: datetime
+    policy_revision: int = Field(default=0, ge=0)
+    policy: HostGovernancePolicyModel | None = None
 
 
 class ClaimRequest(BaseModel):
@@ -93,6 +103,16 @@ class TerminalResultResponse(BaseModel):
     status: Literal["recorded", "already_recorded"]
 
 
+class HostTaskSubmissionRequest(BaseModel):
+    host_id: str = Field(min_length=1)
+    goal: str = Field(min_length=1)
+    device_id: str | None = Field(default=None, min_length=1)
+
+
+class HostTaskSubmissionResponse(BaseModel):
+    task_id: str
+
+
 class StaleLeaseConflict(BaseModel):
     code: Literal["stale_lease"] = "stale_lease"
     detail: str
@@ -116,6 +136,9 @@ class PlannerDecisionRequest(BaseModel):
 class PlannerDecisionResponse(BaseModel):
     tool_name: str
     arguments: dict[str, Any] = Field(default_factory=dict)
+    input_tokens: int | None = Field(default=None, ge=0)
+    output_tokens: int | None = Field(default=None, ge=0)
+    total_tokens: int | None = Field(default=None, ge=0)
 
 
 class PlannerDecisionError(BaseModel):

packages/cloud-platform/cloud/migrations/versions/0004_cloud_governance.py

@@ -0,0 +1,56 @@
+"""Add durable Cloud governance policies."""
+
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "0004_cloud_governance"
+down_revision = "0003_cloud_user_authentication"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    inspector = sa.inspect(op.get_bind())
+    tables = set(inspector.get_table_names())
+    if "cloud_user_submission_policies" not in tables:
+        op.create_table(
+            "cloud_user_submission_policies",
+            sa.Column(
+                "user_id",
+                sa.String(),
+                sa.ForeignKey("cloud_users.id", ondelete="CASCADE"),
+                primary_key=True,
+            ),
+            sa.Column("revision", sa.Integer(), nullable=False),
+            sa.Column("submission_enabled", sa.Integer(), nullable=False),
+            sa.Column("allowed_host_ids_json", sa.Text(), nullable=True),
+            sa.Column("allowed_device_targets_json", sa.Text(), nullable=True),
+            sa.Column("updated_at", sa.String(), nullable=False),
+        )
+    if "cloud_host_governance_policies" not in tables:
+        op.create_table(
+            "cloud_host_governance_policies",
+            sa.Column(
+                "host_id",
+                sa.String(),
+                sa.ForeignKey("host_registrations.host_id", ondelete="CASCADE"),
+                primary_key=True,
+            ),
+            sa.Column("revision", sa.Integer(), nullable=False),
+            sa.Column("self_submission_enabled", sa.Integer(), nullable=False),
+            sa.Column("max_active_tasks", sa.Integer(), nullable=True),
+            sa.Column("daily_token_budget", sa.Integer(), nullable=True),
+            sa.Column("updated_at", sa.String(), nullable=False),
+        )
+
+
+def downgrade() -> None:
+    inspector = sa.inspect(op.get_bind())
+    tables = set(inspector.get_table_names())
+    if "cloud_host_governance_policies" in tables:
+        op.drop_table("cloud_host_governance_policies")
+    if "cloud_user_submission_policies" in tables:
+        op.drop_table("cloud_user_submission_policies")

packages/cloud-platform/cloud/repository.py

@@ -15,6 +15,7 @@ if TYPE_CHECKING:
         UserAccount,
         UserSession,
     )
+    from cloud.governance import HostGovernancePolicy, UserSubmissionPolicy
 
 
 AttemptStatus = Literal["assigned", "dispatched", "done", "failed", "expired"]
@@ -283,6 +284,30 @@ class CloudRepository(Protocol):
 
     def cleanup_auth_state(self, *, now: datetime, limit: int) -> int: ...
 
+    def get_user_submission_policy(self, user_id: str) -> UserSubmissionPolicy | None: ...
+
+    def upsert_user_submission_policy(
+        self,
+        *,
+        user_id: str,
+        submission_enabled: bool,
+        allowed_host_ids: tuple[str, ...] | None,
+        allowed_device_targets: tuple[tuple[str, str], ...] | None,
+        updated_at: datetime,
+    ) -> UserSubmissionPolicy: ...
+
+    def get_host_governance_policy(self, host_id: str) -> HostGovernancePolicy | None: ...
+
+    def upsert_host_governance_policy(
+        self,
+        *,
+        host_id: str,
+        self_submission_enabled: bool,
+        max_active_tasks: int | None,
+        daily_token_budget: int | None,
+        updated_at: datetime,
+    ) -> HostGovernancePolicy: ...
+
     def list_reserved_device_ids(self, *, now: datetime) -> set[str]: ...
 
     def assign_task(

packages/cloud-platform/cloud/scheduler.py

@@ -31,6 +31,8 @@ class TaskConstraints:
 
     driver_type: str | None = None
     capability_tags: list[str] = field(default_factory=list)
+    target_host_id: str | None = None
+    target_device_id: str | None = None
 
 
 @dataclass
@@ -195,6 +197,10 @@ class TaskScheduler:
 
 
 def _matches(device: "PooledDevice", constraints: TaskConstraints) -> bool:
+    if constraints.target_host_id and device.host_id != constraints.target_host_id:
+        return False
+    if constraints.target_device_id and device.device_id != constraints.target_device_id:
+        return False
     if constraints.driver_type and device.driver_type != constraints.driver_type:
         return False
     if constraints.capability_tags:

packages/cloud-platform/cloud/schema.py

@@ -9,7 +9,7 @@ from alembic.runtime.migration import MigrationContext
 from cloud.database import create_database_engine, normalize_database_url
 
 
-HEAD_REVISION = "0003_cloud_user_authentication"
+HEAD_REVISION = "0004_cloud_governance"
 
 
 class SchemaVersionError(RuntimeError):

packages/cloud-platform/cloud/sdk/api.py

@@ -22,6 +22,7 @@ from cloud.auth import (
     NullAuthProvider,
     Principal,
 )
+from cloud.governance import TaskSubmissionPolicyError, enforce_user_submission_policy
 from cloud.sdk.models import (
     DeviceResponse,
     ErrorResponse,
@@ -94,8 +95,21 @@ def create_cloud_router(
         payload: TaskSubmissionRequest,
         request: Request,
     ) -> TaskSubmissionResponse:
-        _authorize(request, TASKS_SUBMIT_SCOPE)
-        task_constraints = _build_constraints(payload.constraints)
+        principal = _authorize(request, TASKS_SUBMIT_SCOPE)
+        try:
+            task_constraints = _build_constraints(payload.constraints)
+            _validate_task_target(pool, task_constraints)
+            _enforce_user_policy(principal, scheduler.store, task_constraints)
+        except TaskSubmissionPolicyError as exc:
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail=str(exc),
+            ) from exc
+        except ValueError as exc:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail=str(exc),
+            ) from exc
         try:
             task_id = scheduler.submit(
                 goal=payload.goal,
@@ -128,6 +142,8 @@ def create_cloud_router(
             attempt_count=task.attempt_count,
             lease_expires_at=task.lease_expires_at,
             failure_reason=task.failure_reason,
+            target_host_id=task.constraints.target_host_id,
+            target_device_id=task.constraints.target_device_id,
         )
 
     @router.get("/tasks", response_model=TaskListResponse)
@@ -158,6 +174,8 @@ def create_cloud_router(
                     assigned_host_id=task.assigned_host_id,
                     attempt_count=task.attempt_count,
                     failure_reason=task.failure_reason,
+                    target_host_id=task.constraints.target_host_id,
+                    target_device_id=task.constraints.target_device_id,
                     created_at=task.created_at,
                 )
                 for task in tasks
@@ -303,4 +321,35 @@ def _build_constraints(model):
     return TaskConstraints(
         driver_type=model.driver_type,
         capability_tags=list(model.capability_tags),
+        target_host_id=model.target_host_id,
+        target_device_id=model.target_device_id,
+    )
+
+
+def _validate_task_target(pool, constraints) -> None:
+    if constraints.target_device_id and not constraints.target_host_id:
+        raise ValueError("target_device_id requires target_host_id")
+    if constraints.target_host_id is None:
+        return
+    if not any(host.host_id == constraints.target_host_id for host in pool.list_hosts()):
+        raise ValueError(f"target host {constraints.target_host_id!r} is not known")
+    if constraints.target_device_id is not None and not any(
+        device.host_id == constraints.target_host_id
+        and device.device_id == constraints.target_device_id
+        for device in pool.list_devices()
+    ):
+        raise ValueError(
+            f"target device {constraints.target_device_id!r} is not owned by "
+            f"host {constraints.target_host_id!r}"
+        )
+
+
+def _enforce_user_policy(principal, store, constraints) -> None:
+    if not principal.id.startswith("user:"):
+        return
+    policy = store.get_user_submission_policy(principal.id.removeprefix("user:"))
+    enforce_user_submission_policy(
+        policy,
+        target_host_id=constraints.target_host_id,
+        target_device_id=constraints.target_device_id,
     )

packages/cloud-platform/cloud/sdk/client.py

@@ -72,15 +72,24 @@ class CloudClient:
         workflow_definition_id: str | None = None,
         driver_type: str | None = None,
         capability_tags: list[str] | None = None,
+        target_host_id: str | None = None,
+        target_device_id: str | None = None,
     ) -> dict[str, Any]:
         payload: dict[str, Any] = {
             "goal": goal,
             "workflow_definition_id": workflow_definition_id,
         }
-        if driver_type is not None or capability_tags is not None:
+        if (
+            driver_type is not None
+            or capability_tags is not None
+            or target_host_id is not None
+            or target_device_id is not None
+        ):
             payload["constraints"] = {
                 "driver_type": driver_type,
                 "capability_tags": list(capability_tags or []),
+                "target_host_id": target_host_id,
+                "target_device_id": target_device_id,
             }
         resp = self._request("POST", "/tasks", json=payload)
         return resp.json()
@@ -209,6 +218,28 @@ class CloudClient:
     def revoke_user_sessions(self, user_id: str) -> None:
         self._request("DELETE", f"/users/{user_id}/sessions")
 
+    # ------------------------------------------------------------- governance
+
+    def get_user_submission_policy(self, user_id: str) -> dict[str, Any]:
+        return self._request("GET", f"/users/{user_id}/submission-policy").json()
+
+    def update_user_submission_policy(
+        self, user_id: str, **policy: Any
+    ) -> dict[str, Any]:
+        return self._request(
+            "PUT", f"/users/{user_id}/submission-policy", json=policy
+        ).json()
+
+    def get_host_governance_policy(self, host_id: str) -> dict[str, Any]:
+        return self._request("GET", f"/hosts/{host_id}/governance-policy").json()
+
+    def update_host_governance_policy(
+        self, host_id: str, **policy: Any
+    ) -> dict[str, Any]:
+        return self._request(
+            "PUT", f"/hosts/{host_id}/governance-policy", json=policy
+        ).json()
+
     # ------------------------------------------------------------------ helpers
 
     def _url(self, path: str) -> str:

packages/cloud-platform/cloud/sdk/governance_api.py

@@ -0,0 +1,169 @@
+from __future__ import annotations
+
+from uuid import uuid4
+
+from fastapi import APIRouter, HTTPException, Request, status
+
+from cloud.auth import AuthProvider, GOVERNANCE_ADMIN_SCOPE, GOVERNANCE_READ_SCOPE
+from cloud.observability import current_correlation_id
+from cloud.sdk.models import (
+    HostGovernancePolicyRequest,
+    HostGovernancePolicyResponse,
+    UserSubmissionPolicyRequest,
+    UserSubmissionPolicyResponse,
+)
+from cloud.user_auth import AuthAuditEvent
+from core.models import utc_now
+
+
+def create_governance_router(*, repository, auth_provider: AuthProvider) -> APIRouter:
+    router = APIRouter(prefix="/v1", tags=["cloud-governance"])
+
+    def authorize(request: Request, required_scope: str):
+        principal = auth_provider.authenticate(request)
+        if principal is None:
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail="unauthorized",
+                headers={"WWW-Authenticate": "Bearer"},
+            )
+        if principal.must_change_password or not principal.has_scope(required_scope):
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail=f"missing required scope: {required_scope}",
+            )
+        return principal
+
+    @router.get(
+        "/users/{user_id}/submission-policy",
+        response_model=UserSubmissionPolicyResponse,
+    )
+    def get_user_policy(user_id: str, request: Request) -> UserSubmissionPolicyResponse:
+        authorize(request, GOVERNANCE_READ_SCOPE)
+        policy = repository.get_user_submission_policy(user_id)
+        if policy is None:
+            raise HTTPException(status_code=404, detail="submission policy not found")
+        return UserSubmissionPolicyResponse(
+            user_id=policy.user_id,
+            revision=policy.revision,
+            submission_enabled=policy.submission_enabled,
+            allowed_host_ids=list(policy.allowed_host_ids)
+            if policy.allowed_host_ids is not None
+            else None,
+            allowed_device_targets=[
+                {"host_id": host_id, "device_id": device_id}
+                for host_id, device_id in policy.allowed_device_targets or ()
+            ]
+            if policy.allowed_device_targets is not None
+            else None,
+            updated_at=policy.updated_at,
+        )
+
+    @router.put(
+        "/users/{user_id}/submission-policy",
+        response_model=UserSubmissionPolicyResponse,
+    )
+    def put_user_policy(
+        user_id: str,
+        payload: UserSubmissionPolicyRequest,
+        request: Request,
+    ) -> UserSubmissionPolicyResponse:
+        principal = authorize(request, GOVERNANCE_ADMIN_SCOPE)
+        try:
+            policy = repository.upsert_user_submission_policy(
+                user_id=user_id,
+                submission_enabled=payload.submission_enabled,
+                allowed_host_ids=_unique_strings(payload.allowed_host_ids),
+                allowed_device_targets=(
+                    tuple((item.host_id, item.device_id) for item in payload.allowed_device_targets)
+                    if payload.allowed_device_targets is not None
+                    else None
+                ),
+                updated_at=utc_now(),
+            )
+        except KeyError as exc:
+            raise HTTPException(status_code=404, detail="user not found") from exc
+        _audit(repository, principal.id, user_id, "user_submission_policy_update")
+        return UserSubmissionPolicyResponse(
+            user_id=policy.user_id,
+            revision=policy.revision,
+            submission_enabled=policy.submission_enabled,
+            allowed_host_ids=list(policy.allowed_host_ids)
+            if policy.allowed_host_ids is not None
+            else None,
+            allowed_device_targets=[
+                {"host_id": host_id, "device_id": device_id}
+                for host_id, device_id in policy.allowed_device_targets or ()
+            ]
+            if policy.allowed_device_targets is not None
+            else None,
+            updated_at=policy.updated_at,
+        )
+
+    @router.get(
+        "/hosts/{host_id}/governance-policy",
+        response_model=HostGovernancePolicyResponse,
+    )
+    def get_host_policy(host_id: str, request: Request) -> HostGovernancePolicyResponse:
+        authorize(request, GOVERNANCE_READ_SCOPE)
+        policy = repository.get_host_governance_policy(host_id)
+        if policy is None:
+            raise HTTPException(status_code=404, detail="Host policy not found")
+        return _host_response(policy)
+
+    @router.put(
+        "/hosts/{host_id}/governance-policy",
+        response_model=HostGovernancePolicyResponse,
+    )
+    def put_host_policy(
+        host_id: str,
+        payload: HostGovernancePolicyRequest,
+        request: Request,
+    ) -> HostGovernancePolicyResponse:
+        principal = authorize(request, GOVERNANCE_ADMIN_SCOPE)
+        try:
+            policy = repository.upsert_host_governance_policy(
+                host_id=host_id,
+                self_submission_enabled=payload.self_submission_enabled,
+                max_active_tasks=payload.max_active_tasks,
+                daily_token_budget=payload.daily_token_budget,
+                updated_at=utc_now(),
+            )
+        except KeyError as exc:
+            raise HTTPException(status_code=404, detail="Host not found") from exc
+        _audit(repository, principal.id, host_id, "host_governance_policy_update")
+        return _host_response(policy)
+
+    return router
+
+
+def _unique_strings(values: list[str] | None) -> tuple[str, ...] | None:
+    if values is None:
+        return None
+    return tuple(dict.fromkeys(value for value in values if value))
+
+
+def _host_response(policy) -> HostGovernancePolicyResponse:
+    return HostGovernancePolicyResponse(
+        host_id=policy.host_id,
+        revision=policy.revision,
+        self_submission_enabled=policy.self_submission_enabled,
+        max_active_tasks=policy.max_active_tasks,
+        daily_token_budget=policy.daily_token_budget,
+        updated_at=policy.updated_at,
+    )
+
+
+def _audit(repository, actor_id: str, target: str, action: str) -> None:
+    repository.record_auth_audit(
+        AuthAuditEvent(
+            id=uuid4().hex,
+            occurred_at=utc_now(),
+            actor_principal_id=actor_id,
+            target_user_id=target if action.startswith("user_") else None,
+            action=action,
+            outcome="success",
+            correlation_id=current_correlation_id(),
+            metadata={"target": target},
+        )
+    )

packages/cloud-platform/cloud/sdk/models.py

@@ -11,6 +11,8 @@ from pydantic import BaseModel, Field
 class TaskConstraintsModel(BaseModel):
     driver_type: str | None = None
     capability_tags: list[str] = Field(default_factory=list)
+    target_host_id: str | None = Field(default=None, min_length=1)
+    target_device_id: str | None = Field(default=None, min_length=1)
 
 
 class TaskSubmissionRequest(BaseModel):
@@ -33,6 +35,8 @@ class TaskStatusResponse(BaseModel):
     attempt_count: int = 0
     lease_expires_at: datetime | None = None
     failure_reason: str | None = None
+    target_host_id: str | None = None
+    target_device_id: str | None = None
 
 
 class TaskListItem(BaseModel):
@@ -44,6 +48,8 @@ class TaskListItem(BaseModel):
     assigned_host_id: str | None = None
     attempt_count: int = 0
     failure_reason: str | None = None
+    target_host_id: str | None = None
+    target_device_id: str | None = None
     created_at: datetime
 
 
@@ -145,3 +151,32 @@ class PasswordResetRequest(BaseModel):
 
 class ErrorResponse(BaseModel):
     detail: str
+
+
+class DeviceTargetModel(BaseModel):
+    host_id: str = Field(min_length=1)
+    device_id: str = Field(min_length=1)
+
+
+class UserSubmissionPolicyRequest(BaseModel):
+    submission_enabled: bool = True
+    allowed_host_ids: list[str] | None = None
+    allowed_device_targets: list[DeviceTargetModel] | None = None
+
+
+class UserSubmissionPolicyResponse(UserSubmissionPolicyRequest):
+    user_id: str
+    revision: int
+    updated_at: datetime
+
+
+class HostGovernancePolicyRequest(BaseModel):
+    self_submission_enabled: bool = True
+    max_active_tasks: int | None = Field(default=None, ge=1)
+    daily_token_budget: int | None = Field(default=None, ge=1)
+
+
+class HostGovernancePolicyResponse(HostGovernancePolicyRequest):
+    host_id: str
+    revision: int
+    updated_at: datetime

packages/cloud-platform/cloud/sql_repository.py

@@ -19,9 +19,11 @@ from cloud.db_models import (
     ScheduledTaskRow,
     TaskAttemptRow,
     AuthAuditRow,
+    HostGovernancePolicyRow,
     LoginThrottleRow,
     UserRow,
     UserSessionRow,
+    UserSubmissionPolicyRow,
 )
 from cloud.observability import current_correlation_id
 from core.models import utc_now
@@ -807,6 +809,86 @@ class SQLAlchemyCloudRepository:
                 removed += len(stale_throttles)
         return removed
 
+    # -------------------------------------------------------------- governance
+
+    def get_user_submission_policy(self, user_id: str) -> Any | None:
+        with self._sessions() as session:
+            row = session.get(UserSubmissionPolicyRow, user_id)
+        return _user_submission_policy_from_row(row) if row is not None else None
+
+    def upsert_user_submission_policy(
+        self,
+        *,
+        user_id: str,
+        submission_enabled: bool,
+        allowed_host_ids: tuple[str, ...] | None,
+        allowed_device_targets: tuple[tuple[str, str], ...] | None,
+        updated_at: datetime,
+    ) -> Any:
+        with self._sessions.begin() as session:
+            if session.get(UserRow, user_id) is None:
+                raise KeyError(f"unknown user {user_id!r}")
+            row = session.get(UserSubmissionPolicyRow, user_id)
+            if row is None:
+                row = UserSubmissionPolicyRow(
+                    user_id=user_id,
+                    revision=1,
+                    submission_enabled=1 if submission_enabled else 0,
+                    allowed_host_ids_json=_dump_optional_list(allowed_host_ids),
+                    allowed_device_targets_json=_dump_optional_list(
+                        allowed_device_targets
+                    ),
+                    updated_at=_iso(updated_at),
+                )
+                session.add(row)
+            else:
+                row.revision += 1
+                row.submission_enabled = 1 if submission_enabled else 0
+                row.allowed_host_ids_json = _dump_optional_list(allowed_host_ids)
+                row.allowed_device_targets_json = _dump_optional_list(
+                    allowed_device_targets
+                )
+                row.updated_at = _iso(updated_at)
+            session.flush()
+            return _user_submission_policy_from_row(row)
+
+    def get_host_governance_policy(self, host_id: str) -> Any | None:
+        with self._sessions() as session:
+            row = session.get(HostGovernancePolicyRow, host_id)
+        return _host_governance_policy_from_row(row) if row is not None else None
+
+    def upsert_host_governance_policy(
+        self,
+        *,
+        host_id: str,
+        self_submission_enabled: bool,
+        max_active_tasks: int | None,
+        daily_token_budget: int | None,
+        updated_at: datetime,
+    ) -> Any:
+        with self._sessions.begin() as session:
+            if session.get(HostRow, host_id) is None:
+                raise KeyError(f"unknown host {host_id!r}")
+            row = session.get(HostGovernancePolicyRow, host_id)
+            if row is None:
+                row = HostGovernancePolicyRow(
+                    host_id=host_id,
+                    revision=1,
+                    self_submission_enabled=1 if self_submission_enabled else 0,
+                    max_active_tasks=max_active_tasks,
+                    daily_token_budget=daily_token_budget,
+                    updated_at=_iso(updated_at),
+                )
+                session.add(row)
+            else:
+                row.revision += 1
+                row.self_submission_enabled = 1 if self_submission_enabled else 0
+                row.max_active_tasks = max_active_tasks
+                row.daily_token_budget = daily_token_budget
+                row.updated_at = _iso(updated_at)
+            session.flush()
+            return _host_governance_policy_from_row(row)
+
     def list_reserved_device_ids(self, *, now: datetime) -> set[str]:
         with self._sessions() as session:
             device_ids = session.scalars(
@@ -1218,6 +1300,8 @@ def _task_from_row(row: ScheduledTaskRow) -> Any:
         constraints=TaskConstraints(
             driver_type=constraints_data.get("driver_type"),
             capability_tags=list(constraints_data.get("capability_tags") or []),
+            target_host_id=constraints_data.get("target_host_id"),
+            target_device_id=constraints_data.get("target_device_id"),
         ),
         status=row.status,
         assigned_device_id=row.assigned_device_id,
@@ -1232,6 +1316,66 @@ def _task_from_row(row: ScheduledTaskRow) -> Any:
     )
 
 
+def _dump_optional_list(value: tuple[Any, ...] | None) -> str | None:
+    return json.dumps(value, ensure_ascii=False) if value is not None else None
+
+
+def _load_optional_string_tuple(value: str | None) -> tuple[str, ...] | None:
+    if value is None:
+        return None
+    try:
+        parsed = json.loads(value)
+    except (TypeError, ValueError):
+        parsed = []
+    return tuple(item for item in parsed if isinstance(item, str))
+
+
+def _load_optional_device_targets(
+    value: str | None,
+) -> tuple[tuple[str, str], ...] | None:
+    if value is None:
+        return None
+    try:
+        parsed = json.loads(value)
+    except (TypeError, ValueError):
+        parsed = []
+    return tuple(
+        (item[0], item[1])
+        for item in parsed
+        if isinstance(item, list | tuple)
+        and len(item) == 2
+        and all(isinstance(part, str) for part in item)
+    )
+
+
+def _user_submission_policy_from_row(row: UserSubmissionPolicyRow) -> Any:
+    from cloud.governance import UserSubmissionPolicy
+
+    return UserSubmissionPolicy(
+        user_id=row.user_id,
+        revision=row.revision,
+        submission_enabled=bool(row.submission_enabled),
+        allowed_host_ids=_load_optional_string_tuple(row.allowed_host_ids_json),
+        allowed_device_targets=_load_optional_device_targets(
+            row.allowed_device_targets_json
+        ),
+        updated_at=_parse_dt(row.updated_at) or utc_now(),
+    )
+
+
+def _host_governance_policy_from_row(row: HostGovernancePolicyRow) -> Any:
+    from cloud.governance import HostGovernancePolicy
+
+    return HostGovernancePolicy(
+        host_id=row.host_id,
+        revision=row.revision,
+        self_submission_enabled=bool(row.self_submission_enabled),
+        max_active_tasks=row.max_active_tasks,
+        daily_token_budget=row.daily_token_budget,
+        updated_at=_parse_dt(row.updated_at) or utc_now(),
+    )
+
+
 def _task_attempt_from_row(row: TaskAttemptRow) -> Any:
     from cloud.repository import TaskAttemptRecord