feat(online-agent): add owner-controlled same-binding execution rebind

active-runtime/app/aios_agent_task_execution_rebind.py

@@ -0,0 +1,722 @@
+"""Owner-controlled CAS-bound execution-envelope rebind for Online Agent Tasks.
+
+This service exists for a narrow recovery case: a Task was projected/claimed with a read-only
+source_capture envelope before its already-registered deterministic plan became authoritative.
+
+It never creates Tasks/Jobs/branches and never accepts arbitrary metadata patches. It only replaces
+metadata.execution + the derived execution_projection for the same governed binding, after an
+official NEEDS_REWORK transition has returned the Task to QUEUED with no active lease.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timezone
+import hashlib
+import hmac
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+import json
+import os
+from pathlib import Path
+import re
+import sqlite3
+from typing import Any, Callable, Mapping, Sequence
+import urllib.error
+import urllib.request
+
+from aios_online_agent_execution import (
+    GovernedExecutionBinding,
+    validate_execution_envelope,
+)
+
+TASK_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{2,179}$")
+KEY_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{7,159}$")
+SHA_RE = re.compile(r"^[0-9a-f]{64}$")
+MAX_BODY = 64 * 1024
+AUTHORIZATION_ID_ENV = "AIOS_AGENT_TASK_EXECUTION_REBIND_AUTHORIZATION_ID"
+
+
+class ExecutionRebindError(RuntimeError):
+    def __init__(self, code: str, status: int = 409) -> None:
+        super().__init__(code)
+        self.code = code
+        self.status = status
+
+
+def _canonical(value: object) -> bytes:
+    return json.dumps(
+        value, sort_keys=True, separators=(",", ":"), ensure_ascii=False
+    ).encode("utf-8")
+
+
+def _sha(value: object) -> str:
+    return hashlib.sha256(_canonical(value)).hexdigest()
+
+
+def _now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _require_string(
+    value: object,
+    code: str,
+    pattern: re.Pattern[str] | None = None,
+) -> str:
+    if not isinstance(value, str) or not value.strip():
+        raise ExecutionRebindError(code, 400)
+    value = value.strip()
+    if pattern is not None and pattern.fullmatch(value) is None:
+        raise ExecutionRebindError(code, 400)
+    return value
+
+
+def _required_authorization_id() -> str:
+    value = os.environ.get(AUTHORIZATION_ID_ENV, "").strip()
+    if not value:
+        raise RuntimeError("TASK_EXECUTION_REBIND_AUTHORIZATION_ID_MISSING")
+    if KEY_RE.fullmatch(value) is None:
+        raise RuntimeError("TASK_EXECUTION_REBIND_AUTHORIZATION_ID_INVALID")
+    return value
+
+
+@dataclass(frozen=True)
+class ExecutionRebindRequest:
+    expected_status: str
+    expected_revision: int
+    expected_attempt: int
+    previous_verification_sha256: str
+    expected_old_execution_digest: str
+    expected_new_execution_digest: str
+    authorization_id: str
+    idempotency_key: str
+    reason: str
+
+    @classmethod
+    def from_mapping(cls, value: Mapping[str, Any]) -> "ExecutionRebindRequest":
+        allowed = {
+            "expected_status",
+            "expected_revision",
+            "expected_attempt",
+            "previous_verification_sha256",
+            "expected_old_execution_digest",
+            "expected_new_execution_digest",
+            "authorization_id",
+            "idempotency_key",
+            "reason",
+        }
+        if set(value) != allowed:
+            raise ExecutionRebindError("EXECUTION_REBIND_FIELDS_INVALID", 400)
+
+        status = _require_string(value["expected_status"], "EXPECTED_STATUS_INVALID")
+        if status != "QUEUED":
+            raise ExecutionRebindError("EXPECTED_STATUS_INVALID", 400)
+
+        revision = value["expected_revision"]
+        attempt = value["expected_attempt"]
+        if (
+            not isinstance(revision, int)
+            or isinstance(revision, bool)
+            or revision < 0
+            or not isinstance(attempt, int)
+            or isinstance(attempt, bool)
+            or attempt < 1
+        ):
+            raise ExecutionRebindError("EXECUTION_REBIND_INTEGER_INVALID", 400)
+
+        verification = _require_string(
+            value["previous_verification_sha256"], "VERIFICATION_SHA256_INVALID"
+        )
+        old_digest = _require_string(
+            value["expected_old_execution_digest"], "OLD_EXECUTION_DIGEST_INVALID"
+        )
+        new_digest = _require_string(
+            value["expected_new_execution_digest"], "NEW_EXECUTION_DIGEST_INVALID"
+        )
+        for digest, code in (
+            (verification, "VERIFICATION_SHA256_INVALID"),
+            (old_digest, "OLD_EXECUTION_DIGEST_INVALID"),
+            (new_digest, "NEW_EXECUTION_DIGEST_INVALID"),
+        ):
+            if SHA_RE.fullmatch(digest) is None:
+                raise ExecutionRebindError(code, 400)
+        if old_digest == new_digest:
+            raise ExecutionRebindError("EXECUTION_REBIND_DIGEST_UNCHANGED", 400)
+
+        authorization = _require_string(
+            value["authorization_id"], "AUTHORIZATION_ID_INVALID", KEY_RE
+        )
+        key = _require_string(value["idempotency_key"], "IDEMPOTENCY_KEY_INVALID", KEY_RE)
+        reason = _require_string(value["reason"], "EXECUTION_REBIND_REASON_INVALID")
+        if len(reason) > 512:
+            raise ExecutionRebindError("EXECUTION_REBIND_REASON_INVALID", 400)
+
+        return cls(
+            status,
+            revision,
+            attempt,
+            verification,
+            old_digest,
+            new_digest,
+            authorization,
+            key,
+            reason,
+        )
+
+    def public(self) -> dict[str, Any]:
+        return {
+            "expected_status": self.expected_status,
+            "expected_revision": self.expected_revision,
+            "expected_attempt": self.expected_attempt,
+            "previous_verification_sha256": self.previous_verification_sha256,
+            "expected_old_execution_digest": self.expected_old_execution_digest,
+            "expected_new_execution_digest": self.expected_new_execution_digest,
+            "authorization_id": self.authorization_id,
+            "idempotency_key": self.idempotency_key,
+            "reason": self.reason,
+        }
+
+
+class CanonicalBridgeExecutionResolver:
+    """Resolve the already-registered plan from the canonical governed Bridge."""
+
+    def __init__(self, base_url: str, control_token: str) -> None:
+        self.base_url = base_url.rstrip("/")
+        self.control_token = control_token
+
+    def __call__(
+        self,
+        task_id: str,
+        required_capabilities: Sequence[str],
+    ) -> dict[str, Any]:
+        body = json.dumps(
+            {
+                "task_id": task_id,
+                "required_capabilities": list(required_capabilities),
+            },
+            sort_keys=True,
+            separators=(",", ":"),
+        ).encode("utf-8")
+        request = urllib.request.Request(
+            self.base_url + "/v2/execution-envelope",
+            method="POST",
+            data=body,
+            headers={
+                "Authorization": "Bearer " + self.control_token,
+                "Content-Type": "application/json",
+                "Accept": "application/json",
+            },
+        )
+        try:
+            with urllib.request.urlopen(request, timeout=15) as response:
+                raw = response.read()
+                status = response.status
+        except urllib.error.HTTPError as exc:
+            raw = exc.read()
+            status = exc.code
+        try:
+            value = json.loads(raw.decode("utf-8"))
+        except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+            raise ExecutionRebindError("CANONICAL_EXECUTION_RESPONSE_INVALID", 502) from exc
+        if not isinstance(value, Mapping):
+            raise ExecutionRebindError("CANONICAL_EXECUTION_RESPONSE_INVALID", 502)
+        if status != 200:
+            raise ExecutionRebindError(
+                str(value.get("error") or "CANONICAL_EXECUTION_RESOLUTION_FAILED"),
+                502 if status >= 500 else 409,
+            )
+        return dict(value)
+
+
+class ExecutionRebindStore:
+    """Official owner-controlled mutation with CAS, plan resolution and durable audit."""
+
+    def __init__(
+        self,
+        db_path: Path,
+        resolver: Callable[[str, Sequence[str]], Mapping[str, Any]],
+    ) -> None:
+        self.db_path = Path(db_path)
+        self.resolver = resolver
+
+    def _connect(self) -> sqlite3.Connection:
+        connection = sqlite3.connect(self.db_path, timeout=30.0)
+        connection.row_factory = sqlite3.Row
+        connection.execute("PRAGMA foreign_keys=ON")
+        connection.execute("PRAGMA busy_timeout=30000")
+        return connection
+
+    @staticmethod
+    def _columns(
+        connection: sqlite3.Connection,
+        table: str,
+    ) -> dict[str, sqlite3.Row]:
+        return {
+            str(row["name"]): row
+            for row in connection.execute(f'PRAGMA table_info("{table}")')
+        }
+
+    @staticmethod
+    def _json_value(raw: object, code: str) -> dict[str, Any]:
+        if isinstance(raw, Mapping):
+            return dict(raw)
+        if not isinstance(raw, str):
+            raise ExecutionRebindError(code, 409)
+        try:
+            value = json.loads(raw)
+        except json.JSONDecodeError as exc:
+            raise ExecutionRebindError(code, 409) from exc
+        if not isinstance(value, Mapping):
+            raise ExecutionRebindError(code, 409)
+        return dict(value)
+
+    @staticmethod
+    def _verification_hash(raw: object) -> str:
+        if raw in (None, ""):
+            value: object = {}
+        elif isinstance(raw, str):
+            try:
+                value = json.loads(raw)
+            except json.JSONDecodeError as exc:
+                raise ExecutionRebindError("TASK_VERIFICATION_INVALID", 409) from exc
+        else:
+            value = raw
+        return _sha(value)
+
+    @staticmethod
+    def _metadata_column(columns: Mapping[str, sqlite3.Row]) -> str:
+        for name in ("metadata_json", "metadata"):
+            if name in columns:
+                return name
+        raise ExecutionRebindError("TASK_METADATA_SCHEMA_UNSUPPORTED", 503)
+
+    @staticmethod
+    def _required_capabilities(
+        task: sqlite3.Row,
+        columns: Mapping[str, sqlite3.Row],
+        old_execution: Mapping[str, Any],
+    ) -> list[str]:
+        for field in ("required_capabilities_json", "required_capabilities"):
+            if field not in columns:
+                continue
+            raw = task[field]
+            if raw in (None, ""):
+                continue
+            if isinstance(raw, str):
+                try:
+                    value = json.loads(raw)
+                except json.JSONDecodeError:
+                    value = [item for item in raw.split(",") if item.strip()]
+            else:
+                value = raw
+            if isinstance(value, list) and all(isinstance(item, str) for item in value):
+                return [item.strip() for item in value if item.strip()]
+        value = old_execution.get("capability_scope")
+        if isinstance(value, list) and all(isinstance(item, str) for item in value):
+            return [item for item in value if item]
+        raise ExecutionRebindError("TASK_CAPABILITIES_UNAVAILABLE", 503)
+
+    @staticmethod
+    def _event_table(
+        connection: sqlite3.Connection,
+    ) -> tuple[str, dict[str, sqlite3.Row]]:
+        tables = {
+            row["name"]
+            for row in connection.execute(
+                "SELECT name FROM sqlite_master WHERE type='table'"
+            )
+        }
+        for name in ("task_events", "events"):
+            if name not in tables:
+                continue
+            columns = ExecutionRebindStore._columns(connection, name)
+            if "task_id" in columns and ("event_type" in columns or "type" in columns):
+                return name, columns
+        raise ExecutionRebindError("TASK_EVENT_TABLE_UNSUPPORTED", 503)
+
+    @staticmethod
+    def _insert_event(
+        connection: sqlite3.Connection,
+        task_id: str,
+        payload: Mapping[str, Any],
+    ) -> None:
+        table, columns = ExecutionRebindStore._event_table(connection)
+        values: dict[str, Any] = {"task_id": task_id}
+        if "event_type" in columns:
+            values["event_type"] = "TASK_EXECUTION_REBIND_AUTHORIZED"
+        else:
+            values["type"] = "TASK_EXECUTION_REBIND_AUTHORIZED"
+        payload_text = json.dumps(payload, sort_keys=True, separators=(",", ":"))
+        if "payload_json" in columns:
+            values["payload_json"] = payload_text
+        elif "payload" in columns:
+            values["payload"] = payload_text
+        else:
+            raise ExecutionRebindError("TASK_EVENT_PAYLOAD_UNSUPPORTED", 503)
+        if "created_at" in columns:
+            values["created_at"] = _now()
+        if "agent_id" in columns:
+            values["agent_id"] = "owner-control"
+        if "actor_id" in columns:
+            values["actor_id"] = "owner-control"
+        names = list(values)
+        quoted = ",".join('"' + name + '"' for name in names)
+        placeholders = ",".join("?" for _ in names)
+        connection.execute(
+            f'INSERT INTO "{table}" ({quoted}) VALUES ({placeholders})',
+            tuple(values[name] for name in names),
+        )
+
+    @staticmethod
+    def _ensure_rebind_table(connection: sqlite3.Connection) -> None:
+        connection.execute(
+            """
+            CREATE TABLE IF NOT EXISTS task_execution_rebinds (
+              task_id TEXT NOT NULL,
+              idempotency_key TEXT NOT NULL,
+              request_sha256 TEXT NOT NULL,
+              authorization_id TEXT NOT NULL,
+              previous_verification_sha256 TEXT NOT NULL,
+              old_execution_digest TEXT NOT NULL,
+              new_execution_digest TEXT NOT NULL,
+              expected_revision INTEGER NOT NULL,
+              applied_revision INTEGER NOT NULL,
+              attempt INTEGER NOT NULL,
+              reason TEXT NOT NULL,
+              created_at TEXT NOT NULL,
+              PRIMARY KEY(task_id,idempotency_key)
+            )
+            """
+        )
+
+    def _load_and_validate(
+        self,
+        connection: sqlite3.Connection,
+        task_id: str,
+        request: ExecutionRebindRequest,
+    ) -> tuple[sqlite3.Row, dict[str, sqlite3.Row], str, dict[str, Any], dict[str, Any]]:
+        columns = self._columns(connection, "tasks")
+        required = {
+            "task_id",
+            "status",
+            "attempt",
+            "revision",
+            "claimed_by",
+            "lease_token_hash",
+            "lease_expires_at",
+            "verification_json",
+        }
+        if not required.issubset(columns):
+            raise ExecutionRebindError("TASK_SCHEMA_EXECUTION_REBIND_UNSUPPORTED", 503)
+        metadata_column = self._metadata_column(columns)
+        task = connection.execute(
+            "SELECT * FROM tasks WHERE task_id=?",
+            (task_id,),
+        ).fetchone()
+        if task is None:
+            raise ExecutionRebindError("TASK_NOT_FOUND", 404)
+
+        if (
+            str(task["status"]) != request.expected_status
+            or int(task["revision"]) != request.expected_revision
+            or int(task["attempt"]) != request.expected_attempt
+        ):
+            raise ExecutionRebindError("EXECUTION_REBIND_CAS_MISMATCH", 409)
+        if task["lease_token_hash"] not in (None, "") or task["lease_expires_at"] not in (None, ""):
+            raise ExecutionRebindError("EXECUTION_REBIND_ACTIVE_LEASE_DENY", 409)
+
+        verification_sha = self._verification_hash(task["verification_json"])
+        if verification_sha != request.previous_verification_sha256:
+            raise ExecutionRebindError("EXECUTION_REBIND_VERIFICATION_DRIFT", 409)
+
+        metadata = self._json_value(
+            task[metadata_column],
+            "TASK_METADATA_INVALID",
+        )
+        execution = metadata.get("execution")
+        if not isinstance(execution, Mapping):
+            raise ExecutionRebindError("TASK_EXECUTION_METADATA_MISSING", 409)
+        old_execution = validate_execution_envelope(dict(execution))
+        if old_execution["execution_digest"] != request.expected_old_execution_digest:
+            raise ExecutionRebindError("EXECUTION_REBIND_OLD_DIGEST_DRIFT", 409)
+        if old_execution.get("action") != "source_capture":
+            raise ExecutionRebindError("EXECUTION_REBIND_SOURCE_CAPTURE_REQUIRED", 409)
+
+        projection = metadata.get("execution_projection")
+        if isinstance(projection, Mapping):
+            projected_digest = str(projection.get("execution_digest") or "")
+            if projected_digest and projected_digest != request.expected_old_execution_digest:
+                raise ExecutionRebindError("EXECUTION_REBIND_PROJECTION_DRIFT", 409)
+        return task, columns, metadata_column, metadata, old_execution
+
+    def rebind(
+        self,
+        task_id: str,
+        request: ExecutionRebindRequest,
+    ) -> dict[str, Any]:
+        task_id = _require_string(task_id, "TASK_ID_INVALID", TASK_RE)
+        required_authorization = _required_authorization_id()
+        if request.authorization_id != required_authorization:
+            raise ExecutionRebindError("EXECUTION_REBIND_AUTHORIZATION_ID_MISMATCH", 403)
+
+        request_sha = _sha(request.public())
+
+        # Idempotent replay must win before CAS validation because a successful
+        # rebind necessarily increments the Task revision and changes the persisted envelope.
+        with self._connect() as connection:
+            self._ensure_rebind_table(connection)
+            replay = connection.execute(
+                "SELECT * FROM task_execution_rebinds WHERE task_id=? AND idempotency_key=?",
+                (task_id, request.idempotency_key),
+            ).fetchone()
+            if replay is not None:
+                if replay["request_sha256"] != request_sha:
+                    raise ExecutionRebindError("EXECUTION_REBIND_IDEMPOTENCY_CONFLICT", 409)
+                connection.commit()
+                return {
+                    "task_id": task_id,
+                    "status": "QUEUED",
+                    "attempt": int(replay["attempt"]),
+                    "revision": int(replay["applied_revision"]),
+                    "old_execution_digest": replay["old_execution_digest"],
+                    "new_execution_digest": replay["new_execution_digest"],
+                    "authorization_id": replay["authorization_id"],
+                    "idempotent": True,
+                }
+
+        # Read/CAS snapshot first. Canonical Bridge resolution happens outside the DB write lock.
+        with self._connect() as connection:
+            task, columns, _, _, old_execution = self._load_and_validate(
+                connection, task_id, request
+            )
+            capabilities = self._required_capabilities(task, columns, old_execution)
+
+        binding = GovernedExecutionBinding.from_mapping(old_execution)
+        resolved = dict(self.resolver(task_id, capabilities))
+        new_execution = validate_execution_envelope(resolved, binding=binding)
+        if new_execution.get("action") != "deterministic":
+            raise ExecutionRebindError("EXECUTION_REBIND_DETERMINISTIC_PLAN_REQUIRED", 409)
+        if new_execution["execution_digest"] != request.expected_new_execution_digest:
+            raise ExecutionRebindError("EXECUTION_REBIND_NEW_DIGEST_DRIFT", 409)
+        if new_execution["execution_digest"] == old_execution["execution_digest"]:
+            raise ExecutionRebindError("EXECUTION_REBIND_DIGEST_UNCHANGED", 409)
+
+        # Re-read under BEGIN IMMEDIATE and apply only if the exact snapshot is unchanged.
+        with self._connect() as connection:
+            connection.execute("BEGIN IMMEDIATE")
+            self._ensure_rebind_table(connection)
+
+            replay = connection.execute(
+                "SELECT * FROM task_execution_rebinds WHERE task_id=? AND idempotency_key=?",
+                (task_id, request.idempotency_key),
+            ).fetchone()
+            if replay is not None:
+                if replay["request_sha256"] != request_sha:
+                    raise ExecutionRebindError("EXECUTION_REBIND_IDEMPOTENCY_CONFLICT", 409)
+                connection.commit()
+                return {
+                    "task_id": task_id,
+                    "status": "QUEUED",
+                    "attempt": int(replay["attempt"]),
+                    "revision": int(replay["applied_revision"]),
+                    "old_execution_digest": replay["old_execution_digest"],
+                    "new_execution_digest": replay["new_execution_digest"],
+                    "authorization_id": replay["authorization_id"],
+                    "idempotent": True,
+                }
+
+            task, columns, metadata_column, metadata, current_old = self._load_and_validate(
+                connection, task_id, request
+            )
+            if current_old["execution_digest"] != old_execution["execution_digest"]:
+                raise ExecutionRebindError("EXECUTION_REBIND_OLD_DIGEST_DRIFT", 409)
+
+            new_metadata = dict(metadata)
+            new_metadata["execution"] = new_execution
+            new_metadata["execution_projection"] = {
+                "schema": "aios.online-agent.task-projection.v1",
+                "task_id": binding.task_id,
+                "job_id": binding.job_id,
+                "execution_digest": new_execution["execution_digest"],
+            }
+            applied_revision = request.expected_revision + 1
+            previous_claimed_by = task["claimed_by"]
+
+            cursor = connection.execute(
+                f"""
+                UPDATE tasks
+                   SET "{metadata_column}"=?, revision=?, claimed_by=NULL,
+                       lease_token_hash=NULL, lease_expires_at=NULL
+                 WHERE task_id=? AND status=? AND revision=? AND attempt=?
+                   AND claimed_by IS ? AND lease_token_hash IS NULL AND lease_expires_at IS NULL
+                """,
+                (
+                    json.dumps(new_metadata, sort_keys=True, separators=(",", ":")),
+                    applied_revision,
+                    task_id,
+                    request.expected_status,
+                    request.expected_revision,
+                    request.expected_attempt,
+                    previous_claimed_by,
+                ),
+            )
+            if cursor.rowcount != 1:
+                raise ExecutionRebindError("EXECUTION_REBIND_CAS_MISMATCH", 409)
+
+            event_payload = {
+                "schema": "aios.task-execution-rebind.event.v1",
+                "task_id": task_id,
+                "job_id": binding.job_id,
+                "authorization_id": request.authorization_id,
+                "idempotency_key": request.idempotency_key,
+                "reason": request.reason,
+                "previous_verification_sha256": request.previous_verification_sha256,
+                "old_execution_digest": old_execution["execution_digest"],
+                "new_execution_digest": new_execution["execution_digest"],
+                "old_action": old_execution.get("action"),
+                "new_action": new_execution.get("action"),
+                "previous_revision": request.expected_revision,
+                "applied_revision": applied_revision,
+                "attempt": request.expected_attempt,
+                "request_sha256": request_sha,
+                "previous_claimed_by": previous_claimed_by,
+                "stale_claim_normalized": previous_claimed_by not in (None, ""),
+            }
+            self._insert_event(connection, task_id, event_payload)
+            connection.execute(
+                """
+                INSERT INTO task_execution_rebinds(
+                  task_id,idempotency_key,request_sha256,authorization_id,
+                  previous_verification_sha256,old_execution_digest,new_execution_digest,
+                  expected_revision,applied_revision,attempt,reason,created_at
+                ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)
+                """,
+                (
+                    task_id,
+                    request.idempotency_key,
+                    request_sha,
+                    request.authorization_id,
+                    request.previous_verification_sha256,
+                    old_execution["execution_digest"],
+                    new_execution["execution_digest"],
+                    request.expected_revision,
+                    applied_revision,
+                    request.expected_attempt,
+                    request.reason,
+                    _now(),
+                ),
+            )
+            connection.commit()
+            return {
+                "task_id": task_id,
+                "status": "QUEUED",
+                "attempt": request.expected_attempt,
+                "revision": applied_revision,
+                "old_execution_digest": old_execution["execution_digest"],
+                "new_execution_digest": new_execution["execution_digest"],
+                "authorization_id": request.authorization_id,
+                "idempotent": False,
+            }
+
+
+class ExecutionRebindHandler(BaseHTTPRequestHandler):
+    store: ExecutionRebindStore | None = None
+    control_token: str = ""
+
+    def log_message(self, *_args: Any) -> None:
+        return
+
+    def _json(self, value: Mapping[str, Any], status: int) -> None:
+        data = json.dumps(
+            dict(value), sort_keys=True, separators=(",", ":")
+        ).encode("utf-8")
+        self.send_response(status)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(data)))
+        self.end_headers()
+        self.wfile.write(data)
+
+    def _authorized(self) -> bool:
+        header = self.headers.get("Authorization", "")
+        return (
+            header.startswith("Bearer ")
+            and bool(self.control_token)
+            and hmac.compare_digest(header[7:], self.control_token)
+        )
+
+    def do_GET(self) -> None:
+        if self.path == "/health":
+            self._json(
+                {"status": "ok", "service": "aios-agent-task-execution-rebind"},
+                200,
+            )
+        else:
+            self._json({"error": "NOT_FOUND"}, 404)
+
+    def do_POST(self) -> None:
+        match = re.fullmatch(r"/v1/tasks/([^/]+)/execution-rebind", self.path)
+        if match is None:
+            self._json({"error": "NOT_FOUND"}, 404)
+            return
+        if not self._authorized():
+            self._json({"error": "UNAUTHORIZED"}, 401)
+            return
+        try:
+            length = int(self.headers.get("Content-Length", "0"))
+        except ValueError:
+            self._json({"error": "CONTENT_LENGTH_INVALID"}, 400)
+            return
+        if length < 1 or length > MAX_BODY:
+            self._json({"error": "REQUEST_SIZE_INVALID"}, 413)
+            return
+        try:
+            value = json.loads(self.rfile.read(length).decode("utf-8"))
+            if not isinstance(value, Mapping):
+                raise ValueError
+            request = ExecutionRebindRequest.from_mapping(value)
+            if self.store is None:
+                raise ExecutionRebindError("RUNTIME_UNAVAILABLE", 503)
+            result = self.store.rebind(match.group(1), request)
+            self._json(result, 200 if result["idempotent"] else 201)
+        except ExecutionRebindError as exc:
+            self._json({"error": exc.code}, exc.status)
+        except (UnicodeDecodeError, json.JSONDecodeError, ValueError):
+            self._json({"error": "REQUEST_INVALID"}, 400)
+
+
+def load_control_token(path: Path) -> str:
+    if path.is_symlink() or not path.is_file():
+        raise RuntimeError("TASK_EXECUTION_REBIND_CONTROL_CREDENTIAL_UNAVAILABLE")
+    value = path.read_text(encoding="utf-8").strip()
+    if len(value) < 32:
+        raise RuntimeError("TASK_EXECUTION_REBIND_CONTROL_CREDENTIAL_INVALID")
+    return value
+
+
+def serve() -> None:
+    db = Path(os.environ.get("AIOS_AGENT_TASK_DB", "/var/lib/aios-agent-task/state.sqlite3"))
+    _required_authorization_id()
+    token_path = Path(
+        os.environ.get(
+            "AIOS_AGENT_TASK_CONTROL_TOKEN_FILE",
+            str(
+                Path(os.environ.get("CREDENTIALS_DIRECTORY", "/run/credentials"))
+                / "agent_task_control_token"
+            ),
+        )
+    )
+    token = load_control_token(token_path)
+    bridge = os.environ.get(
+        "AIOS_GOVERNED_WRITE_API_URL",
+        "http://127.0.0.1:18892",
+    )
+    host = os.environ.get("AIOS_TASK_EXECUTION_REBIND_HOST", "127.0.0.1")
+    port = int(os.environ.get("AIOS_TASK_EXECUTION_REBIND_PORT", "18942"))
+    ExecutionRebindHandler.store = ExecutionRebindStore(
+        db,
+        CanonicalBridgeExecutionResolver(bridge, token),
+    )
+    ExecutionRebindHandler.control_token = token
+    ThreadingHTTPServer((host, port), ExecutionRebindHandler).serve_forever()
+
+
+if __name__ == "__main__":
+    serve()

active-runtime/app/aios_online_agent_compact_map.py

@@ -42,7 +42,7 @@ DEFAULT_TERMS = {
     "backoff", "circuit", "health", "liveness", "scope", "tenant", "project", "trace",
     "message", "checkpoint", "recovery", "idempotency", "dispatch", "assignment",
 }
-CONTRACT_TERMS: dict[str, tuple[str, ...]] = {
+FLEET_CONTRACT_TERMS: dict[str, tuple[str, ...]] = {
     "FLEET_AGENT_REGISTRATION": ("agent", "register", "worker", "capability", "version"),
     "FLEET_CAPACITY_SCHEDULING": ("capacity", "slot", "concurrency", "scheduler", "queue"),
     "FLEET_FAIRNESS": ("fair", "priority", "queue", "tenant", "project", "starvation"),
@@ -52,6 +52,30 @@ CONTRACT_TERMS: dict[str, tuple[str, ...]] = {
     "FLEET_IDEMPOTENT_DISPATCH": ("idempot", "dispatch", "claim", "lease", "result", "attempt"),
     "FLEET_FAILURE_RECOVERY": ("checkpoint", "resume", "recover", "retry", "cleanup", "receipt"),
 }
+
+MODEL_GOVERNANCE_CONTRACT_TERMS: dict[str, tuple[str, ...]] = {
+    "MODEL_PROVIDER_REGISTRY": ("model", "provider", "registry", "endpoint", "capability"),
+    "MODEL_SELECTION_POLICY": ("model", "select", "route", "policy", "capability"),
+    "MODEL_VERSION_PINNING": ("model", "version", "pin", "revision", "snapshot"),
+    "MODEL_BUDGET_ACCOUNTING": ("budget", "token", "cost", "quota", "usage"),
+    "MODEL_FALLBACK_POLICY": ("fallback", "retry", "route", "provider", "model"),
+    "MODEL_SCOPE_ISOLATION": ("tenant", "project", "scope", "policy", "model"),
+    "MODEL_CREDENTIAL_BOUNDARY": ("credential", "secret", "provider", "tool", "token"),
+    "MODEL_TRACE_AUDIT": ("trace", "audit", "model", "provider", "request"),
+    "MODEL_EVALUATION_GATE": ("eval", "evaluation", "approval", "quality", "gate"),
+    "MODEL_ROUTING_RESILIENCE": ("route", "retry", "fallback", "timeout", "health"),
+}
+
+PROFILE_CLASSIFICATION = {
+    "fleet": "P2_INTERFACE_MAP_CAPTURED",
+    "model_governance": "MODEL_GOVERNANCE_INTERFACE_MAP_CAPTURED",
+}
+
+MODEL_PROFILE_MARKERS = {
+    "model", "provider", "llm", "inference", "routing", "router", "budget",
+    "token", "cost", "quota", "fallback", "evaluation", "eval", "approval",
+    "version", "pinning",
+}
 SENSITIVE_RE = re.compile(
     r"(?i)(authorization\s*:\s*bearer\s+\S+|"
     r"(?:password|passwd|secret|token|api[_-]?key)\s*[:=]\s*['\"]?[^'\"\s,;]+)"
@@ -92,6 +116,27 @@ def _signal_terms(signals: Sequence[str]) -> set[str]:
     return set(sorted(terms)[:96])
 
 
+def _profile(source_signals: Sequence[str]) -> str:
+    tokens: set[str] = set()
+    for signal in source_signals:
+        if not isinstance(signal, str):
+            continue
+        tokens.update(
+            token.lower()
+            for token in re.findall(r"[A-Za-z][A-Za-z0-9_.-]{2,63}", signal)
+        )
+    model_score = len(tokens & MODEL_PROFILE_MARKERS)
+    return "model_governance" if model_score >= 2 else "fleet"
+
+
+def _contract_terms(profile: str) -> Mapping[str, tuple[str, ...]]:
+    if profile == "model_governance":
+        return MODEL_GOVERNANCE_CONTRACT_TERMS
+    if profile == "fleet":
+        return FLEET_CONTRACT_TERMS
+    raise CompactInterfaceMapError("INTERFACE_MAP_PROFILE_INVALID")
+
+
 def _candidate_files(root: Path, terms: set[str]) -> list[dict[str, Any]]:
     root = root.resolve()
     candidates: list[dict[str, Any]] = []
@@ -190,10 +235,11 @@ def _surface_ref(symbol: Mapping[str, Any]) -> str:
 def _contract_map(
     symbols: Sequence[Mapping[str, Any]],
     modules: Sequence[Mapping[str, Any]],
+    contract_terms: Mapping[str, tuple[str, ...]],
 ) -> dict[str, Any]:
     result: dict[str, Any] = {}
     module_paths = [str(item["path"]) for item in modules]
-    for contract, terms in CONTRACT_TERMS.items():
+    for contract, terms in contract_terms.items():
         scored: list[tuple[int, str]] = []
         for symbol in symbols:
             haystack = _surface_ref(symbol).lower()
@@ -216,37 +262,67 @@ def _contract_map(
     return result
 
 
-def _gap_evidence(symbols: Sequence[Mapping[str, Any]]) -> tuple[list[str], list[str], list[str]]:
+def _gap_evidence(
+    symbols: Sequence[Mapping[str, Any]],
+    profile: str,
+) -> tuple[list[str], list[str], list[str]]:
     haystack = "\n".join(_surface_ref(item).lower() for item in symbols)
 
-    invariants = [
-        "Online Task claim/heartbeat remains the authoritative worker ownership boundary.",
-        "Governed mutation remains separated from read-only analysis and requires WriterLease.",
-        "P1 attempt-aware result idempotency and durable checkpoint recovery are reusable fleet primitives.",
-        "Scope/trace/message lineage must remain attached when work is scheduled or reassigned.",
-    ]
-    failure_windows: list[str] = []
-    if "scheduler" not in haystack:
-        failure_windows.append("No first-class durable fleet scheduler symbol detected; worker-local polling may still be the scheduling mechanism.")
-    if "capacity" not in haystack and "slot" not in haystack:
-        failure_windows.append("No explicit durable worker-capacity/slot primitive detected.")
-    if "fair" not in haystack and "starvation" not in haystack:
-        failure_windows.append("No explicit fairness/starvation policy symbol detected.")
-    if "circuit" not in haystack:
-        failure_windows.append("No explicit circuit-breaker primitive detected for repeated worker/task failures.")
-    if "reassign" not in haystack:
-        failure_windows.append("No explicit worker-reassignment primitive detected beyond lease expiry/retry semantics.")
+    if profile == "model_governance":
+        invariants = [
+            "Online Task claim/heartbeat remains the authoritative worker ownership boundary.",
+            "Provider/model credentials remain outside uncredentialed planner/executor processes.",
+            "P0-A tenant/project scope and P0-C trace/audit lineage must constrain model routing.",
+            "P1 retry/checkpoint semantics remain authoritative for model-call execution recovery.",
+        ]
+        failure_windows: list[str] = []
+        if "provider" not in haystack and "model" not in haystack:
+            failure_windows.append("No first-class model/provider registry surface detected.")
+        if "budget" not in haystack and "cost" not in haystack and "quota" not in haystack:
+            failure_windows.append("No explicit durable model budget/cost/quota accounting surface detected.")
+        if "fallback" not in haystack and "route" not in haystack:
+            failure_windows.append("No explicit model fallback/routing policy surface detected.")
+        if "eval" not in haystack and "approval" not in haystack:
+            failure_windows.append("No explicit evaluation/approval gate surface detected for model changes.")
+        if "version" not in haystack and "pin" not in haystack:
+            failure_windows.append("No explicit model-version pinning surface detected.")
+        desired = (
+            ("trace", "P0-C trace/audit lineage"),
+            ("scope", "P0-A tenant/project scope isolation"),
+            ("credential", "P0-C credential governance"),
+            ("checkpoint", "P1 durable execution recovery"),
+            ("message", "P0-B MessageEnvelope/handoff"),
+            ("retry", "P1 retry/idempotency"),
+        )
+    else:
+        invariants = [
+            "Online Task claim/heartbeat remains the authoritative worker ownership boundary.",
+            "Governed mutation remains separated from read-only analysis and requires WriterLease.",
+            "P1 attempt-aware result idempotency and durable checkpoint recovery are reusable fleet primitives.",
+            "Scope/trace/message lineage must remain attached when work is scheduled or reassigned.",
+        ]
+        failure_windows = []
+        if "scheduler" not in haystack:
+            failure_windows.append("No first-class durable fleet scheduler symbol detected; worker-local polling may still be the scheduling mechanism.")
+        if "capacity" not in haystack and "slot" not in haystack:
+            failure_windows.append("No explicit durable worker-capacity/slot primitive detected.")
+        if "fair" not in haystack and "starvation" not in haystack:
+            failure_windows.append("No explicit fairness/starvation policy symbol detected.")
+        if "circuit" not in haystack:
+            failure_windows.append("No explicit circuit-breaker primitive detected for repeated worker/task failures.")
+        if "reassign" not in haystack:
+            failure_windows.append("No explicit worker-reassignment primitive detected beyond lease expiry/retry semantics.")
+        desired = (
+            ("checkpoint", "P1 durable checkpoint/resume"),
+            ("message", "P0-B MessageEnvelope/handoff"),
+            ("trace", "P0-C trace correlation"),
+            ("scope", "P0-A scope isolation"),
+            ("lease", "claim/WriterLease ownership"),
+            ("heartbeat", "worker liveness"),
+            ("idempot", "retry-safe dispatch/result"),
+        )
 
     reuse_points: list[str] = []
-    desired = (
-        ("checkpoint", "P1 durable checkpoint/resume"),
-        ("message", "P0-B MessageEnvelope/handoff"),
-        ("trace", "P0-C trace correlation"),
-        ("scope", "P0-A scope isolation"),
-        ("lease", "claim/WriterLease ownership"),
-        ("heartbeat", "worker liveness"),
-        ("idempot", "retry-safe dispatch/result"),
-    )
     for token, label in desired:
         match = next((item for item in symbols if token in _surface_ref(item).lower()), None)
         if match is not None:
@@ -287,6 +363,7 @@ def build_compact_interface_map(
     if re.fullmatch(r"[0-9a-f]{40}", source_commit) is None:
         raise CompactInterfaceMapError("INTERFACE_MAP_SOURCE_COMMIT_INVALID")
 
+    profile = _profile(source_signals)
     terms = _signal_terms(source_signals)
     candidates = _candidate_files(root, terms)
     if not candidates:
@@ -363,9 +440,11 @@ def build_compact_interface_map(
             }
         )
 
-    invariants, failure_windows, reuse_points = _gap_evidence(all_symbols)
+    invariants, failure_windows, reuse_points = _gap_evidence(all_symbols, profile)
     result = {
         "schema": SCHEMA,
+        "profile": profile,
+        "classification": PROFILE_CLASSIFICATION[profile],
         "source_commit": source_commit,
         "implementation_map": {
             "modules": modules,
@@ -379,6 +458,6 @@ def build_compact_interface_map(
             "reuse_points": reuse_points,
         },
         "bounded_excerpts": excerpts,
-        "contracts": _contract_map(all_symbols, modules),
+        "contracts": _contract_map(all_symbols, modules, _contract_terms(profile)),
     }
     return _trim_to_budget(result)

active-runtime/app/aios_online_agent_worker.py

@@ -1526,9 +1526,18 @@ def build_compact_interface_map_result(
     contracts = interface_map.get("contracts")
     if not isinstance(implementation_map, Mapping) or not isinstance(excerpts, list) or not isinstance(contracts, Mapping):
         raise WorkerError("COMPACT_INTERFACE_MAP_RESULT_INVALID", phase="result")
+    profile = str(interface_map.get("profile") or "fleet")
+    classification = str(interface_map.get("classification") or "")
+    allowed_classifications = {
+        "fleet": "P2_INTERFACE_MAP_CAPTURED",
+        "model_governance": "MODEL_GOVERNANCE_INTERFACE_MAP_CAPTURED",
+    }
+    if allowed_classifications.get(profile) != classification:
+        raise WorkerError("COMPACT_INTERFACE_MAP_PROFILE_BINDING_INVALID", phase="result")
     result = {
         "status": "SUCCEEDED",
-        "classification": "P2_INTERFACE_MAP_CAPTURED",
+        "classification": classification,
+        "interface_map_profile": profile,
         "verification_state": "VERIFIED_BY_WORKER_PENDING_SERVER_REVIEW",
         "execution_mode": "compact_interface_map",
         "execution_backend": "generic-online-agent-worker-v2",

active-runtime/app/task_plane/fleet.py

@@ -363,7 +363,7 @@ class FleetStateStore:
             row = self.conn.execute(
                 """
                 SELECT COUNT(*) FROM fleet_dispatches
-                 WHERE worker_id=? AND state IN ('RESERVED','CLAIMED') AND reserved_until>?
+                 WHERE worker_id=? AND state='RESERVED' AND reserved_until>?
                 """,
                 (worker_id, now),
             ).fetchone()
@@ -412,6 +412,13 @@ class FleetStateStore:
                 """,
                 (task_id,),
             ).fetchone()
+            # Capacity is rechecked inside the same durable reservation transaction.
+            # This closes the cross-thread/process race between eligibility scoring and reserve.
+            policy_row = self.conn.execute(
+                "SELECT max_slots FROM fleet_worker_capacity WHERE worker_id=?",
+                (worker_id,),
+            ).fetchone()
+            max_slots = int(policy_row[0]) if policy_row else 1
             if existing and str(existing[6]) in {"RESERVED", "CLAIMED"} and float(existing[5]) > now:
                 if str(existing[1]) != worker_id or int(existing[2]) != task_revision:
                     raise FleetConflict("FLEET_TASK_ALREADY_RESERVED")
@@ -419,6 +426,32 @@ class FleetStateStore:
                     str(existing[0]), task_id, worker_id, task_revision, str(existing[3]),
                     str(existing[4]), float(existing[5]), score
                 )
+
+            claim_columns = {
+                str(row[1])
+                for row in self.conn.execute("PRAGMA table_info(claims)").fetchall()
+            }
+            active_claims = 0
+            if {"worker_id", "status"}.issubset(claim_columns):
+                sql = "SELECT COUNT(*) FROM claims WHERE worker_id=? AND status='ACTIVE'"
+                params: list[Any] = [worker_id]
+                if "lease_expires_at" in claim_columns:
+                    sql += " AND lease_expires_at>?"
+                    params.append(now)
+                row = self.conn.execute(sql, tuple(params)).fetchone()
+                active_claims = int(row[0] if row else 0)
+            row = self.conn.execute(
+                """
+                SELECT COUNT(*) FROM fleet_dispatches
+                 WHERE worker_id=? AND task_id<>?
+                   AND state='RESERVED' AND reserved_until>?
+                """,
+                (worker_id, task_id, now),
+            ).fetchone()
+            active_reservations = int(row[0] if row else 0)
+            if active_claims + active_reservations >= max_slots:
+                raise FleetConflict("FLEET_WORKER_CAPACITY_EXHAUSTED")
+
             self.conn.execute(
                 """
                 INSERT INTO fleet_dispatches(
@@ -682,19 +715,23 @@ class FleetScheduler:
         return getter(worker_id)
 
     def _active_claims(self, worker_id: str, now: float) -> int:
-        columns = {
-            str(row[1])
-            for row in self.store.conn.execute("PRAGMA table_info(claims)").fetchall()
-        }
-        required = {"worker_id", "status"}
-        if not required.issubset(columns):
-            raise FleetError("FLEET_CLAIMS_SCHEMA_UNSUPPORTED")
-        sql = "SELECT COUNT(*) FROM claims WHERE worker_id=? AND status='ACTIVE'"
-        params: list[Any] = [worker_id]
-        if "lease_expires_at" in columns:
-            sql += " AND lease_expires_at>?"
-            params.append(now)
-        row = self.store.conn.execute(sql, tuple(params)).fetchone()
+        # The TaskPlane store owns one SQLite connection in the current runtime.
+        # Fleet reads must share its lock just like fleet writes do, otherwise two
+        # concurrent pullers can issue overlapping operations on the same handle.
+        with self.state.lock:
+            columns = {
+                str(row[1])
+                for row in self.store.conn.execute("PRAGMA table_info(claims)").fetchall()
+            }
+            required = {"worker_id", "status"}
+            if not required.issubset(columns):
+                raise FleetError("FLEET_CLAIMS_SCHEMA_UNSUPPORTED")
+            sql = "SELECT COUNT(*) FROM claims WHERE worker_id=? AND status='ACTIVE'"
+            params: list[Any] = [worker_id]
+            if "lease_expires_at" in columns:
+                sql += " AND lease_expires_at>?"
+                params.append(now)
+            row = self.store.conn.execute(sql, tuple(params)).fetchone()
         return int(row[0] if row else 0)
 
     @staticmethod
@@ -764,6 +801,52 @@ class FleetScheduler:
             and _task_id(task)
         ]
 
+    def rank_for_worker(
+        self,
+        worker_id: str,
+        *,
+        actor_context: Any = None,
+        now: float | None = None,
+    ) -> list[object]:
+        """Return fleet-ranked tasks without reserving or claiming anything."""
+        now = float(self.now_fn() if now is None else now)
+        worker_id = _text(worker_id)
+        if not worker_id:
+            return []
+        ranked: list[object] = []
+        for task in self._tasks(actor_context):
+            eligible, _, _ = self._eligible_worker(worker_id, task, now)
+            if eligible:
+                ranked.append(task)
+        return sorted(ranked, key=lambda task: self._task_score(task, now))
+
+    def reserve_for_worker(
+        self,
+        worker_id: str,
+        task: object,
+        *,
+        now: float | None = None,
+    ) -> FleetDispatchDecision:
+        """Reserve exactly one already-ranked task, rechecking eligibility and capacity."""
+        now = float(self.now_fn() if now is None else now)
+        worker_id = _text(worker_id)
+        eligible, _, policy = self._eligible_worker(worker_id, task, now)
+        if not eligible:
+            raise FleetConflict("FLEET_WORKER_NOT_ELIGIBLE")
+        metadata = _task_metadata(task)
+        scope_weight = _float(metadata.get("fleet_scope_weight"), 1.0, 0.05, 100.0)
+        return self.state.reserve(
+            task_id=_task_id(task),
+            worker_id=worker_id,
+            task_revision=_task_revision(task),
+            scope_key=_task_scope_key(task),
+            failure_domain=policy.failure_domain,
+            score=self._task_score(task, now),
+            reservation_seconds=self.reservation_seconds,
+            now=now,
+            scope_weight=scope_weight,
+        )
+
     def select(
         self,
         worker_ids: Sequence[str],

active-runtime/app/task_plane/scheduler.py

@@ -84,3 +84,193 @@ from .fleet import (
     FleetStateStore,
     FleetWorkerPolicy,
 )
+# R4.2 P2-B fleet-aware pull scheduling.
+#
+# The original TaskScheduler remains the security oracle and claim-CAS implementation.
+# Fleet ordering only controls the order presented to that oracle.
+import inspect as _fleet_inspect
+import threading as _fleet_threading
+from collections.abc import Mapping as _FleetMapping
+
+from .worker_registry import WorkerRegistry as _FleetWorkerRegistry
+from .errors import NotFoundError as _FleetNotFoundError
+
+_LegacyTaskScheduler = TaskScheduler
+_LegacyScheduleParameters = _fleet_inspect.signature(
+    _LegacyTaskScheduler.schedule_for_worker
+).parameters
+_LegacyNextParameters = _fleet_inspect.signature(
+    _LegacyTaskScheduler.next_queued_task
+).parameters
+
+
+class _FleetOrderedStoreProxy:
+    def __init__(self, delegate, tasks):
+        self._delegate = delegate
+        self._tasks = tuple(tasks)
+
+    def list_tasks(self, *args, **kwargs):
+        return list(self._tasks)
+
+    def __getattr__(self, name):
+        return getattr(self._delegate, name)
+
+
+def _fleet_claim_field(claim, name, default=None):
+    if isinstance(claim, _FleetMapping):
+        return claim.get(name, default)
+    return getattr(claim, name, default)
+
+
+class TaskScheduler(_LegacyTaskScheduler):
+    """Fleet-integrated pull scheduler preserving legacy security and claim semantics."""
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        store = getattr(self, "store", args[0] if args else None)
+        if store is None:
+            raise FleetError("FLEET_SCHEDULER_STORE_REQUIRED")
+        registry = getattr(self, "workers", None)
+        if registry is None:
+            registry = _FleetWorkerRegistry(store)
+        self.fleet_scheduler = FleetScheduler(store, registry)
+        self._fleet_local = _fleet_threading.local()
+
+    def _fleet_legacy_next(self, worker_id, tasks, *args, **kwargs):
+        proxy = _FleetOrderedStoreProxy(self.store, tasks)
+        legacy = _LegacyTaskScheduler(proxy)
+        legacy_kwargs = dict(kwargs)
+        if "actor_context" not in _LegacyNextParameters:
+            legacy_kwargs.pop("actor_context", None)
+        return legacy.next_queued_task(worker_id, *args, **legacy_kwargs)
+
+    def next_queued_task(self, worker_id, *args, **kwargs):
+        actor_context = kwargs.get(
+            "actor_context", getattr(self._fleet_local, "actor_context", None)
+        )
+
+        # Preserve legacy fail-closed worker checks before fleet ordering.
+        probe_kwargs = dict(kwargs)
+        if "actor_context" not in _LegacyNextParameters:
+            probe_kwargs.pop("actor_context", None)
+        legacy_probe = _LegacyTaskScheduler(self.store).next_queued_task(
+            worker_id, *args, **probe_kwargs
+        )
+        if legacy_probe is None:
+            return None
+
+        ranked = self.fleet_scheduler.rank_for_worker(
+            worker_id, actor_context=actor_context
+        )
+        # Preserve FleetScheduler fairness order. The legacy scheduler is used
+        # only as a fail-closed security oracle: present one fleet-ranked task
+        # at a time so its internal sort cannot reorder across scopes.
+        # Keep the legacy priority dimension ahead of Fleet fairness for
+        # compatibility with existing TaskPlane scheduling contracts; the
+        # Fleet order remains authoritative among equal-priority tasks.
+        def _legacy_priority(task):
+            plan = getattr(task, "task_plan", task)
+            metadata = getattr(plan, "metadata", {}) or {}
+            if not isinstance(metadata, _FleetMapping):
+                return 0
+            try:
+                return int(metadata.get("priority", 0))
+            except (TypeError, ValueError):
+                return 0
+
+        ranked = sorted(ranked, key=lambda task: -_legacy_priority(task))
+        for ranked_candidate in ranked:
+            try:
+                candidate = self._fleet_legacy_next(
+                    worker_id, [ranked_candidate], *args, **kwargs
+                )
+            except _FleetNotFoundError:
+                continue
+            if candidate is None:
+                continue
+
+            ranked_id = getattr(
+                ranked_candidate,
+                "task_id",
+                getattr(getattr(ranked_candidate, "task_plan", None), "task_id", None),
+            )
+            candidate_id = getattr(
+                candidate,
+                "task_id",
+                getattr(getattr(candidate, "task_plan", None), "task_id", None),
+            )
+            if candidate_id != ranked_id:
+                raise FleetConflict("FLEET_LEGACY_ORACLE_ORDER_VIOLATION")
+
+            try:
+                decision = self.fleet_scheduler.reserve_for_worker(
+                    worker_id, candidate
+                )
+            except FleetConflict:
+                continue
+            self._fleet_local.pending = decision
+            return candidate
+        return None
+
+    def schedule_for_worker(self, worker_id, *args, **kwargs):
+        actor_context = kwargs.get("actor_context")
+        legacy_kwargs = dict(kwargs)
+        if "actor_context" not in _LegacyScheduleParameters:
+            legacy_kwargs.pop("actor_context", None)
+        self._fleet_local.actor_context = actor_context
+        self._fleet_local.pending = None
+        try:
+            claim = _LegacyTaskScheduler.schedule_for_worker(
+                self, worker_id, *args, **legacy_kwargs
+            )
+            decision = getattr(self._fleet_local, "pending", None)
+            if claim is None:
+                if decision is not None:
+                    self.fleet_scheduler.state.release(
+                        decision.task_id, state="RELEASED"
+                    )
+                return None
+            if decision is not None:
+                claim_id = _fleet_claim_field(claim, "claim_id")
+                claim_task_id = _fleet_claim_field(claim, "task_id", decision.task_id)
+                if not claim_id:
+                    self.fleet_scheduler.state.release(
+                        decision.task_id, state="RELEASED"
+                    )
+                    raise FleetError("FLEET_CLAIM_ID_REQUIRED")
+                if claim_task_id != decision.task_id:
+                    self.fleet_scheduler.state.release(
+                        decision.task_id, state="RELEASED"
+                    )
+                    raise FleetConflict("FLEET_CLAIM_TASK_MISMATCH")
+                self.fleet_scheduler.state.bind_claim(
+                    decision.reservation_id, str(claim_id)
+                )
+            return claim
+        except Exception:
+            decision = getattr(self._fleet_local, "pending", None)
+            if decision is not None:
+                self.fleet_scheduler.state.release(
+                    decision.task_id, state="RELEASED"
+                )
+            raise
+        finally:
+            self._fleet_local.pending = None
+            self._fleet_local.actor_context = None
+
+    def record_worker_failure(self, worker_id, failure_class, **kwargs):
+        return self.fleet_scheduler.state.record_worker_failure(
+            worker_id, failure_class, **kwargs
+        )
+
+    def record_worker_success(self, worker_id, **kwargs):
+        return self.fleet_scheduler.state.record_worker_success(
+            worker_id, **kwargs
+        )
+
+    def quarantine_task_worker(
+        self, task_id, worker_id, failure_class, **kwargs
+    ):
+        return self.fleet_scheduler.state.record_task_worker_failure(
+            task_id, worker_id, failure_class, **kwargs
+        )

deploy/systemd/aios-agent-task-execution-rebind.service

@@ -0,0 +1,37 @@
+[Unit]
+Description=AIOS owner-controlled Task execution-rebind API
+After=aios-agent-task-api.service
+Wants=aios-agent-task-api.service
+
+[Service]
+Type=simple
+User=aios-agent-task
+Group=aios-agent-task
+LoadCredential=agent_task_control_token:/etc/aios-gitops/agent-task-control.token
+EnvironmentFile=/etc/aios-gitops/agent-task-execution-rebind.env
+Environment=AIOS_AGENT_TASK_DB=/var/lib/aios-agent-task/state.sqlite3
+Environment=AIOS_AGENT_TASK_CONTROL_TOKEN_FILE=%d/agent_task_control_token
+Environment=AIOS_GOVERNED_WRITE_API_URL=http://127.0.0.1:18892
+Environment=AIOS_TASK_EXECUTION_REBIND_HOST=127.0.0.1
+Environment=AIOS_TASK_EXECUTION_REBIND_PORT=18942
+ExecStart=/usr/bin/python3 -B /opt/aios/current/active-runtime/app/aios_agent_task_execution_rebind.py
+Restart=on-failure
+RestartSec=2
+NoNewPrivileges=true
+PrivateTmp=true
+ProtectSystem=strict
+ProtectHome=true
+ProtectKernelTunables=true
+ProtectKernelModules=true
+ProtectControlGroups=true
+LockPersonality=true
+RestrictSUIDSGID=true
+RestrictRealtime=true
+RestrictNamespaces=true
+MemoryDenyWriteExecute=true
+RestrictAddressFamilies=AF_UNIX AF_INET
+ReadWritePaths=/var/lib/aios-agent-task
+UMask=0077
+
+[Install]
+WantedBy=multi-user.target

docs/superpowers/specs/2026-08-17-r42-p2b-fleet-order-preservation-repair.md

@@ -0,0 +1,17 @@
+# R4.2 P2-B Fleet Order Preservation Repair
+
+P2-C live acceptance exposed a real integration defect in the verified P2-B candidate. The fleet
+scheduler correctly produced durable cross-scope fairness order, but the integration passed the
+entire ranked list to the legacy scheduler. The legacy scheduler then applied its own priority /
+created_at / task_id sort and silently discarded fleet ordering.
+
+The repair preserves the legacy scheduler as the fail-closed security oracle while changing how it
+is called: one fleet-ranked candidate is presented at a time. A candidate rejected by legacy
+security/policy advances to the next fleet candidate. A candidate returned by the oracle must have
+the same task ID as the candidate presented, otherwise execution fails closed.
+
+Task ownership remains unchanged: the original schedule_for_worker/claim_task CAS remains
+authoritative, and a fleet reservation is still only bound after the claim succeeds.
+
+This repair does not modify FleetScheduler state semantics, capacity, WriterLease, checkpoint,
+protected refs or Production behavior.

docs/superpowers/specs/2026-08-17-r42-p2b-fleet-runtime-integration.md

@@ -0,0 +1,23 @@
+# R4.2 P2-B Fleet Runtime Integration
+
+P2-B connects the verified P2 fleet primitives to the existing TaskPlane pull scheduler without
+replacing its security or ownership logic.
+
+The integration uses two stages. `FleetScheduler.rank_for_worker()` performs only read-only
+capacity/fairness/fault-isolation ranking. The original `TaskScheduler.next_queued_task()` is then
+run over that ordered list and remains the fail-closed security oracle for worker liveness,
+capabilities, data classification, execution zone and other existing policy checks. Only a task
+accepted by the original scheduler receives a short durable fleet reservation.
+
+The original `TaskScheduler.schedule_for_worker()` remains the sole claim path and continues to call
+the existing TaskPlane `claim_task` CAS. After a successful claim, the claim ID is bound to the
+fleet reservation. A claim error releases the reservation. A fleet reservation is never treated as
+task ownership.
+
+P2-B also makes the reservation capacity check atomic inside the fleet SQLite transaction so two
+concurrent pullers cannot oversubscribe the same worker slot. Circuit and task/worker quarantine
+state are enforced on the real pull path. Explicit execution-failure signals can be recorded through
+the integrated scheduler's failure/success/quarantine methods; policy/claim failures are not
+automatically mislabeled as worker crashes.
+
+No new daemon, database or production write path is introduced.

docs/superpowers/specs/2026-08-18-generic-compact-interface-map-profiles.md

@@ -0,0 +1,18 @@
+# Generic Compact Interface-Map Profiles
+
+The compact interface-map runtime is a reusable read-only control-plane capability. It keeps the
+canonical `source_capture` action and `compact_interface_map` adapter, but selects a bounded semantic
+profile from the task's source signals.
+
+The existing `fleet` profile remains backward compatible and returns
+`P2_INTERFACE_MAP_CAPTURED` with the `FLEET_*` contracts.
+
+The new `model_governance` profile is selected when model/provider/routing/budget/evaluation terms
+are present. It returns `MODEL_GOVERNANCE_INTERFACE_MAP_CAPTURED` and model-specific contracts for
+provider registry, selection policy, version pinning, budget accounting, fallback, scope isolation,
+credential boundary, trace/audit, evaluation gates and routing resilience.
+
+Profile and classification are emitted by the compact-map module and revalidated by the worker
+before result submission. Unknown or mismatched profile/classification pairs fail closed.
+
+No new Bridge action, daemon, database, WriterLease, mutation path or credential surface is added.

docs/superpowers/specs/2026-08-18-owner-controlled-execution-rebind.md

@@ -0,0 +1,36 @@
+# Owner-Controlled Same-Binding Execution Rebind
+
+## Purpose
+
+`aios_agent_task_execution_rebind.py` closes a projection/plan race without replacing the Task,
+Governed Job, target branch, generation, or idempotency lineage.
+
+The only supported transition is:
+
+```text
+official NEEDS_REWORK
+→ Task QUEUED with no active lease
+→ persisted source_capture envelope
+→ canonical Bridge resolves an already-registered deterministic plan
+→ CAS-bound metadata.execution + execution_projection replacement
+→ audit event
+→ normal worker claim
+```
+
+The service rejects arbitrary metadata patches. It derives the new envelope from canonical Bridge
+`/v2/execution-envelope`, validates every governed binding coordinate against the persisted envelope,
+requires `source_capture → deterministic`, requires exact old/new execution digests supplied by the
+Owner recovery request, and increments the Task revision under `BEGIN IMMEDIATE`.
+
+It records `TASK_EXECUTION_REBIND_AUTHORIZED` and an idempotent `task_execution_rebinds` receipt.
+No Task/Job/branch replacement is created and the worker still performs the real claim/heartbeat,
+WriterLease and governed commit.
+
+The endpoint is loopback-only:
+
+`POST /v1/tasks/{task_id}/execution-rebind`
+
+and requires the existing Owner control credential plus
+`AIOS_AGENT_TASK_EXECUTION_REBIND_AUTHORIZATION_ID`.
+
+Direct/manual Task database edits remain forbidden; the service is the audited mutation boundary.

tests/task_plane/public_interface/test_task_plane_fleet_order_preservation.py

@@ -0,0 +1,70 @@
+from __future__ import annotations
+
+import importlib.util
+from pathlib import Path
+import sys
+
+ROOT = Path(__file__).resolve().parents[3]
+ACTIVE_RUNTIME = ROOT / "active-runtime"
+if str(ACTIVE_RUNTIME) not in sys.path:
+    sys.path.insert(0, str(ACTIVE_RUNTIME))
+
+from app.task_plane.scheduler import TaskScheduler
+
+
+def _security_contract():
+    path = Path(__file__).with_name("test_task_plane_security_contract.py")
+    spec = importlib.util.spec_from_file_location("p2b_order_security_contract", path)
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+def _create_task(service, contract, task_id, tenant, project):
+    return service.create_task(
+        {
+            "task_id": task_id,
+            "issuer": {"type": "agent", "id": contract.ADMIN_IDENTITY},
+            "required_capabilities": ["code.read"],
+            "scope_type": "tenant",
+            "tenant_id": tenant,
+            "project_id": project,
+            "metadata": {"fleet_priority": 0},
+        }
+    )
+
+
+def _task_id(claim):
+    return getattr(
+        claim,
+        "task_id",
+        claim.get("task_id") if isinstance(claim, dict) else None,
+    )
+
+
+def test_P2B_FLEET_ORDER_is_preserved_through_legacy_claim_oracle():
+    contract = _security_contract()
+    service = contract._service()
+    service.register_worker(contract._worker_payload(max_concurrency=3))
+    worker_id = contract.WORKER_ID
+
+    scheduler = TaskScheduler(service.store)
+    scheduler.fleet_scheduler.configure_worker(
+        worker_id, max_slots=3, failure_domain="zone-a"
+    )
+
+    _create_task(service, contract, "p2b-order-a1", "tenant-a", "project-a")
+    _create_task(service, contract, "p2b-order-a2", "tenant-a", "project-a")
+    _create_task(service, contract, "p2b-order-b1", "tenant-b", "project-b")
+
+    first = scheduler.schedule_for_worker(worker_id)
+    second = scheduler.schedule_for_worker(worker_id)
+
+    assert _task_id(first) == "p2b-order-a1"
+    assert _task_id(second) == "p2b-order-b1"
+
+    rows = service.store.conn.execute(
+        "SELECT scope_key,dispatch_count FROM fleet_scope_state ORDER BY scope_key"
+    ).fetchall()
+    assert len(rows) >= 2

tests/task_plane/public_interface/test_task_plane_fleet_runtime_integration.py

@@ -0,0 +1,181 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+import importlib.util
+from pathlib import Path
+import sqlite3
+import sys
+import threading
+import time
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[3]
+ACTIVE_RUNTIME = ROOT / "active-runtime"
+if str(ACTIVE_RUNTIME) not in sys.path:
+    sys.path.insert(0, str(ACTIVE_RUNTIME))
+
+from app.task_plane.fleet import FleetConflict, FleetScheduler
+from app.task_plane.errors import NotFoundError
+from app.task_plane.scheduler import TaskScheduler
+
+
+def _security_contract():
+    path = Path(__file__).with_name("test_task_plane_security_contract.py")
+    spec = importlib.util.spec_from_file_location("p2b_security_contract", path)
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+def test_P2B_REAL_DISPATCH_claim_binds_fleet_reservation_and_circuit():
+    contract = _security_contract()
+    service = contract._service()
+    service.register_worker(contract._worker_payload(max_concurrency=2))
+    worker_id = contract.WORKER_ID
+
+    service.create_task(
+        {
+            "task_id": "task-p2b-real-1",
+            "issuer": {"type": "agent", "id": contract.ADMIN_IDENTITY},
+            "required_capabilities": ["code.read"],
+        }
+    )
+
+    scheduler = TaskScheduler(service.store)
+    scheduler.fleet_scheduler.configure_worker(
+        worker_id,
+        max_slots=2,
+        failure_domain="zone-real",
+        labels={"runtime": "task-plane"},
+    )
+    first = scheduler.schedule_for_worker(worker_id)
+    assert first is not None
+    assert first.task_id == "task-p2b-real-1"
+
+    row = service.store.conn.execute(
+        "SELECT state,claim_id,worker_id FROM fleet_dispatches WHERE task_id=?",
+        ("task-p2b-real-1",),
+    ).fetchone()
+    assert tuple(row) == ("CLAIMED", first.claim_id, worker_id)
+
+    # A real circuit blocks the actual scheduler even though one capacity slot remains.
+    failure_now = time.time()
+    scheduler.record_worker_failure(
+        worker_id, "runtime_crash", threshold=1, now=failure_now
+    )
+    service.create_task(
+        {
+            "task_id": "task-p2b-real-2",
+            "issuer": {"type": "agent", "id": contract.ADMIN_IDENTITY},
+            "required_capabilities": ["code.read"],
+        }
+    )
+    with pytest.raises(NotFoundError, match="no queued tasks available"):
+        scheduler.schedule_for_worker(worker_id)
+
+    scheduler.record_worker_success(worker_id, now=failure_now + 1.0)
+    second = scheduler.schedule_for_worker(worker_id)
+    assert second is not None
+    assert second.task_id == "task-p2b-real-2"
+
+
+@dataclass
+class Plan:
+    task_id: str
+    required_capabilities: list[str] = field(default_factory=lambda: ["code.read"])
+    metadata: dict = field(default_factory=dict)
+    scope_type: str = "tenant"
+    tenant_id: str = "tenant-a"
+    project_id: str | None = "project-a"
+    user_id: str | None = None
+
+
+@dataclass
+class Task:
+    task_id: str
+    revision: int = 1
+    status: str = "QUEUED"
+    created_at: float = 1.0
+    task_plan: Plan | None = None
+
+    def __post_init__(self):
+        if self.task_plan is None:
+            self.task_plan = Plan(self.task_id)
+
+
+@dataclass
+class Worker:
+    worker_id: str
+    capabilities: list[str] = field(default_factory=lambda: ["code.read"])
+
+
+class Store:
+    def __init__(self):
+        self.conn = sqlite3.connect(":memory:", check_same_thread=False)
+        self._lock = threading.RLock()
+        self.conn.execute(
+            "CREATE TABLE claims("
+            "claim_id TEXT PRIMARY KEY,task_id TEXT,worker_id TEXT,status TEXT,"
+            "lease_expires_at REAL)"
+        )
+        self.tasks = []
+        self.workers = {}
+
+    def list_tasks(self, _ctx=None):
+        return list(self.tasks)
+
+    def get_worker(self, worker_id):
+        return self.workers[worker_id]
+
+
+class Registry:
+    def __init__(self):
+        self.online = set()
+
+    def is_online(self, worker_id, now=None):
+        return worker_id in self.online
+
+
+def _fleet():
+    store = Store()
+    registry = Registry()
+    store.workers["w-a"] = Worker("w-a")
+    registry.online.add("w-a")
+    fleet = FleetScheduler(store, registry, now_fn=lambda: 1000.0)
+    fleet.configure_worker("w-a", max_slots=1, failure_domain="zone-a")
+    return store, fleet
+
+
+def test_P2B_CAPACITY_RESERVATION_is_atomic_under_concurrent_pullers():
+    store, fleet = _fleet()
+    tasks = [Task("task-a"), Task("task-b")]
+    store.tasks = tasks
+    barrier = threading.Barrier(2)
+    outcomes = []
+
+    def reserve(task):
+        barrier.wait()
+        try:
+            fleet.reserve_for_worker("w-a", task, now=1000.0)
+            outcomes.append("reserved")
+        except FleetConflict:
+            outcomes.append("conflict")
+
+    threads = [threading.Thread(target=reserve, args=(task,)) for task in tasks]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join()
+
+    assert sorted(outcomes) == ["conflict", "reserved"]
+    assert fleet.state.reserved_slots("w-a", 1000.0) == 1
+
+
+def test_P2B_RANK_is_read_only_until_legacy_security_accepts_candidate():
+    store, fleet = _fleet()
+    store.tasks = [Task("task-a"), Task("task-b")]
+    ranked = fleet.rank_for_worker("w-a", now=1000.0)
+    assert [task.task_id for task in ranked] == ["task-a", "task-b"]
+    assert fleet.state.reserved_slots("w-a", 1000.0) == 0

tests/test_agent_task_execution_rebind.py

@@ -0,0 +1,293 @@
+from __future__ import annotations
+
+from pathlib import Path
+import hashlib
+import json
+import sqlite3
+import sys
+
+import pytest
+
+ROOT=Path(__file__).resolve().parents[1]
+sys.path.insert(0,str(ROOT/"active-runtime/app"))
+
+from aios_agent_task_execution_rebind import (
+    ExecutionRebindError,
+    ExecutionRebindRequest,
+    ExecutionRebindStore,
+)
+from aios_online_agent_execution import (
+    GovernedExecutionBinding,
+    project_execution_envelope,
+)
+
+AUTHORIZATION="owner-execution-rebind-20260818"
+TASK_ID="task-rebind-a"
+JOB_ID="job-rebind-a"
+BASE="a"*40
+BRANCH="agent/task-rebind-a"
+IDEMPOTENCY="task-rebind-a-binding-v1"
+
+
+@pytest.fixture(autouse=True)
+def _auth(monkeypatch):
+    monkeypatch.setenv(
+        "AIOS_AGENT_TASK_EXECUTION_REBIND_AUTHORIZATION_ID",
+        AUTHORIZATION,
+    )
+
+
+def canon(value):
+    return hashlib.sha256(
+        json.dumps(value,sort_keys=True,separators=(",",":")).encode()
+    ).hexdigest()
+
+
+def binding():
+    return GovernedExecutionBinding.from_mapping({
+        "task_id":TASK_ID,
+        "job_id":JOB_ID,
+        "repository":"aios/aios",
+        "base_oid":BASE,
+        "expected_branch_oid":BASE,
+        "target_branch":BRANCH,
+        "branch_mode":"CREATE_NEW",
+        "generation":1,
+        "idempotency_key":IDEMPOTENCY,
+    })
+
+
+def old_envelope():
+    return project_execution_envelope(
+        binding(),
+        {
+            "action":"source_capture",
+            "adapter":"compact_interface_map",
+            "capability_scope":[
+                "git.read","git.test","governed.write.branch","governed.write.file"
+            ],
+        },
+    )
+
+
+def new_envelope():
+    return project_execution_envelope(
+        binding(),
+        {
+            "action":"deterministic",
+            "adapter":"deterministic",
+            "capability_scope":[
+                "git.read","git.test","governed.write.branch","governed.write.file"
+            ],
+            "change_set":[
+                {
+                    "path":"active-runtime/app/example.py",
+                    "operation":"write_file",
+                    "content":"VALUE=1\\n",
+                }
+            ],
+            "verification":[
+                {
+                    "label":"unit",
+                    "argv":["/usr/bin/python3","-m","pytest","-q"],
+                    "cwd":".",
+                    "timeout_seconds":120,
+                }
+            ],
+            "governance":{
+                "commit_idempotency_key":"task-rebind-a-commit-v1",
+            },
+        },
+    )
+
+
+def make_db(tmp_path, *, status="QUEUED", revision=5, attempt=2, lease=False):
+    path=tmp_path/"state.sqlite3"
+    con=sqlite3.connect(path)
+    con.executescript("""
+    CREATE TABLE tasks(
+      task_id TEXT PRIMARY KEY,
+      status TEXT NOT NULL,
+      attempt INTEGER NOT NULL,
+      max_attempts INTEGER NOT NULL,
+      revision INTEGER NOT NULL,
+      claimed_by TEXT,
+      lease_token_hash TEXT,
+      lease_expires_at TEXT,
+      verification_json TEXT,
+      result_json TEXT,
+      metadata_json TEXT,
+      required_capabilities_json TEXT
+    );
+    CREATE TABLE task_events(
+      event_id INTEGER PRIMARY KEY AUTOINCREMENT,
+      task_id TEXT,
+      event_type TEXT,
+      agent_id TEXT,
+      payload_json TEXT,
+      created_at TEXT
+    );
+    """)
+    verification={"decision":"NEEDS_REWORK","reason":"WRONG_ENVELOPE_BEFORE_PLAN"}
+    old=old_envelope()
+    metadata={
+        "execution":old,
+        "execution_projection":{
+            "schema":"aios.online-agent.task-projection.v1",
+            "task_id":TASK_ID,
+            "job_id":JOB_ID,
+            "execution_digest":old["execution_digest"],
+        },
+        "preserved":{"value":"unchanged"},
+    }
+    con.execute(
+        "INSERT INTO tasks VALUES (?,?,?,?,?,?,?,?,?,?,?,?)",
+        (
+            TASK_ID,status,attempt,3,revision,
+            "previous-worker" if not lease else "active-worker",
+            "lease-hash" if lease else None,
+            "2099-01-01T00:00:00Z" if lease else None,
+            json.dumps(verification),
+            json.dumps({"status":"SUCCEEDED"}),
+            json.dumps(metadata),
+            json.dumps([
+                "git.read","git.test","governed.write.branch","governed.write.file"
+            ]),
+        ),
+    )
+    con.commit();con.close()
+    return path,verification
+
+
+def request(verification, *, revision=5, old=None, new=None, key="rebind-task-a-v1"):
+    old=old or old_envelope()
+    new=new or new_envelope()
+    return ExecutionRebindRequest.from_mapping({
+        "expected_status":"QUEUED",
+        "expected_revision":revision,
+        "expected_attempt":2,
+        "previous_verification_sha256":canon(verification),
+        "expected_old_execution_digest":old["execution_digest"],
+        "expected_new_execution_digest":new["execution_digest"],
+        "authorization_id":AUTHORIZATION,
+        "idempotency_key":key,
+        "reason":"projection-before-plan race recovery",
+    })
+
+
+def resolver(_task_id,_caps):
+    return new_envelope()
+
+
+def test_rebind_cas_updates_only_execution_projection_and_audits(tmp_path):
+    path,verification=make_db(tmp_path)
+    result=ExecutionRebindStore(path,resolver).rebind(
+        TASK_ID,request(verification)
+    )
+    assert result["status"]=="QUEUED"
+    assert result["revision"]==6
+    assert result["idempotent"] is False
+
+    con=sqlite3.connect(path)
+    row=con.execute(
+        "SELECT status,attempt,revision,claimed_by,metadata_json FROM tasks"
+    ).fetchone()
+    assert row[:4]==("QUEUED",2,6,None)
+    metadata=json.loads(row[4])
+    assert metadata["preserved"]=={"value":"unchanged"}
+    assert metadata["execution"]["action"]=="deterministic"
+    assert metadata["execution"]["execution_digest"]==new_envelope()["execution_digest"]
+    assert metadata["execution_projection"]["execution_digest"]==new_envelope()["execution_digest"]
+
+    event=con.execute(
+        "SELECT event_type,payload_json FROM task_events"
+    ).fetchone()
+    assert event[0]=="TASK_EXECUTION_REBIND_AUTHORIZED"
+    payload=json.loads(event[1])
+    assert payload["old_action"]=="source_capture"
+    assert payload["new_action"]=="deterministic"
+    assert payload["stale_claim_normalized"] is True
+    assert con.execute("SELECT COUNT(*) FROM task_execution_rebinds").fetchone()[0]==1
+
+
+def test_rebind_idempotent_replay(tmp_path):
+    path,verification=make_db(tmp_path)
+    store=ExecutionRebindStore(path,resolver)
+    req=request(verification)
+    first=store.rebind(TASK_ID,req)
+    second=store.rebind(TASK_ID,req)
+    assert first["idempotent"] is False
+    assert second["idempotent"] is True
+    assert second["new_execution_digest"]==first["new_execution_digest"]
+
+
+def test_rebind_rejects_active_lease(tmp_path):
+    path,verification=make_db(tmp_path,lease=True)
+    with pytest.raises(ExecutionRebindError,match="EXECUTION_REBIND_ACTIVE_LEASE_DENY"):
+        ExecutionRebindStore(path,resolver).rebind(TASK_ID,request(verification))
+
+
+def test_rebind_rejects_verification_drift(tmp_path):
+    path,_=make_db(tmp_path)
+    with pytest.raises(ExecutionRebindError,match="EXECUTION_REBIND_VERIFICATION_DRIFT"):
+        ExecutionRebindStore(path,resolver).rebind(
+            TASK_ID,request({"decision":"NEEDS_REWORK","reason":"other"})
+        )
+
+
+def test_rebind_rejects_nonqueued_task(tmp_path):
+    path,verification=make_db(tmp_path,status="VERIFYING")
+    with pytest.raises(ExecutionRebindError,match="EXECUTION_REBIND_CAS_MISMATCH"):
+        ExecutionRebindStore(path,resolver).rebind(TASK_ID,request(verification))
+
+
+def test_rebind_rejects_non_deterministic_canonical_resolution(tmp_path):
+    path,verification=make_db(tmp_path)
+    def wrong(_task,_caps):
+        return old_envelope()
+    with pytest.raises(ExecutionRebindError,match="EXECUTION_REBIND_DETERMINISTIC_PLAN_REQUIRED"):
+        ExecutionRebindStore(path,wrong).rebind(TASK_ID,request(verification))
+
+
+def test_rebind_rejects_binding_drift(tmp_path):
+    path,verification=make_db(tmp_path)
+    different=project_execution_envelope(
+        GovernedExecutionBinding.from_mapping({
+            "task_id":TASK_ID,
+            "job_id":"job-other",
+            "repository":"aios/aios",
+            "base_oid":BASE,
+            "expected_branch_oid":BASE,
+            "target_branch":BRANCH,
+            "branch_mode":"CREATE_NEW",
+            "generation":1,
+            "idempotency_key":IDEMPOTENCY,
+        }),
+        {
+            "action":"deterministic",
+            "capability_scope":["git.read"],
+            "change_set":[{"path":"x.py","operation":"write_file","content":"x=1\\n"}],
+        },
+    )
+    def wrong(_task,_caps):
+        return different
+    with pytest.raises(Exception):
+        ExecutionRebindStore(path,wrong).rebind(TASK_ID,request(verification))
+
+
+def test_rebind_request_is_narrow_and_digest_bound():
+    verification={"decision":"NEEDS_REWORK"}
+    data=request(verification).public()
+    data["arbitrary_metadata"]={"x":1}
+    with pytest.raises(ExecutionRebindError,match="EXECUTION_REBIND_FIELDS_INVALID"):
+        ExecutionRebindRequest.from_mapping(data)
+
+
+def test_execution_rebind_unit_is_unprivileged_and_owner_credential_bound():
+    unit=(ROOT/"deploy/systemd/aios-agent-task-execution-rebind.service").read_text()
+    assert "User=aios-agent-task" in unit
+    assert "Group=aios-agent-task" in unit
+    assert "LoadCredential=agent_task_control_token:" in unit
+    assert "Environment=AIOS_GOVERNED_WRITE_API_URL=http://127.0.0.1:18892" in unit
+    assert "RestrictAddressFamilies=AF_UNIX AF_INET" in unit
+    assert "User=root" not in unit

tests/test_online_agent_compact_interface_map_profiles.py

@@ -0,0 +1,139 @@
+from __future__ import annotations
+
+from pathlib import Path
+from types import SimpleNamespace
+import json
+import sys
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+APP = ROOT / "active-runtime/app"
+sys.path.insert(0, str(APP))
+
+from aios_online_agent_compact_map import build_compact_interface_map
+import aios_online_agent_worker as worker
+from aios_online_agent_worker import build_compact_interface_map_result
+
+
+def _repo(tmp_path: Path) -> Path:
+    root = tmp_path / "repo"
+    (root / "active-runtime/app").mkdir(parents=True)
+    (root / "tests").mkdir(parents=True)
+    (root / "active-runtime/app/model_router.py").write_text(
+        """
+class ProviderRegistry:
+    def register_provider(self, provider_id, model_version):
+        return provider_id, model_version
+
+class ModelRouter:
+    def select_model(self, tenant_id, budget_tokens, fallback_provider):
+        return fallback_provider
+
+def record_model_cost(trace_id, token_count, cost_usd):
+    return trace_id, token_count, cost_usd
+
+def evaluate_model_version(model_id, version, approval_gate):
+    return approval_gate
+""".strip() + "\\n",
+        encoding="utf-8",
+    )
+    (root / "tests/test_model_router.py").write_text(
+        """
+def test_model_budget_and_fallback():
+    assert True
+""".strip() + "\\n",
+        encoding="utf-8",
+    )
+    return root
+
+
+def _envelope(base_oid: str):
+    return SimpleNamespace(
+        base_oid=base_oid,
+        job_id="job-model-governance",
+        repository="aios/aios",
+        expected_branch_oid=base_oid,
+        target_branch="agent/model-governance-map",
+        generation=1,
+        digest="e" * 64,
+    )
+
+
+def test_model_governance_profile_contracts_and_classification(tmp_path):
+    root = _repo(tmp_path)
+    value = build_compact_interface_map(
+        root,
+        [
+            "model provider registry routing budget token cost fallback",
+            "evaluation approval version pinning tenant project credential trace",
+        ],
+        source_commit="a" * 40,
+    )
+    assert value["profile"] == "model_governance"
+    assert value["classification"] == "MODEL_GOVERNANCE_INTERFACE_MAP_CAPTURED"
+    assert set(value["contracts"]) == {
+        "MODEL_PROVIDER_REGISTRY",
+        "MODEL_SELECTION_POLICY",
+        "MODEL_VERSION_PINNING",
+        "MODEL_BUDGET_ACCOUNTING",
+        "MODEL_FALLBACK_POLICY",
+        "MODEL_SCOPE_ISOLATION",
+        "MODEL_CREDENTIAL_BOUNDARY",
+        "MODEL_TRACE_AUDIT",
+        "MODEL_EVALUATION_GATE",
+        "MODEL_ROUTING_RESILIENCE",
+    }
+    assert "FLEET_FAIRNESS" not in value["contracts"]
+
+
+def test_fleet_profile_is_backward_compatible(tmp_path):
+    root = _repo(tmp_path)
+    value = build_compact_interface_map(
+        root,
+        ["agent fleet scheduler worker capacity fairness"],
+        source_commit="b" * 40,
+    )
+    assert value["profile"] == "fleet"
+    assert value["classification"] == "P2_INTERFACE_MAP_CAPTURED"
+    assert "FLEET_AGENT_REGISTRATION" in value["contracts"]
+
+
+def test_worker_binds_model_profile_to_model_classification():
+    interface_map = {
+        "schema": "aios.engineering.interface-map.v1",
+        "profile": "model_governance",
+        "classification": "MODEL_GOVERNANCE_INTERFACE_MAP_CAPTURED",
+        "source_commit": "c" * 40,
+        "implementation_map": {"modules": [], "classes": [], "functions": [], "tests": []},
+        "bounded_excerpts": [],
+        "contracts": {"MODEL_PROVIDER_REGISTRY": {"status": "GAP", "surfaces": []}},
+        "result_bytes": 512,
+    }
+    result = build_compact_interface_map_result(
+        "r42-model-governance-interface-map-20260818",
+        _envelope("c" * 40),
+        {"interface_map": interface_map},
+    )
+    assert result["classification"] == "MODEL_GOVERNANCE_INTERFACE_MAP_CAPTURED"
+    assert result["interface_map_profile"] == "model_governance"
+    assert result["writer_lease"] == "NONE"
+
+
+def test_worker_rejects_profile_classification_mismatch():
+    interface_map = {
+        "schema": "aios.engineering.interface-map.v1",
+        "profile": "model_governance",
+        "classification": "P2_INTERFACE_MAP_CAPTURED",
+        "source_commit": "c" * 40,
+        "implementation_map": {"modules": [], "classes": [], "functions": [], "tests": []},
+        "bounded_excerpts": [],
+        "contracts": {},
+        "result_bytes": 512,
+    }
+    with pytest.raises(worker.WorkerError, match="COMPACT_INTERFACE_MAP_PROFILE_BINDING_INVALID"):
+        build_compact_interface_map_result(
+            "r42-model-governance-interface-map-20260818",
+            _envelope("c" * 40),
+            {"interface_map": interface_map},
+        )