feat(task-plane): enforce first-class scope isolation

active-runtime/app/task_plane/api.py

@@ -34,6 +34,7 @@ from .models import (
 )
 from .policy import PolicyEngine
 from .store import TaskPlaneStore
+from .scope import bind_task_scope
 from .worker_registry import WorkerRegistry
 
 
@@ -68,28 +69,54 @@ class TaskPlaneService:
         self.delegation = DelegationEngine(store)
         self.store.probe_attestation_verifier = self.policy.probe_attestation_verifier
 
-    def create_task(self, body: dict[str, Any], idempotency_key: str | None = None) -> dict[str, Any]:
+    def _authorized_task(self, task_id: str, actor: str, action: str) -> TaskRecord:
+        task = self.store.get_task(task_id, self.policy.actor_context(actor))
+        self.policy.assert_allowed(action, actor, task=task)
+        return task
+
+    def create_task(
+        self,
+        body: dict[str, Any],
+        idempotency_key: str | None = None,
+        *,
+        actor_identity: str | None = None,
+    ) -> dict[str, Any]:
         if body.get("parent_task_id"):
             raise ForbiddenError("child tasks must use the delegate endpoint")
         issuer = body.get("issuer", {})
-        actor = issuer.get("id", "anon") if isinstance(issuer, dict) else "anon"
+        declared_actor = issuer.get("id", "anon") if isinstance(issuer, dict) else "anon"
+        actor = actor_identity or declared_actor
         self.policy.assert_allowed("create", actor)
         try:
             plan = TaskPlan.from_dict(body)
         except (TypeError, ValueError) as exc:
             raise ValidationError("invalid task plan") from exc
+        if actor_identity is not None:
+            plan = bind_task_scope(
+                plan, self.policy.actor_context(actor_identity), public_create=True
+            )
         task = self.store.create_task_with_event(plan, idempotency_key=idempotency_key)
         return task.to_dict()
 
     def list_tasks(self, actor: str | None = None) -> list[dict[str, Any]]:
+        context = None
         if actor is not None:
             self.policy.assert_allowed("read", actor)
-        return [task.to_dict(include_task=False) for task in self.store.list_tasks()]
+            context = self.policy.actor_context(actor)
+        return [
+            task.to_dict(include_task=False)
+            for task in self.store.list_tasks(context)
+        ]
 
     def get_task(self, task_id: str, actor: str | None = None) -> dict[str, Any]:
+        context = None
         if actor is not None:
-            self.policy.assert_allowed("read", actor, task=self.store.get_task(task_id))
-        return self.store.get_task(task_id).to_dict()
+            self.policy.assert_allowed("read", actor)
+            context = self.policy.actor_context(actor)
+        task = self.store.get_task(task_id, context)
+        if actor is not None:
+            self.policy.assert_allowed("read", actor, task=task)
+        return task.to_dict()
 
     def get_worker(self, worker_id: str, actor: str | None = None) -> dict[str, Any]:
         if actor is not None:
@@ -163,9 +190,8 @@ class TaskPlaneService:
         if not worker_id:
             raise ValidationError("worker_id required")
 
-        task = self.store.get_task(task_id)
+        task = self._authorized_task(task_id, actor, "claim")
         self.policy.assert_readonly_worker_binding(actor, str(worker_id))
-        self.policy.assert_allowed("claim", actor, task=task)
         lease_seconds = _bounded_int(
             body.get("lease_seconds", 30), "lease_seconds", 1, 3600
         )
@@ -184,10 +210,9 @@ class TaskPlaneService:
         if not claim_id:
             raise ValidationError("claim_id required")
         actor = body.get("actor_identity") or body.get("worker_identity") or "worker"
-        task = self.store.get_task(task_id)
+        task = self._authorized_task(task_id, actor, "heartbeat")
         claim = self.store.get_claim(claim_id)
         self.policy.assert_readonly_worker_binding(actor, claim.worker_id)
-        self.policy.assert_allowed("heartbeat", actor, task=task)
         if claim.worker_identity != actor:
             raise ForbiddenError("claim identity mismatch")
         lease_seconds = _bounded_int(
@@ -206,10 +231,9 @@ class TaskPlaneService:
         message = body.get("message")
         if progress is None:
             raise ValidationError("progress required")
-        task = self.store.get_task(task_id)
+        task = self._authorized_task(task_id, actor, "progress")
         claim = self.store.get_claim(claim_id)
         self.policy.assert_readonly_worker_binding(actor, claim.worker_id)
-        self.policy.assert_allowed("progress", actor, task=task)
         if task.task_plan.data_classification == "credential-secret":
             if message not in {None, ""} or contains_secret_text(stage):
                 raise ValidationError(
@@ -234,7 +258,7 @@ class TaskPlaneService:
         if not claim_id:
             raise ValidationError("claim_id required")
         actor = body.get("actor_identity") or body.get("worker_identity") or "worker"
-        task = self.store.get_task(task_id)
+        task = self._authorized_task(task_id, actor, "result")
 
         status = body.get("status", "COMPLETED")
         outputs = body.get("outputs", [])
@@ -247,7 +271,6 @@ class TaskPlaneService:
             raise ValidationError("evidence items must be objects")
         claim = self.store.get_claim(claim_id)
         self.policy.assert_readonly_worker_binding(actor, claim.worker_id)
-        self.policy.assert_allowed("result", actor, task=task)
         if claim.worker_identity != actor:
             raise ForbiddenError("claim identity mismatch")
         if task.task_plan.data_classification == "credential-secret":
@@ -326,8 +349,7 @@ class TaskPlaneService:
         reason = body.get("reason", "failed")
         actor = body.get("actor_identity") or body.get("worker_identity") or "worker"
         self.policy.assert_allowed("fail", actor)
-        task = self.store.get_task(task_id)
-        self.policy.assert_allowed("fail", actor, task=task)
+        task = self._authorized_task(task_id, actor, "fail")
         claim = self.store.get_claim(claim_id)
         if claim.worker_identity != actor:
             raise ForbiddenError("claim identity mismatch")
@@ -339,8 +361,7 @@ class TaskPlaneService:
         actor = body.get("actor_identity") or body.get("worker_identity") or "owner"
         self.policy.assert_allowed("cancel", actor)
         reason = body.get("reason", "cancelled")
-        task = self.store.get_task(task_id)
-        self.policy.assert_allowed("cancel", actor, task=task)
+        task = self._authorized_task(task_id, actor, "cancel")
         updated = self.store.cancel_task_with_event(
             task_id, task.revision, reason=reason
         )
@@ -349,8 +370,7 @@ class TaskPlaneService:
     def delegate(self, task_id: str, body: dict[str, Any]) -> dict[str, Any]:
         actor = body.get("actor_identity") or "owner"
         self.policy.assert_allowed("delegate", actor)
-        parent = self.store.get_task(task_id)
-        self.policy.assert_allowed("delegate", actor, task=parent)
+        parent = self._authorized_task(task_id, actor, "delegate")
         if "task" in body:
             try:
                 child_plan = TaskPlan.from_dict(body["task"])
@@ -364,25 +384,31 @@ class TaskPlaneService:
             except (TypeError, ValueError) as exc:
                 raise ValidationError("invalid child task plan") from exc
         child_plan.parent_task_id = task_id
+        child_plan = bind_task_scope(
+            child_plan, self.policy.actor_context(actor), public_create=False
+        )
         child = self.delegation.delegate(task_id, child_plan, caller=actor)
         return child.to_dict()
 
     def get_events(self, task_id: str, actor: str | None = None) -> list[dict[str, Any]]:
         if actor is not None:
-            self.policy.assert_allowed("audit", actor, task=self.store.get_task(task_id))
-        self.store.get_task(task_id)
+            self._authorized_task(task_id, actor, "audit")
+        else:
+            self.store.get_task(task_id)
         return self.store.list_events(task_id)
 
     def get_evidence(self, task_id: str, actor: str | None = None) -> list[dict[str, Any]]:
         if actor is not None:
-            self.policy.assert_allowed("audit", actor, task=self.store.get_task(task_id))
-        self.store.get_task(task_id)
+            self._authorized_task(task_id, actor, "audit")
+        else:
+            self.store.get_task(task_id)
         return self.evidence.list(task_id)
 
     def get_children(self, task_id: str, actor: str | None = None) -> list[str]:
         if actor is not None:
-            self.policy.assert_allowed("read", actor, task=self.store.get_task(task_id))
-        self.store.get_task(task_id)
+            self._authorized_task(task_id, actor, "read")
+        else:
+            self.store.get_task(task_id)
         return self.store.list_children(task_id)
 
 
@@ -506,7 +532,11 @@ class TaskPlaneRequestHandler(BaseHTTPRequestHandler):
                 ):
                     raise ForbiddenError("issuer identity mismatch")
                 body["issuer"] = {"type": "authenticated", "id": identity}
-                return 201, self.service.create_task(body or {}, headers.get("idempotency-key"))
+                return 201, self.service.create_task(
+                    body or {},
+                    headers.get("idempotency-key"),
+                    actor_identity=identity,
+                )
             if body.get("actor_identity") not in {None, identity}:
                 raise ForbiddenError("actor identity mismatch")
             body["actor_identity"] = identity

active-runtime/app/task_plane/models.py

@@ -13,6 +13,8 @@ from dataclasses import dataclass, field
 from datetime import datetime, timezone
 from typing import Any
 
+from .scope import normalize_task_scope
+
 
 SENSITIVE_KEYS = {
     "secret",
@@ -66,6 +68,12 @@ TASK_PLAN_KEYS = {
     "allowed_input_paths",
     "allowed_output_paths",
     "output_policy",
+    "scope_type",
+    "tenant_id",
+    "user_id",
+    "project_id",
+    "conversation_id",
+    "run_id",
 }
 
 
@@ -237,6 +245,12 @@ class TaskPlan:
     schema_version: int = 1
     task_id: str | None = None
     task_type: str = "code.change"
+    scope_type: str = "legacy"
+    tenant_id: str | None = None
+    user_id: str | None = None
+    project_id: str | None = None
+    conversation_id: str | None = None
+    run_id: str | None = None
     issuer: dict[str, Any] = field(default_factory=dict)
     intent: str | None = None
     required_capabilities: list[str] = field(default_factory=list)
@@ -311,10 +325,17 @@ class TaskPlan:
                 raise ValueError("credential-secret requires metadata-only logging")
             if output_policy != "hash-count-status-only":
                 raise ValueError("credential-secret requires restricted output")
+        scope = normalize_task_scope(data)
         return cls(
             schema_version=schema_version,
             task_id=validate_identifier(data["task_id"], "task_id"),
             task_type=validate_type(data.get("task_type", "code.change"), "task_type"),
+            scope_type=scope.scope_type,
+            tenant_id=scope.tenant_id,
+            user_id=scope.user_id,
+            project_id=scope.project_id,
+            conversation_id=scope.conversation_id,
+            run_id=scope.run_id,
             issuer=issuer,
             intent=data.get("intent"),
             required_capabilities=_normalize_list(data.get("required_capabilities", [])),
@@ -348,6 +369,12 @@ class TaskPlan:
             "schema_version": self.schema_version,
             "task_id": self.task_id,
             "task_type": self.task_type,
+            "scope_type": self.scope_type,
+            "tenant_id": self.tenant_id,
+            "user_id": self.user_id,
+            "project_id": self.project_id,
+            "conversation_id": self.conversation_id,
+            "run_id": self.run_id,
             "issuer": self.issuer,
             "intent": self.intent,
             "required_capabilities": list(self.required_capabilities),

active-runtime/app/task_plane/policy.py

@@ -3,10 +3,11 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-from typing import Callable
+from typing import Any, Callable, Mapping
 
 from .errors import ForbiddenError, LockedError
 from .models import ProbeEvidence, ResultEnvelope, TaskRecord
+from .scope import TaskActorContext, actor_context as build_actor_context
 
 
 RISK_RANK = {"low": 1, "medium": 2, "high": 3}
@@ -93,6 +94,10 @@ class PolicyEngine:
         result_signature_verifier: Callable[[ResultEnvelope], bool] | None = None,
         probe_attestation_verifier: Callable[[ProbeEvidence], bool] | None = None,
         identity_capabilities: dict[str, set[str]] | None = None,
+        identity_actor_contexts: Mapping[
+            str, TaskActorContext | Mapping[str, Any]
+        ]
+        | None = None,
         admin_identities: set[str] | None = None,
         readonly_worker_execution_enabled: bool = False,
         readonly_worker_profiles: list[ReadonlyWorkerProfile] | None = None,
@@ -109,12 +114,26 @@ class PolicyEngine:
             identity: set(capabilities)
             for identity, capabilities in (identity_capabilities or {}).items()
         }
+        self.identity_actor_contexts = dict(identity_actor_contexts or {})
         self.admin_identities = set(admin_identities or ())
         self.readonly_worker_execution_enabled = readonly_worker_execution_enabled
         self.readonly_worker_profiles = {
             profile.identity: profile for profile in (readonly_worker_profiles or [])
         }
 
+    def actor_context(self, identity: str) -> TaskActorContext:
+        configured = self.identity_actor_contexts.get(
+            identity, self.identity_actor_contexts.get("*")
+        )
+        capabilities = set(
+            self.identity_capabilities.get(
+                identity, self.identity_capabilities.get("*", set())
+            )
+        )
+        if identity in self.admin_identities:
+            capabilities.add("task.admin")
+        return build_actor_context(identity, configured, capabilities)
+
     def _readonly_worker_allows(
         self, action: str, actor: str, task: TaskRecord | None
     ) -> bool:

active-runtime/app/task_plane/scope.py

@@ -0,0 +1,181 @@
+"""Trusted TaskPlane scope model and SQL authorization helpers."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, replace
+from typing import Any, Mapping
+
+from .errors import ForbiddenError, ValidationError
+
+SCOPE_FIELDS = ("tenant_id", "user_id", "project_id", "conversation_id", "run_id")
+PRIVILEGED_CAPABILITIES = frozenset(
+    {"*", "admin", "system.admin", "task.admin", "task.read.all"}
+)
+
+
+def _value(value: Any, field: str) -> Any:
+    if hasattr(value, "task_plan"):
+        value = value.task_plan
+    if isinstance(value, Mapping):
+        return value.get(field)
+    return getattr(value, field, None)
+
+
+def _text(value: Any) -> str | None:
+    if value is None:
+        return None
+    result = str(value).strip()
+    return result or None
+
+
+@dataclass(frozen=True)
+class TaskScope:
+    scope_type: str
+    tenant_id: str | None = None
+    user_id: str | None = None
+    project_id: str | None = None
+    conversation_id: str | None = None
+    run_id: str | None = None
+
+
+@dataclass(frozen=True)
+class TaskActorContext:
+    identity: str
+    scope_type: str = "legacy"
+    tenant_id: str | None = None
+    user_id: str | None = None
+    project_id: str | None = None
+    conversation_id: str | None = None
+    run_id: str | None = None
+    capabilities: frozenset[str] = frozenset()
+
+
+def normalize_task_scope(value: Any) -> TaskScope:
+    fields = {field: _text(_value(value, field)) for field in SCOPE_FIELDS}
+    declared = _text(_value(value, "scope_type"))
+    if declared is None:
+        declared = "tenant" if fields["tenant_id"] else "legacy"
+    if declared == "tenant":
+        if fields["tenant_id"] is None:
+            raise ValidationError(
+                "tenant scope requires tenant_id", code="TASK_SCOPE_TENANT_ID_REQUIRED"
+            )
+    elif declared == "legacy":
+        if any(fields.values()):
+            raise ValidationError(
+                "legacy scope cannot contain scoped ownership",
+                code="TASK_SCOPE_LEGACY_WITH_SCOPED_FIELDS",
+            )
+    elif declared == "system":
+        if any(fields[field] for field in ("tenant_id", "user_id", "project_id", "conversation_id")):
+            raise ValidationError(
+                "system scope cannot contain tenant ownership",
+                code="TASK_SCOPE_SYSTEM_WITH_TENANT_FIELDS",
+            )
+    else:
+        raise ValidationError("invalid task scope type", code="TASK_SCOPE_TYPE_INVALID")
+    return TaskScope(declared, **fields)
+
+
+def actor_is_privileged(actor: TaskActorContext) -> bool:
+    return bool(actor.capabilities & PRIVILEGED_CAPABILITIES)
+
+
+def actor_context(
+    identity: str,
+    configured: TaskActorContext | Mapping[str, Any] | None,
+    capabilities: set[str] | frozenset[str],
+) -> TaskActorContext:
+    if isinstance(configured, TaskActorContext):
+        context = configured
+    else:
+        data = dict(configured or {})
+        data.pop("identity", None)
+        data.pop("capabilities", None)
+        context = TaskActorContext(identity=str(identity), **data)
+    scope = normalize_task_scope(context)
+    return replace(
+        context,
+        identity=str(identity),
+        scope_type=scope.scope_type,
+        tenant_id=scope.tenant_id,
+        user_id=scope.user_id,
+        project_id=scope.project_id,
+        conversation_id=scope.conversation_id,
+        run_id=scope.run_id,
+        capabilities=frozenset(str(item) for item in capabilities),
+    )
+
+
+def bind_task_scope(plan: Any, actor: TaskActorContext, *, public_create: bool) -> Any:
+    requested = normalize_task_scope(plan)
+    if actor_is_privileged(actor):
+        trusted = normalize_task_scope(actor)
+        if requested.scope_type == "legacy" and trusted.scope_type != "legacy":
+            plan.scope_type = trusted.scope_type
+            for field in SCOPE_FIELDS:
+                setattr(plan, field, getattr(actor, field))
+        return plan
+    trusted = normalize_task_scope(actor)
+    if public_create and trusted.scope_type == "legacy":
+        raise ForbiddenError(
+            "authenticated identity has no tenant scope",
+            code="TASK_ACTOR_SCOPE_UNAVAILABLE",
+        )
+    if trusted.scope_type != "tenant":
+        if requested.scope_type == "legacy":
+            plan.scope_type = trusted.scope_type
+            for field in SCOPE_FIELDS:
+                setattr(plan, field, getattr(actor, field))
+            return plan
+        if requested.scope_type != trusted.scope_type:
+            raise ForbiddenError("task scope mismatch", code="TASK_SCOPE_MISMATCH_DENY")
+        return plan
+    if requested.scope_type not in {"legacy", "tenant"}:
+        raise ForbiddenError("task scope mismatch", code="TASK_SCOPE_MISMATCH_DENY")
+    updates: dict[str, Any] = {"scope_type": "tenant"}
+    for field in SCOPE_FIELDS:
+        declared = getattr(requested, field)
+        authoritative = getattr(actor, field)
+        if declared is not None and declared != authoritative:
+            raise ForbiddenError(
+                "task scope mismatch",
+                code="TASK_SCOPE_MISMATCH_DENY",
+                details={"field": field},
+            )
+        updates[field] = authoritative
+    for field, item in updates.items():
+        setattr(plan, field, item)
+    return plan
+
+
+def actor_scope_sql(actor: TaskActorContext | None) -> tuple[str, tuple[Any, ...]]:
+    if actor is None or actor_is_privileged(actor):
+        return "", ()
+    scope = normalize_task_scope(actor)
+    if scope.scope_type == "legacy":
+        return "scope_type = 'legacy'", ()
+    if scope.scope_type == "system":
+        return "scope_type = 'system'", ()
+    clauses = ["scope_type = 'tenant'", "tenant_id = ?"]
+    params: list[Any] = [scope.tenant_id]
+    for field in ("user_id", "project_id", "conversation_id", "run_id"):
+        value = getattr(scope, field)
+        if value is None:
+            clauses.append(f"{field} IS NULL")
+        else:
+            clauses.append(f"({field} IS NULL OR {field} = ?)")
+            params.append(value)
+    return " AND ".join(clauses), tuple(params)
+
+
+def task_scope_values(value: Any) -> tuple[str | None, ...]:
+    scope = normalize_task_scope(value)
+    return (
+        scope.scope_type,
+        scope.tenant_id,
+        scope.user_id,
+        scope.project_id,
+        scope.conversation_id,
+        scope.run_id,
+    )

active-runtime/app/task_plane/store.py

@@ -32,6 +32,7 @@ from .models import (
     redact_value,
     validate_identifier,
 )
+from .scope import TaskActorContext, actor_scope_sql, task_scope_values
 
 
 EventEnvelope = tuple[str, str, str, str, str, float]
@@ -134,6 +135,12 @@ class TaskPlaneStore:
                     status TEXT NOT NULL,
                     task_json TEXT NOT NULL,
                     task_type TEXT NOT NULL,
+                    scope_type TEXT NOT NULL DEFAULT 'legacy',
+                    tenant_id TEXT,
+                    user_id TEXT,
+                    project_id TEXT,
+                    conversation_id TEXT,
+                    run_id TEXT,
                     root_task_id TEXT NOT NULL,
                     parent_task_id TEXT,
                     correlation_id TEXT,
@@ -272,6 +279,21 @@ class TaskPlaneStore:
                 )
                 """
             )
+            self._ensure_columns(
+                "tasks",
+                {
+                    "scope_type": "TEXT NOT NULL DEFAULT 'legacy'",
+                    "tenant_id": "TEXT",
+                    "user_id": "TEXT",
+                    "project_id": "TEXT",
+                    "conversation_id": "TEXT",
+                    "run_id": "TEXT",
+                },
+            )
+            self.conn.execute(
+                "UPDATE tasks SET scope_type='legacy' "
+                "WHERE scope_type IS NULL OR TRIM(scope_type)=''"
+            )
             self._ensure_columns(
                 "workers",
                 {
@@ -299,6 +321,14 @@ class TaskPlaneStore:
             )
             self.conn.execute("CREATE INDEX IF NOT EXISTS ix_tasks_status ON tasks(status)")
             self.conn.execute("CREATE INDEX IF NOT EXISTS ix_tasks_parent ON tasks(parent_task_id)")
+            self.conn.execute(
+                "CREATE INDEX IF NOT EXISTS ix_tasks_scope_tenant_project "
+                "ON tasks(scope_type, tenant_id, project_id)"
+            )
+            self.conn.execute(
+                "CREATE INDEX IF NOT EXISTS ix_tasks_scope_owner "
+                "ON tasks(scope_type, tenant_id, user_id, conversation_id, run_id)"
+            )
             self.conn.execute("CREATE INDEX IF NOT EXISTS ix_workers_status ON workers(status)")
             self.conn.execute("CREATE INDEX IF NOT EXISTS ix_claims_task ON claims(task_id)")
             self.conn.execute(
@@ -347,6 +377,42 @@ class TaskPlaneStore:
             raise NotFoundError("task")
         task_json = json.loads(row["task_json"])
         task_plan = TaskPlan.from_dict(task_json)
+        stored_scope = tuple(
+            row[field]
+            for field in (
+                "scope_type",
+                "tenant_id",
+                "user_id",
+                "project_id",
+                "conversation_id",
+                "run_id",
+            )
+        )
+        json_has_scope = any(
+            field in task_json
+            for field in (
+                "scope_type",
+                "tenant_id",
+                "user_id",
+                "project_id",
+                "conversation_id",
+                "run_id",
+            )
+        )
+        if json_has_scope and stored_scope != task_scope_values(task_plan):
+            raise ConflictError(
+                "task scope persistence mismatch",
+                code="TASK_SCOPE_PERSISTENCE_MISMATCH",
+            )
+        if not json_has_scope:
+            (
+                task_plan.scope_type,
+                task_plan.tenant_id,
+                task_plan.user_id,
+                task_plan.project_id,
+                task_plan.conversation_id,
+                task_plan.run_id,
+            ) = stored_scope
         task_plan.parent_task_id = row["parent_task_id"]
         task_plan.correlation_id = row["correlation_id"]
         return TaskRecord(
@@ -389,6 +455,7 @@ class TaskPlaneStore:
         task_id = task_plan.task_id or f"task-{uuid.uuid4()}"
         task_plan.task_id = task_id
         task_json = redact_value(task_plan.to_dict(redact=False))
+        scope_values = task_scope_values(task_plan)
         now = _now()
         with self._transaction():
             if idempotency_key:
@@ -397,7 +464,11 @@ class TaskPlaneStore:
                     (idempotency_key,),
                 ).fetchone()
                 if existing:
-                    if _canonical(json.loads(existing["task_json"])) != _canonical(task_json):
+                    existing_plan = TaskPlan.from_dict(
+                        json.loads(existing["task_json"])
+                    )
+                    existing_json = redact_value(existing_plan.to_dict(redact=False))
+                    if _canonical(existing_json) != _canonical(task_json):
                         raise ConflictError("idempotency key payload mismatch")
                     return self.get_task(existing["task_id"])
             root_task_id = task_id
@@ -412,14 +483,18 @@ class TaskPlaneStore:
             self.conn.execute(
                 """
                 INSERT INTO tasks
-                (task_id, revision, status, task_json, task_type, root_task_id, parent_task_id, correlation_id, idempotency_key, created_at, updated_at, failure_reason)
-                VALUES (?, 1, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL)
+                (task_id, revision, status, task_json, task_type,
+                 scope_type, tenant_id, user_id, project_id, conversation_id, run_id,
+                 root_task_id, parent_task_id, correlation_id, idempotency_key,
+                 created_at, updated_at, failure_reason)
+                VALUES (?, 1, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL)
                 """,
                 (
                     task_id,
                     "CREATED",
                     json.dumps(task_json, ensure_ascii=False),
                     task_plan.task_type,
+                    *scope_values,
                     root_task_id,
                     task_plan.parent_task_id,
                     task_plan.correlation_id,
@@ -451,21 +526,30 @@ class TaskPlaneStore:
                 )
             return self.get_task(task.task_id)
 
-    def get_task(self, task_id: str) -> TaskRecord:
+    def get_task(
+        self, task_id: str, actor_context: TaskActorContext | None = None
+    ) -> TaskRecord:
         task_id = _validated_id(task_id, "task_id")
+        predicate, scope_params = actor_scope_sql(actor_context)
+        sql = "SELECT * FROM tasks WHERE task_id=?"
+        if predicate:
+            sql += f" AND ({predicate})"
         with self._lock:
-            row = self.conn.execute(
-                "SELECT * FROM tasks WHERE task_id=?", (task_id,)
-            ).fetchone()
+            row = self.conn.execute(sql, (task_id, *scope_params)).fetchone()
         if row is None:
             raise NotFoundError(f"task {task_id} not found")
         return self._row_to_task(row)
 
-    def list_tasks(self) -> list[TaskRecord]:
+    def list_tasks(
+        self, actor_context: TaskActorContext | None = None
+    ) -> list[TaskRecord]:
+        predicate, scope_params = actor_scope_sql(actor_context)
+        sql = "SELECT * FROM tasks"
+        if predicate:
+            sql += f" WHERE ({predicate})"
+        sql += " ORDER BY created_at DESC"
         with self._lock:
-            rows = self.conn.execute(
-                "SELECT * FROM tasks ORDER BY created_at DESC"
-            ).fetchall()
+            rows = self.conn.execute(sql, scope_params).fetchall()
         return [self._row_to_task(row) for row in rows]
 
     def get_task_budget(self, task_id: str) -> Budget:

docs/superpowers/specs/2026-08-16-r42-p0a-taskplane-scope-isolation-design.md

@@ -0,0 +1,54 @@
+# R4.2 P0-A TaskPlane Scope Isolation Design
+
+Status: approved candidate design. This document is not evidence of a protected-main merge or production deployment.
+
+## Boundary
+
+- Task: `r42-p0a-taskplane-scope-isolation-20260816`
+- Governed Job: `job-3fd743b51971c6007923699c1233f84c`
+- Repository: `aios/aios`
+- Base: `0eb6e1a864edfdaeef37b438991707a110e21558`
+- Target branch: `agent/r42-p0a-taskplane-scope-isolation-20260816`
+- Generation: `1`
+
+The implementation reuses this exact boundary. It does not create a replacement Task, Job, branch, generation, or idempotency key.
+
+## Model and invariants
+
+`TaskPlan` gains first-class `scope_type`, `tenant_id`, `user_id`, `project_id`, `conversation_id`, and `run_id` fields. The fields participate in accepted-key validation, parsing, serialization, persistence, and authorization.
+
+Historical rows without scope data normalize to `scope_type=legacy`; no tenant is inferred. Tenant-scoped tasks require a non-empty `tenant_id`. System/global resources use an explicit `scope_type=system` invariant and do not receive a fabricated tenant. A task may add user, project, conversation, and run ownership within its tenant.
+
+## Trusted ActorContext
+
+TaskPlane defines a small immutable `TaskActorContext` used only for authorization. The HTTP authenticator returns an identity. `PolicyEngine` resolves that identity through server-owned policy configuration into the effective context and capabilities. Request fields such as `tenant_id`, `user_id`, `group_id`, and `role` are declarations only and never expand authority.
+
+Public create, list, and get paths always use the resolved context. Internal compatibility calls may omit a context and retain explicit legacy behavior, but the HTTP adapter never uses that path.
+
+## Persistence
+
+SQLite remains the current runtime. `_create_schema()` performs an idempotent forward migration that adds the six scope columns and indexes, then backfills only missing `scope_type` values to `legacy`.
+
+New writes persist scope in both the canonical `task_json` representation and first-class columns. Reads accept historical JSON without scope, using legacy columns, but reject a new-row mismatch between JSON and columns. SQL list/get predicates are parameterized and applied before row hydration.
+
+PostgreSQL remains a separate migration target and is not claimed by this change.
+
+## Authorization flow
+
+1. HTTP bearer authentication yields an identity.
+2. `PolicyEngine` verifies capability and resolves the identity to `TaskActorContext`.
+3. Create binds the requested scope to the trusted context and rejects mismatches.
+4. List/get pass the trusted context to `TaskPlaneStore`.
+5. Store applies tenant and ownership predicates in SQL; privileged capabilities may explicitly cross scope.
+
+Tenant actors cannot read legacy tasks, another tenant, or mismatched owned resources. Legacy actors can only read legacy tasks. System actors require explicit privileged capability for cross-scope access.
+
+## Failure behavior
+
+Missing tenant scope, forged scope, unknown scope types, JSON/column divergence, and unauthorized list/get operations fail closed with canonical TaskPlane errors. Public get does not reveal whether an out-of-scope task exists.
+
+## Verification
+
+Tests exercise real model round trips, SQLite migration/restart, Service/Policy/Store wiring, tenant A versus tenant B list/get denial, ownership filtering, forged create payload rejection, legacy isolation, privileged access, and SQL parameterization. Existing TaskPlane API, RBAC, recovery/security, and security-contract suites remain mandatory.
+
+The candidate is complete only after real `TASK_CLAIMED`, `TASK_HEARTBEAT`, exact WriterLease, governed commit, OID convergence, result submission, and `TASK_VERIFIED`. Protected main and production remain unchanged.

docs/task-plane/r42-p0a-scope-isolation.md

@@ -0,0 +1,9 @@
+# R4.2 P0-A TaskPlane Scope Isolation
+
+Status: candidate implementation. This is not proof of protected-main promotion or production deployment.
+
+Task scope is now represented by `scope_type`, `tenant_id`, `user_id`, `project_id`, `conversation_id`, and `run_id` in the model and SQLite persistence. Historical rows become `legacy`; their tenant is never guessed. Tenant tasks require `tenant_id`. System resources use `scope_type=system` without a fabricated tenant.
+
+The HTTP adapter uses the authenticated identity. `PolicyEngine` resolves that identity through server-owned `identity_actor_contexts`; request payload fields do not grant scope. Public create binds the trusted scope, and public list/get apply parameterized scope predicates in `TaskPlaneStore`.
+
+SQLite is the current runtime database. PostgreSQL remains a separate migration and runtime-acceptance gate.

tests/task_plane/public_interface/test_shadow_client_rbac.py

@@ -45,6 +45,9 @@ class ShadowClientRBACPublicInterfaceTests(unittest.TestCase):
             identity_verifier=self.IdentityVerifier({"shadow-client"}),
             execution_enabled=False,
             identity_capabilities={"shadow-client": set(capabilities)},
+            identity_actor_contexts={
+                "shadow-client": {"scope_type": "system"}
+            },
         )
         service = self.TaskPlaneService(store, policy=policy)
 

tests/task_plane/public_interface/test_task_plane_scope_isolation.py

@@ -0,0 +1,298 @@
+from __future__ import annotations
+
+import json
+import os
+import sqlite3
+import sys
+import threading
+import urllib.error
+import urllib.request
+from http.server import ThreadingHTTPServer
+
+import pytest
+
+ROOT = os.getcwd()
+if os.path.join(ROOT, "active-runtime") not in sys.path:
+    sys.path.append(os.path.join(ROOT, "active-runtime"))
+
+from app.task_plane.api import TaskPlaneRequestHandler, TaskPlaneService
+from app.task_plane.errors import ForbiddenError, NotFoundError, ValidationError
+from app.task_plane.models import TaskPlan
+from app.task_plane.policy import PolicyEngine
+from app.task_plane.store import TaskPlaneStore
+
+
+def _policy() -> PolicyEngine:
+    return PolicyEngine(
+        identity_capabilities={
+            "actor-a": {"task.creator", "task.read"},
+            "actor-b": {"task.creator", "task.read"},
+            "reader-a": {"task.read"},
+            "reader-b": {"task.read"},
+            "owner": {"task.creator", "task.read", "task.read.all"},
+            "unbound": {"task.creator", "task.read"},
+        },
+        identity_actor_contexts={
+            "actor-a": {
+                "scope_type": "tenant",
+                "tenant_id": "tenant-a",
+                "user_id": "user-a",
+                "project_id": "project-a",
+                "conversation_id": "conversation-a",
+                "run_id": "run-a",
+            },
+            "reader-a": {
+                "scope_type": "tenant",
+                "tenant_id": "tenant-a",
+                "user_id": "user-a",
+                "project_id": "project-a",
+                "conversation_id": "conversation-a",
+                "run_id": "run-a",
+            },
+            "actor-b": {
+                "scope_type": "tenant",
+                "tenant_id": "tenant-b",
+                "user_id": "user-b",
+                "project_id": "project-b",
+            },
+            "reader-b": {
+                "scope_type": "tenant",
+                "tenant_id": "tenant-b",
+                "user_id": "user-b",
+                "project_id": "project-b",
+            },
+            "owner": {"scope_type": "system"},
+        },
+    )
+
+
+def _service() -> TaskPlaneService:
+    return TaskPlaneService(TaskPlaneStore(":memory:"), policy=_policy())
+
+
+def test_taskplan_scope_fields_round_trip_as_first_class_values() -> None:
+    plan = TaskPlan.from_dict(
+        {
+            "task_id": "task-scope-roundtrip",
+            "scope_type": "tenant",
+            "tenant_id": "tenant-a",
+            "user_id": "user-a",
+            "project_id": "project-a",
+            "conversation_id": "conversation-a",
+            "run_id": "run-a",
+        }
+    )
+    payload = plan.to_dict()
+    assert TaskPlan.from_dict(payload).to_dict() == payload
+
+
+def test_tenant_scope_requires_tenant_id_and_system_scope_does_not() -> None:
+    with pytest.raises(ValidationError) as exc:
+        TaskPlan.from_dict({"task_id": "task-missing-tenant", "scope_type": "tenant"})
+    assert exc.value.error_code == "TASK_SCOPE_TENANT_ID_REQUIRED"
+    system = TaskPlan.from_dict(
+        {"task_id": "task-system", "scope_type": "system", "run_id": "system-run"}
+    )
+    assert system.tenant_id is None
+
+
+def test_ACTOR_BINDING_AUTHZ_MERGE_uses_server_scope() -> None:
+    service = _service()
+    created = service.create_task(
+        {"task_id": "task-a", "task_type": "code.change"},
+        actor_identity="actor-a",
+    )
+    assert created["plan"]["tenant_id"] == "tenant-a"
+    assert created["plan"]["user_id"] == "user-a"
+    assert created["plan"]["project_id"] == "project-a"
+    declared = service.create_task(
+        {
+            "task_id": "task-a-declared",
+            "tenant_id": "tenant-a",
+            "user_id": "user-a",
+            "project_id": "project-a",
+            "conversation_id": "conversation-a",
+            "run_id": "run-a",
+        },
+        actor_identity="actor-a",
+    )
+    assert declared["plan"]["scope_type"] == "tenant"
+
+
+def test_TASK_SCOPE_MISMATCH_DENY_rejects_forged_client_scope() -> None:
+    service = _service()
+    with pytest.raises(ForbiddenError) as exc:
+        service.create_task(
+            {
+                "task_id": "task-forged",
+                "scope_type": "tenant",
+                "tenant_id": "tenant-b",
+                "user_id": "user-b",
+                "project_id": "project-b",
+            },
+            actor_identity="actor-a",
+        )
+    assert exc.value.error_code == "TASK_SCOPE_MISMATCH_DENY"
+
+
+def test_public_create_fails_closed_without_server_actor_scope() -> None:
+    service = _service()
+    with pytest.raises(ForbiddenError) as exc:
+        service.create_task(
+            {"task_id": "task-unbound"}, actor_identity="unbound"
+        )
+    assert exc.value.error_code == "TASK_ACTOR_SCOPE_UNAVAILABLE"
+
+
+def test_TASKPLANE_LIST_AUTHZ_DENY_and_TASK_OWNERSHIP_FILTER() -> None:
+    service = _service()
+    service.create_task({"task_id": "task-a"}, actor_identity="actor-a")
+    service.create_task({"task_id": "task-b"}, actor_identity="actor-b")
+    assert [item["task_id"] for item in service.list_tasks("reader-a")] == ["task-a"]
+    assert [item["task_id"] for item in service.list_tasks("reader-b")] == ["task-b"]
+
+
+def test_TASKPLANE_GET_AUTHZ_DENY_does_not_disclose_cross_tenant_task() -> None:
+    service = _service()
+    service.create_task({"task_id": "task-a"}, actor_identity="actor-a")
+    with pytest.raises(NotFoundError):
+        service.get_task("task-a", "reader-b")
+
+
+def test_http_authentication_drives_server_scope_and_cross_tenant_404() -> None:
+    service = _service()
+
+    class Handler(TaskPlaneRequestHandler):
+        pass
+
+    Handler.service = service
+    Handler.authenticator = lambda token: {
+        "token-a": "actor-a",
+        "token-b": "reader-b",
+    }.get(token)
+    server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    base = f"http://127.0.0.1:{server.server_port}"
+    try:
+        request = urllib.request.Request(
+            base + "/tasks",
+            data=json.dumps({"task_id": "task-http-a"}).encode(),
+            headers={
+                "Authorization": "Bearer token-a",
+                "Content-Type": "application/json",
+            },
+            method="POST",
+        )
+        with urllib.request.urlopen(request, timeout=3) as response:
+            created = json.loads(response.read())
+        assert created["plan"]["tenant_id"] == "tenant-a"
+
+        denied = urllib.request.Request(
+            base + "/tasks/task-http-a",
+            headers={"Authorization": "Bearer token-b"},
+        )
+        with pytest.raises(urllib.error.HTTPError) as exc:
+            urllib.request.urlopen(denied, timeout=3)
+        assert exc.value.code == 404
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join(timeout=3)
+
+
+def test_mutating_endpoint_is_scoped_before_cancel() -> None:
+    policy = PolicyEngine(
+        execution_enabled=True,
+        identity_capabilities={
+            "actor-a": {"task.creator", "task.read"},
+            "actor-b": {"task.cancel"},
+        },
+        identity_actor_contexts={
+            "actor-a": {"scope_type": "tenant", "tenant_id": "tenant-a"},
+            "actor-b": {"scope_type": "tenant", "tenant_id": "tenant-b"},
+        },
+    )
+    service = TaskPlaneService(TaskPlaneStore(":memory:"), policy=policy)
+    service.create_task({"task_id": "task-a"}, actor_identity="actor-a")
+    with pytest.raises(NotFoundError):
+        service.cancel("task-a", {"actor_identity": "actor-b"})
+
+
+def test_privileged_server_actor_can_cross_scope() -> None:
+    service = _service()
+    service.create_task({"task_id": "task-a"}, actor_identity="actor-a")
+    service.create_task({"task_id": "task-b"}, actor_identity="actor-b")
+    assert {item["task_id"] for item in service.list_tasks("owner")} == {
+        "task-a",
+        "task-b",
+    }
+
+
+def test_scope_sql_parameters_do_not_interpolate_tenant_values() -> None:
+    service = _service()
+    service.create_task({"task_id": "task-a"}, actor_identity="actor-a")
+    policy = PolicyEngine(
+        identity_capabilities={"attacker": {"task.read"}},
+        identity_actor_contexts={
+            "attacker": {
+                "scope_type": "tenant",
+                "tenant_id": "tenant-a' OR 1=1 --",
+            }
+        },
+    )
+    attacker_service = TaskPlaneService(service.store, policy=policy)
+    assert attacker_service.list_tasks("attacker") == []
+
+
+def test_existing_sqlite_rows_migrate_to_legacy_without_tenant_guess(tmp_path) -> None:
+    path = tmp_path / "legacy.sqlite3"
+    connection = sqlite3.connect(path)
+    connection.execute(
+        """
+        CREATE TABLE tasks (
+            task_id TEXT PRIMARY KEY, revision INTEGER NOT NULL, status TEXT NOT NULL,
+            task_json TEXT NOT NULL, task_type TEXT NOT NULL, root_task_id TEXT NOT NULL,
+            parent_task_id TEXT, correlation_id TEXT, idempotency_key TEXT UNIQUE,
+            created_at REAL NOT NULL, updated_at REAL NOT NULL, failure_reason TEXT
+        )
+        """
+    )
+    connection.execute(
+        "INSERT INTO tasks VALUES (?,?,?,?,?,?,?,?,?,?,?,?)",
+        (
+            "task-legacy",
+            1,
+            "CREATED",
+            json.dumps({"task_id": "task-legacy"}),
+            "code.change",
+            "task-legacy",
+            None,
+            None,
+            "legacy-key",
+            1.0,
+            1.0,
+            None,
+        ),
+    )
+    connection.commit()
+    connection.close()
+
+    store = TaskPlaneStore(str(path))
+    task = store.get_task("task-legacy")
+    assert task.task_plan.scope_type == "legacy"
+    assert task.task_plan.tenant_id is None
+    replay = store.create_task(
+        TaskPlan.from_dict({"task_id": "task-legacy"}),
+        idempotency_key="legacy-key",
+    )
+    assert replay.task_id == "task-legacy"
+    assert tuple(
+        store.conn.execute(
+            "SELECT scope_type,tenant_id FROM tasks WHERE task_id='task-legacy'"
+        ).fetchone()
+    ) == ("legacy", None)
+    tenant_context = _policy().actor_context("reader-a")
+    with pytest.raises(NotFoundError):
+        store.get_task("task-legacy", tenant_context)
+    store.close()