feat(task-plane): wire fleet scheduler into real pull dispatch

active-runtime/app/task_plane/fleet.py

@@ -0,0 +1,911 @@
+"""Durable multi-agent fleet scheduling for TaskPlane.
+
+The fleet layer decides *which* eligible worker should receive *which* queued task.
+It never replaces TaskPlane claim/lease ownership: callers must still perform the
+authoritative TaskPlane claim CAS after a reservation is returned.
+"""
+from __future__ import annotations
+
+from contextlib import contextmanager
+from dataclasses import dataclass
+import hashlib
+import json
+import math
+import sqlite3
+import threading
+import time
+from typing import Any, Iterable, Mapping, Sequence
+
+DEFAULT_RESERVATION_SECONDS = 30
+DEFAULT_CIRCUIT_THRESHOLD = 3
+DEFAULT_CIRCUIT_SECONDS = 30
+MAX_CIRCUIT_SECONDS = 15 * 60
+MAX_SLOTS = 128
+
+
+class FleetError(RuntimeError):
+    pass
+
+
+class FleetConflict(FleetError):
+    pass
+
+
+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 _text(value: object, default: str = "") -> str:
+    if value is None:
+        return default
+    return str(value).strip() or default
+
+
+def _int(value: object, default: int, minimum: int, maximum: int) -> int:
+    if value is None:
+        return default
+    if isinstance(value, bool):
+        raise FleetError("FLEET_INTEGER_INVALID")
+    try:
+        result = int(value)
+    except (TypeError, ValueError) as exc:
+        raise FleetError("FLEET_INTEGER_INVALID") from exc
+    if result < minimum or result > maximum:
+        raise FleetError("FLEET_INTEGER_OUT_OF_RANGE")
+    return result
+
+
+def _float(value: object, default: float, minimum: float, maximum: float) -> float:
+    if value is None:
+        return default
+    try:
+        result = float(value)
+    except (TypeError, ValueError) as exc:
+        raise FleetError("FLEET_NUMBER_INVALID") from exc
+    if not math.isfinite(result) or result < minimum or result > maximum:
+        raise FleetError("FLEET_NUMBER_OUT_OF_RANGE")
+    return result
+
+
+def _mapping(value: object) -> Mapping[str, Any]:
+    if isinstance(value, Mapping):
+        return value
+    if hasattr(value, "to_dict"):
+        data = value.to_dict()
+        if isinstance(data, Mapping):
+            return data
+    if hasattr(value, "__dict__"):
+        return vars(value)
+    return {}
+
+
+def _attr(value: object, name: str, default: Any = None) -> Any:
+    if isinstance(value, Mapping):
+        return value.get(name, default)
+    return getattr(value, name, default)
+
+
+def _task_plan(task: object) -> object:
+    return _attr(task, "task_plan", _attr(task, "plan", task))
+
+
+def _task_id(task: object) -> str:
+    return _text(_attr(task, "task_id", _attr(_task_plan(task), "task_id", "")))
+
+
+def _task_status(task: object) -> str:
+    return _text(_attr(task, "status", "")).upper()
+
+
+def _task_revision(task: object) -> int:
+    return _int(_attr(task, "revision", 0), 0, 0, 2**31 - 1)
+
+
+def _task_created_at(task: object) -> float:
+    value = _attr(task, "created_at", _attr(task, "updated_at", 0.0))
+    try:
+        return float(value or 0.0)
+    except (TypeError, ValueError):
+        return 0.0
+
+
+def _task_required_capabilities(task: object) -> frozenset[str]:
+    plan = _task_plan(task)
+    value = _attr(plan, "required_capabilities", ())
+    if not isinstance(value, Sequence) or isinstance(value, (str, bytes)):
+        return frozenset()
+    return frozenset(_text(item) for item in value if _text(item))
+
+
+def _task_metadata(task: object) -> Mapping[str, Any]:
+    plan = _task_plan(task)
+    value = _attr(plan, "metadata", {})
+    return value if isinstance(value, Mapping) else {}
+
+
+def _task_scope_key(task: object) -> str:
+    plan = _task_plan(task)
+    scope_type = _text(_attr(plan, "scope_type", "legacy"), "legacy")
+    tenant = _text(_attr(plan, "tenant_id", ""))
+    project = _text(_attr(plan, "project_id", ""))
+    user = _text(_attr(plan, "user_id", ""))
+    if scope_type == "tenant":
+        return "tenant:" + tenant + "|project:" + (project or "*")
+    if scope_type == "system":
+        return "system"
+    return "legacy:" + (user or "*")
+
+
+def _worker_id(worker: object) -> str:
+    return _text(_attr(worker, "worker_id", ""))
+
+
+def _worker_capabilities(worker: object) -> frozenset[str]:
+    value = _attr(worker, "capabilities", ())
+    if isinstance(value, Mapping):
+        value = value.keys()
+    if not isinstance(value, Iterable) or isinstance(value, (str, bytes)):
+        return frozenset()
+    return frozenset(_text(item) for item in value if _text(item))
+
+
+@dataclass(frozen=True)
+class FleetWorkerPolicy:
+    worker_id: str
+    max_slots: int = 1
+    failure_domain: str = "default"
+    weight: float = 1.0
+    labels: Mapping[str, str] | None = None
+    revision: int = 1
+
+
+@dataclass(frozen=True)
+class FleetDispatchDecision:
+    reservation_id: str
+    task_id: str
+    worker_id: str
+    task_revision: int
+    scope_key: str
+    failure_domain: str
+    reserved_until: float
+    score: tuple[float, float, float, str]
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "reservation_id": self.reservation_id,
+            "task_id": self.task_id,
+            "worker_id": self.worker_id,
+            "task_revision": self.task_revision,
+            "scope_key": self.scope_key,
+            "failure_domain": self.failure_domain,
+            "reserved_until": self.reserved_until,
+            "score": list(self.score),
+        }
+
+
+class FleetStateStore:
+    """Small durable fleet journal colocated in the existing TaskPlane SQLite database."""
+
+    def __init__(self, connection: sqlite3.Connection, lock: threading.RLock | None = None) -> None:
+        self.conn = connection
+        self.lock = lock or threading.RLock()
+        self._savepoint_counter = 0
+        self._ensure_schema()
+
+    @contextmanager
+    def _transaction(self):
+        with self.lock:
+            self._savepoint_counter += 1
+            name = f"fleet_{threading.get_ident()}_{self._savepoint_counter}"
+            self.conn.execute(f"SAVEPOINT {name}")
+            try:
+                yield
+            except Exception:
+                self.conn.execute(f"ROLLBACK TO {name}")
+                self.conn.execute(f"RELEASE {name}")
+                raise
+            else:
+                self.conn.execute(f"RELEASE {name}")
+
+    def _ensure_schema(self) -> None:
+        with self.lock:
+            self.conn.executescript(
+                """
+                CREATE TABLE IF NOT EXISTS fleet_worker_capacity (
+                    worker_id TEXT PRIMARY KEY,
+                    max_slots INTEGER NOT NULL,
+                    failure_domain TEXT NOT NULL,
+                    weight REAL NOT NULL,
+                    labels_json TEXT NOT NULL,
+                    revision INTEGER NOT NULL,
+                    updated_at REAL NOT NULL
+                );
+                CREATE TABLE IF NOT EXISTS fleet_scope_state (
+                    scope_key TEXT PRIMARY KEY,
+                    virtual_finish REAL NOT NULL,
+                    dispatch_count INTEGER NOT NULL,
+                    last_dispatched_at REAL,
+                    revision INTEGER NOT NULL
+                );
+                CREATE TABLE IF NOT EXISTS fleet_dispatches (
+                    reservation_id TEXT PRIMARY KEY,
+                    task_id TEXT NOT NULL UNIQUE,
+                    worker_id TEXT NOT NULL,
+                    task_revision INTEGER NOT NULL,
+                    scope_key TEXT NOT NULL,
+                    failure_domain TEXT NOT NULL,
+                    state TEXT NOT NULL,
+                    reserved_until REAL NOT NULL,
+                    claim_id TEXT,
+                    created_at REAL NOT NULL,
+                    updated_at REAL NOT NULL
+                );
+                CREATE INDEX IF NOT EXISTS ix_fleet_dispatch_worker_state
+                    ON fleet_dispatches(worker_id, state, reserved_until);
+                CREATE TABLE IF NOT EXISTS fleet_worker_failures (
+                    worker_id TEXT PRIMARY KEY,
+                    consecutive_failures INTEGER NOT NULL,
+                    circuit_open_until REAL,
+                    last_failure_class TEXT,
+                    revision INTEGER NOT NULL,
+                    updated_at REAL NOT NULL
+                );
+                CREATE TABLE IF NOT EXISTS fleet_task_worker_failures (
+                    task_id TEXT NOT NULL,
+                    worker_id TEXT NOT NULL,
+                    failure_count INTEGER NOT NULL,
+                    blocked_until REAL,
+                    last_failure_class TEXT,
+                    revision INTEGER NOT NULL,
+                    updated_at REAL NOT NULL,
+                    PRIMARY KEY(task_id, worker_id)
+                );
+                """
+            )
+
+    def configure_worker(
+        self,
+        worker_id: str,
+        *,
+        max_slots: int = 1,
+        failure_domain: str = "default",
+        weight: float = 1.0,
+        labels: Mapping[str, str] | None = None,
+        expected_revision: int | None = None,
+        now: float | None = None,
+    ) -> FleetWorkerPolicy:
+        worker_id = _text(worker_id)
+        if not worker_id:
+            raise FleetError("FLEET_WORKER_ID_REQUIRED")
+        max_slots = _int(max_slots, 1, 1, MAX_SLOTS)
+        weight = _float(weight, 1.0, 0.05, 100.0)
+        failure_domain = _text(failure_domain, "default")
+        normalized_labels = {
+            _text(k): _text(v)
+            for k, v in dict(labels or {}).items()
+            if _text(k) and _text(v)
+        }
+        now = float(time.time() if now is None else now)
+        with self._transaction():
+            row = self.conn.execute(
+                "SELECT revision FROM fleet_worker_capacity WHERE worker_id=?",
+                (worker_id,),
+            ).fetchone()
+            current = int(row[0]) if row else 0
+            if expected_revision is not None and current != int(expected_revision):
+                raise FleetConflict("FLEET_WORKER_POLICY_CAS_MISMATCH")
+            revision = current + 1
+            self.conn.execute(
+                """
+                INSERT INTO fleet_worker_capacity(
+                    worker_id,max_slots,failure_domain,weight,labels_json,revision,updated_at
+                ) VALUES (?,?,?,?,?,?,?)
+                ON CONFLICT(worker_id) DO UPDATE SET
+                    max_slots=excluded.max_slots,
+                    failure_domain=excluded.failure_domain,
+                    weight=excluded.weight,
+                    labels_json=excluded.labels_json,
+                    revision=excluded.revision,
+                    updated_at=excluded.updated_at
+                """,
+                (
+                    worker_id,
+                    max_slots,
+                    failure_domain,
+                    weight,
+                    json.dumps(normalized_labels, sort_keys=True, separators=(",", ":")),
+                    revision,
+                    now,
+                ),
+            )
+        return FleetWorkerPolicy(
+            worker_id, max_slots, failure_domain, weight, normalized_labels, revision
+        )
+
+    def worker_policy(self, worker_id: str) -> FleetWorkerPolicy:
+        worker_id = _text(worker_id)
+        with self.lock:
+            row = self.conn.execute(
+                """
+                SELECT worker_id,max_slots,failure_domain,weight,labels_json,revision
+                FROM fleet_worker_capacity WHERE worker_id=?
+                """,
+                (worker_id,),
+            ).fetchone()
+        if row is None:
+            return FleetWorkerPolicy(worker_id=worker_id)
+        return FleetWorkerPolicy(
+            worker_id=str(row[0]),
+            max_slots=int(row[1]),
+            failure_domain=str(row[2]),
+            weight=float(row[3]),
+            labels=json.loads(row[4] or "{}"),
+            revision=int(row[5]),
+        )
+
+    def _cleanup_expired(self, now: float) -> None:
+        self.conn.execute(
+            """
+            UPDATE fleet_dispatches
+               SET state='EXPIRED', updated_at=?
+             WHERE state='RESERVED' AND reserved_until<=?
+            """,
+            (now, now),
+        )
+
+    def reserved_slots(self, worker_id: str, now: float) -> int:
+        with self.lock:
+            self._cleanup_expired(now)
+            row = self.conn.execute(
+                """
+                SELECT COUNT(*) FROM fleet_dispatches
+                 WHERE worker_id=? AND state='RESERVED' AND reserved_until>?
+                """,
+                (worker_id, now),
+            ).fetchone()
+        return int(row[0] if row else 0)
+
+    def scope_virtual_finish(self, scope_key: str) -> float:
+        with self.lock:
+            row = self.conn.execute(
+                "SELECT virtual_finish FROM fleet_scope_state WHERE scope_key=?",
+                (scope_key,),
+            ).fetchone()
+        return float(row[0]) if row else 0.0
+
+    def reserve(
+        self,
+        *,
+        task_id: str,
+        worker_id: str,
+        task_revision: int,
+        scope_key: str,
+        failure_domain: str,
+        score: tuple[float, float, float, str],
+        reservation_seconds: int,
+        now: float,
+        scope_weight: float,
+    ) -> FleetDispatchDecision:
+        reservation_seconds = _int(
+            reservation_seconds, DEFAULT_RESERVATION_SECONDS, 1, 300
+        )
+        reservation_id = "fleet-" + _sha(
+            {
+                "task_id": task_id,
+                "worker_id": worker_id,
+                "task_revision": task_revision,
+                "scope_key": scope_key,
+            }
+        )[:32]
+        reserved_until = now + reservation_seconds
+        with self._transaction():
+            self._cleanup_expired(now)
+            existing = self.conn.execute(
+                """
+                SELECT reservation_id,worker_id,task_revision,scope_key,failure_domain,
+                       reserved_until,state
+                  FROM fleet_dispatches WHERE task_id=?
+                """,
+                (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")
+                return FleetDispatchDecision(
+                    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(
+                    reservation_id,task_id,worker_id,task_revision,scope_key,failure_domain,
+                    state,reserved_until,claim_id,created_at,updated_at
+                ) VALUES (?,?,?,?,?,?,'RESERVED',?,NULL,?,?)
+                ON CONFLICT(task_id) DO UPDATE SET
+                    reservation_id=excluded.reservation_id,
+                    worker_id=excluded.worker_id,
+                    task_revision=excluded.task_revision,
+                    scope_key=excluded.scope_key,
+                    failure_domain=excluded.failure_domain,
+                    state='RESERVED',
+                    reserved_until=excluded.reserved_until,
+                    claim_id=NULL,
+                    updated_at=excluded.updated_at
+                """,
+                (
+                    reservation_id, task_id, worker_id, task_revision, scope_key,
+                    failure_domain, reserved_until, now, now,
+                ),
+            )
+            row = self.conn.execute(
+                "SELECT virtual_finish,dispatch_count,revision FROM fleet_scope_state WHERE scope_key=?",
+                (scope_key,),
+            ).fetchone()
+            current_finish = float(row[0]) if row else 0.0
+            dispatch_count = int(row[1]) if row else 0
+            revision = int(row[2]) if row else 0
+            next_finish = max(current_finish, 0.0) + 1.0 / max(scope_weight, 0.05)
+            self.conn.execute(
+                """
+                INSERT INTO fleet_scope_state(
+                    scope_key,virtual_finish,dispatch_count,last_dispatched_at,revision
+                ) VALUES (?,?,?,?,?)
+                ON CONFLICT(scope_key) DO UPDATE SET
+                    virtual_finish=excluded.virtual_finish,
+                    dispatch_count=excluded.dispatch_count,
+                    last_dispatched_at=excluded.last_dispatched_at,
+                    revision=excluded.revision
+                """,
+                (scope_key, next_finish, dispatch_count + 1, now, revision + 1),
+            )
+        return FleetDispatchDecision(
+            reservation_id, task_id, worker_id, task_revision, scope_key,
+            failure_domain, reserved_until, score
+        )
+
+    def bind_claim(self, reservation_id: str, claim_id: str, *, now: float | None = None) -> None:
+        now = float(time.time() if now is None else now)
+        with self._transaction():
+            row = self.conn.execute(
+                "SELECT state FROM fleet_dispatches WHERE reservation_id=?",
+                (reservation_id,),
+            ).fetchone()
+            if row is None:
+                raise FleetError("FLEET_RESERVATION_NOT_FOUND")
+            if str(row[0]) not in {"RESERVED", "CLAIMED"}:
+                raise FleetConflict("FLEET_RESERVATION_NOT_ACTIVE")
+            self.conn.execute(
+                """
+                UPDATE fleet_dispatches
+                   SET state='CLAIMED', claim_id=?, updated_at=?
+                 WHERE reservation_id=?
+                """,
+                (claim_id, now, reservation_id),
+            )
+
+    def release(
+        self,
+        task_id: str,
+        *,
+        state: str = "RELEASED",
+        now: float | None = None,
+    ) -> None:
+        if state not in {"RELEASED", "COMPLETED", "FAILED", "EXPIRED"}:
+            raise FleetError("FLEET_RESERVATION_STATE_INVALID")
+        now = float(time.time() if now is None else now)
+        with self._transaction():
+            self.conn.execute(
+                "UPDATE fleet_dispatches SET state=?,updated_at=? WHERE task_id=?",
+                (state, now, task_id),
+            )
+
+    def worker_circuit_open(self, worker_id: str, now: float) -> bool:
+        with self.lock:
+            row = self.conn.execute(
+                "SELECT circuit_open_until FROM fleet_worker_failures WHERE worker_id=?",
+                (worker_id,),
+            ).fetchone()
+        return bool(row and row[0] is not None and float(row[0]) > now)
+
+    def task_worker_blocked(self, task_id: str, worker_id: str, now: float) -> bool:
+        with self.lock:
+            row = self.conn.execute(
+                """
+                SELECT blocked_until FROM fleet_task_worker_failures
+                 WHERE task_id=? AND worker_id=?
+                """,
+                (task_id, worker_id),
+            ).fetchone()
+        return bool(row and row[0] is not None and float(row[0]) > now)
+
+    def record_worker_failure(
+        self,
+        worker_id: str,
+        failure_class: str,
+        *,
+        threshold: int = DEFAULT_CIRCUIT_THRESHOLD,
+        base_circuit_seconds: int = DEFAULT_CIRCUIT_SECONDS,
+        now: float | None = None,
+    ) -> dict[str, Any]:
+        now = float(time.time() if now is None else now)
+        threshold = _int(threshold, DEFAULT_CIRCUIT_THRESHOLD, 1, 20)
+        base_circuit_seconds = _int(base_circuit_seconds, DEFAULT_CIRCUIT_SECONDS, 1, 3600)
+        with self._transaction():
+            row = self.conn.execute(
+                """
+                SELECT consecutive_failures,revision FROM fleet_worker_failures
+                 WHERE worker_id=?
+                """,
+                (worker_id,),
+            ).fetchone()
+            failures = (int(row[0]) if row else 0) + 1
+            revision = (int(row[1]) if row else 0) + 1
+            open_until = None
+            if failures >= threshold:
+                exponent = min(failures - threshold, 8)
+                open_until = now + min(
+                    MAX_CIRCUIT_SECONDS, base_circuit_seconds * (2**exponent)
+                )
+            self.conn.execute(
+                """
+                INSERT INTO fleet_worker_failures(
+                    worker_id,consecutive_failures,circuit_open_until,last_failure_class,
+                    revision,updated_at
+                ) VALUES (?,?,?,?,?,?)
+                ON CONFLICT(worker_id) DO UPDATE SET
+                    consecutive_failures=excluded.consecutive_failures,
+                    circuit_open_until=excluded.circuit_open_until,
+                    last_failure_class=excluded.last_failure_class,
+                    revision=excluded.revision,
+                    updated_at=excluded.updated_at
+                """,
+                (worker_id, failures, open_until, _text(failure_class, "unknown"), revision, now),
+            )
+        return {
+            "worker_id": worker_id,
+            "consecutive_failures": failures,
+            "circuit_open_until": open_until,
+            "revision": revision,
+        }
+
+    def record_worker_success(self, worker_id: str, *, now: float | None = None) -> None:
+        now = float(time.time() if now is None else now)
+        with self._transaction():
+            self.conn.execute(
+                """
+                INSERT INTO fleet_worker_failures(
+                    worker_id,consecutive_failures,circuit_open_until,last_failure_class,
+                    revision,updated_at
+                ) VALUES (?,0,NULL,NULL,1,?)
+                ON CONFLICT(worker_id) DO UPDATE SET
+                    consecutive_failures=0,
+                    circuit_open_until=NULL,
+                    last_failure_class=NULL,
+                    revision=revision+1,
+                    updated_at=excluded.updated_at
+                """,
+                (worker_id, now),
+            )
+
+    def record_task_worker_failure(
+        self,
+        task_id: str,
+        worker_id: str,
+        failure_class: str,
+        *,
+        quarantine_seconds: int = 60,
+        now: float | None = None,
+    ) -> dict[str, Any]:
+        now = float(time.time() if now is None else now)
+        quarantine_seconds = _int(quarantine_seconds, 60, 1, 3600)
+        with self._transaction():
+            row = self.conn.execute(
+                """
+                SELECT failure_count,revision FROM fleet_task_worker_failures
+                 WHERE task_id=? AND worker_id=?
+                """,
+                (task_id, worker_id),
+            ).fetchone()
+            count = (int(row[0]) if row else 0) + 1
+            revision = (int(row[1]) if row else 0) + 1
+            blocked_until = now + quarantine_seconds
+            self.conn.execute(
+                """
+                INSERT INTO fleet_task_worker_failures(
+                    task_id,worker_id,failure_count,blocked_until,last_failure_class,
+                    revision,updated_at
+                ) VALUES (?,?,?,?,?,?,?)
+                ON CONFLICT(task_id,worker_id) DO UPDATE SET
+                    failure_count=excluded.failure_count,
+                    blocked_until=excluded.blocked_until,
+                    last_failure_class=excluded.last_failure_class,
+                    revision=excluded.revision,
+                    updated_at=excluded.updated_at
+                """,
+                (
+                    task_id, worker_id, count, blocked_until,
+                    _text(failure_class, "unknown"), revision, now,
+                ),
+            )
+        return {
+            "task_id": task_id,
+            "worker_id": worker_id,
+            "failure_count": count,
+            "blocked_until": blocked_until,
+            "revision": revision,
+        }
+
+
+class FleetScheduler:
+    """Deterministic capacity/fairness scheduler over existing TaskPlane records."""
+
+    def __init__(
+        self,
+        store: Any,
+        workers: Any,
+        *,
+        now_fn: Any = time.time,
+        reservation_seconds: int = DEFAULT_RESERVATION_SECONDS,
+    ) -> None:
+        self.store = store
+        self.workers = workers
+        self.now_fn = now_fn
+        self.reservation_seconds = _int(
+            reservation_seconds, DEFAULT_RESERVATION_SECONDS, 1, 300
+        )
+        connection = getattr(store, "conn", None)
+        if not isinstance(connection, sqlite3.Connection):
+            raise FleetError("FLEET_TASKPLANE_SQLITE_CONNECTION_REQUIRED")
+        lock = getattr(store, "_lock", None)
+        self.state = FleetStateStore(connection, lock)
+
+    def configure_worker(self, worker_id: str, **kwargs: Any) -> FleetWorkerPolicy:
+        return self.state.configure_worker(worker_id, **kwargs)
+
+    def _online(self, worker_id: str, now: float) -> bool:
+        method = getattr(self.workers, "is_online", None)
+        if not callable(method):
+            raise FleetError("FLEET_WORKER_REGISTRY_ONLINE_CHECK_REQUIRED")
+        try:
+            return bool(method(worker_id, now=now))
+        except TypeError:
+            return bool(method(worker_id))
+
+    def _worker_record(self, worker_id: str) -> object:
+        getter = getattr(self.store, "get_worker", None)
+        if not callable(getter):
+            raise FleetError("FLEET_WORKER_LOOKUP_REQUIRED")
+        return getter(worker_id)
+
+    def _active_claims(self, worker_id: str, now: float) -> int:
+        # 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
+    def _labels_match(policy: FleetWorkerPolicy, task: object) -> bool:
+        metadata = _task_metadata(task)
+        affinity = metadata.get("fleet_affinity_labels") or {}
+        if affinity and not isinstance(affinity, Mapping):
+            return False
+        labels = dict(policy.labels or {})
+        for key, value in dict(affinity).items():
+            if labels.get(_text(key)) != _text(value):
+                return False
+        avoid = metadata.get("fleet_avoid_failure_domains") or []
+        if isinstance(avoid, Sequence) and not isinstance(avoid, (str, bytes)):
+            if policy.failure_domain in {_text(item) for item in avoid}:
+                return False
+        return True
+
+    def _eligible_worker(self, worker_id: str, task: object, now: float) -> tuple[bool, int, FleetWorkerPolicy]:
+        if not self._online(worker_id, now):
+            return False, 0, self.state.worker_policy(worker_id)
+        worker = self._worker_record(worker_id)
+        policy = self.state.worker_policy(worker_id)
+        if self.state.worker_circuit_open(worker_id, now):
+            return False, 0, policy
+        if self.state.task_worker_blocked(_task_id(task), worker_id, now):
+            return False, 0, policy
+        if not self._labels_match(policy, task):
+            return False, 0, policy
+        required = _task_required_capabilities(task)
+        capabilities = _worker_capabilities(worker)
+        if required and not required.issubset(capabilities):
+            return False, 0, policy
+        active = self._active_claims(worker_id, now)
+        reserved = self.state.reserved_slots(worker_id, now)
+        load = active + reserved
+        if load >= policy.max_slots:
+            return False, load, policy
+        return True, load, policy
+
+    @staticmethod
+    def _task_priority(task: object, now: float) -> float:
+        metadata = _task_metadata(task)
+        base = _int(metadata.get("fleet_priority"), 0, -100, 100)
+        created = _task_created_at(task)
+        waited = max(0.0, now - created) if created > 0 else 0.0
+        aging = min(100.0, waited / 60.0)
+        return float(base) + aging
+
+    def _task_score(self, task: object, now: float) -> tuple[float, float, float, str]:
+        scope = _task_scope_key(task)
+        virtual_finish = self.state.scope_virtual_finish(scope)
+        priority = self._task_priority(task, now)
+        return (virtual_finish, -priority, _task_created_at(task), _task_id(task))
+
+    def _tasks(self, actor_context: Any = None) -> list[object]:
+        lister = getattr(self.store, "list_tasks", None)
+        if not callable(lister):
+            raise FleetError("FLEET_TASK_LIST_REQUIRED")
+        try:
+            tasks = lister(actor_context)
+        except TypeError:
+            tasks = lister()
+        return [
+            task for task in tasks
+            if _task_status(task) in {"CREATED", "QUEUED"}
+            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],
+        *,
+        actor_context: Any = None,
+        limit: int = 1,
+        now: float | None = None,
+    ) -> list[FleetDispatchDecision]:
+        now = float(self.now_fn() if now is None else now)
+        limit = _int(limit, 1, 1, 100)
+        worker_ids = tuple(dict.fromkeys(_text(item) for item in worker_ids if _text(item)))
+        tasks = sorted(self._tasks(actor_context), key=lambda task: self._task_score(task, now))
+        decisions: list[FleetDispatchDecision] = []
+        local_load: dict[str, int] = {}
+        for task in tasks:
+            if len(decisions) >= limit:
+                break
+            candidates: list[tuple[int, float, str, FleetWorkerPolicy]] = []
+            for worker_id in worker_ids:
+                eligible, load, policy = self._eligible_worker(worker_id, task, now)
+                if not eligible:
+                    continue
+                effective_load = local_load.get(worker_id, load)
+                if effective_load >= policy.max_slots:
+                    continue
+                candidates.append(
+                    (
+                        effective_load,
+                        -policy.weight,
+                        worker_id,
+                        policy,
+                    )
+                )
+            if not candidates:
+                continue
+            candidates.sort(key=lambda item: (item[0], item[1], item[2]))
+            _, _, worker_id, policy = candidates[0]
+            score = self._task_score(task, now)
+            metadata = _task_metadata(task)
+            scope_weight = _float(metadata.get("fleet_scope_weight"), 1.0, 0.05, 100.0)
+            try:
+                decision = 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=score,
+                    reservation_seconds=self.reservation_seconds,
+                    now=now,
+                    scope_weight=scope_weight,
+                )
+            except FleetConflict:
+                continue
+            local_load[worker_id] = local_load.get(
+                worker_id,
+                self._active_claims(worker_id, now)
+                + self.state.reserved_slots(worker_id, now)
+                - 1,
+            ) + 1
+            decisions.append(decision)
+        return decisions

active-runtime/app/task_plane/scheduler.py

@@ -74,3 +74,183 @@ class TaskScheduler:
             operation_digest=task.task_plan.operation_digest(),
         )
         return claim
+
+# R4.2 P2 durable multi-agent fleet scheduling public surface.
+from .fleet import (
+    FleetConflict,
+    FleetDispatchDecision,
+    FleetError,
+    FleetScheduler,
+    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
+
+_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
+        )
+        while ranked:
+            candidate = self._fleet_legacy_next(
+                worker_id, ranked, *args, **kwargs
+            )
+            if candidate is None:
+                return None
+            try:
+                decision = self.fleet_scheduler.reserve_for_worker(
+                    worker_id, candidate
+                )
+            except FleetConflict:
+                candidate_id = getattr(
+                    candidate,
+                    "task_id",
+                    getattr(getattr(candidate, "task_plan", None), "task_id", None),
+                )
+                ranked = [
+                    task
+                    for task in ranked
+                    if getattr(
+                        task,
+                        "task_id",
+                        getattr(getattr(task, "task_plan", None), "task_id", None),
+                    )
+                    != candidate_id
+                ]
+                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
+        )

docs/superpowers/specs/2026-08-17-r42-p2-agent-fleet.md

@@ -0,0 +1,18 @@
+# R4.2 P2 Agent Fleet / Multi-Agent Scheduling / Fault Isolation
+
+P2 adds a durable fleet decision layer inside TaskPlane without replacing TaskPlane claim/lease
+ownership. Worker registration/heartbeat remains authoritative for liveness. TaskPlane claim CAS
+remains authoritative for ownership. Probe-before-reassignment remains mandatory after an expired
+lease. P1 checkpoint/result replay remains authoritative after worker restart.
+
+`FleetScheduler` adds durable worker capacity slots, weighted scope fairness, starvation aging,
+affinity/anti-affinity labels, worker circuit breakers, per-task/worker quarantine and idempotent
+dispatch reservations. Fleet state is colocated in the existing TaskPlane SQLite database and is
+created idempotently; no new daemon or database is introduced.
+
+A fleet reservation is only a scheduling decision. It does not claim a Task. Callers must perform
+the normal TaskPlane claim and may then bind the returned claim ID to the reservation. This preserves
+the existing ownership and recovery invariants.
+
+Failure recovery never bypasses probe evidence. When TaskPlane recovery places work back in a
+claimable state, P2 may quarantine the failed worker and prefer another worker/failure domain.

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.

tests/task_plane/public_interface/test_task_plane_fleet.py

@@ -0,0 +1,206 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+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))
+
+import sqlite3
+import threading
+
+import pytest
+
+from app.task_plane.fleet import FleetConflict, FleetScheduler
+
+
+@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, path=":memory:"):
+        self.conn=sqlite3.connect(path,check_same_thread=False)
+        self._lock=threading.RLock()
+        self.conn.executescript("""
+        CREATE TABLE IF NOT EXISTS 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
+
+
+@pytest.fixture
+def fleet():
+    store=Store(); registry=Registry()
+    for wid in ("w-a","w-b","w-c"):
+        store.workers[wid]=Worker(wid)
+        registry.online.add(wid)
+    scheduler=FleetScheduler(store,registry,now_fn=lambda:1000.0)
+    scheduler.configure_worker("w-a",max_slots=1,failure_domain="zone-a",weight=1)
+    scheduler.configure_worker("w-b",max_slots=2,failure_domain="zone-b",weight=1)
+    scheduler.configure_worker("w-c",max_slots=1,failure_domain="zone-c",weight=2)
+    return store,registry,scheduler
+
+
+def test_FLEET_AGENT_REGISTRATION_capacity_policy_is_durable(fleet):
+    store,_,scheduler=fleet
+    policy=scheduler.state.worker_policy("w-b")
+    assert policy.max_slots==2 and policy.failure_domain=="zone-b"
+    reloaded=type(scheduler)(store, scheduler.workers, now_fn=lambda:1000.0)
+    assert reloaded.state.worker_policy("w-b").max_slots==2
+
+
+def test_FLEET_CAPACITY_SCHEDULING_does_not_oversubscribe_worker(fleet):
+    store,_,scheduler=fleet
+    store.tasks=[Task("task-1"),Task("task-2"),Task("task-3")]
+    decisions=scheduler.select(["w-a"],limit=3,now=1000.0)
+    assert [d.worker_id for d in decisions]==["w-a"]
+
+
+def test_FLEET_CAPACITY_SCHEDULING_accounts_for_active_claims(fleet):
+    store,_,scheduler=fleet
+    store.conn.execute(
+      "INSERT INTO claims VALUES ('claim-1','other','w-a','ACTIVE',2000)"
+    )
+    store.tasks=[Task("task-1")]
+    assert scheduler.select(["w-a"],limit=1,now=1000.0)==[]
+
+
+def test_FLEET_FAIRNESS_scope_virtual_finish_prevents_hot_scope_monopoly(fleet):
+    store,_,scheduler=fleet
+    store.tasks=[
+      Task("a1",created_at=1,task_plan=Plan("a1",tenant_id="tenant-a",project_id="p-a")),
+      Task("a2",created_at=2,task_plan=Plan("a2",tenant_id="tenant-a",project_id="p-a")),
+      Task("b1",created_at=3,task_plan=Plan("b1",tenant_id="tenant-b",project_id="p-b")),
+    ]
+    first=scheduler.select(["w-b"],limit=1,now=1000.0)[0]
+    assert first.task_id=="a1"
+    scheduler.state.release("a1",state="COMPLETED",now=1000.1)
+    store.tasks=[task for task in store.tasks if task.task_id!="a1"]
+    second=scheduler.select(["w-b"],limit=1,now=1000.2)[0]
+    assert second.task_id=="b1"
+
+
+def test_FLEET_FAULT_ISOLATION_opens_worker_circuit_after_threshold(fleet):
+    store,_,scheduler=fleet
+    store.tasks=[Task("task-1")]
+    for _ in range(3):
+        scheduler.state.record_worker_failure("w-a","runtime_crash",now=1000.0)
+    assert scheduler.state.worker_circuit_open("w-a",1001.0)
+    assert scheduler.select(["w-a"],limit=1,now=1001.0)==[]
+
+
+def test_FLEET_WORKER_REASSIGNMENT_avoids_quarantined_worker(fleet):
+    store,_,scheduler=fleet
+    store.tasks=[Task("task-1")]
+    scheduler.state.record_task_worker_failure(
+      "task-1","w-a","lease_expired",quarantine_seconds=60,now=1000.0
+    )
+    decision=scheduler.select(["w-a","w-b"],limit=1,now=1001.0)[0]
+    assert decision.worker_id=="w-b"
+
+
+def test_FLEET_SCOPE_ISOLATION_uses_tenant_project_fairness_keys(fleet):
+    store,_,scheduler=fleet
+    store.tasks=[
+      Task("a",task_plan=Plan("a",tenant_id="tenant-a",project_id="p")),
+      Task("b",task_plan=Plan("b",tenant_id="tenant-b",project_id="p")),
+    ]
+    decisions=scheduler.select(["w-b"],limit=2,now=1000.0)
+    assert {d.scope_key for d in decisions}=={
+      "tenant:tenant-a|project:p","tenant:tenant-b|project:p"
+    }
+
+
+def test_FLEET_IDEMPOTENT_DISPATCH_replay_same_task_worker(fleet):
+    store,_,scheduler=fleet
+    store.tasks=[Task("task-1")]
+    one=scheduler.select(["w-b"],limit=1,now=1000.0)[0]
+    # Reservation remains authoritative; a conflicting second worker cannot steal it.
+    with pytest.raises(FleetConflict):
+        scheduler.state.reserve(
+            task_id="task-1",worker_id="w-c",task_revision=1,
+            scope_key=one.scope_key,failure_domain="zone-c",score=one.score,
+            reservation_seconds=30,now=1001.0,scope_weight=1.0,
+        )
+
+
+def test_FLEET_FAILURE_RECOVERY_worker_success_closes_circuit(fleet):
+    store,_,scheduler=fleet
+    for _ in range(3):
+        scheduler.state.record_worker_failure("w-a","runtime_crash",now=1000.0)
+    scheduler.state.record_worker_success("w-a",now=1001.0)
+    assert not scheduler.state.worker_circuit_open("w-a",1001.1)
+
+
+def test_affinity_and_failure_domain_anti_affinity(fleet):
+    store,_,scheduler=fleet
+    scheduler.configure_worker("w-b",max_slots=2,failure_domain="zone-b",labels={"gpu":"false"})
+    scheduler.configure_worker("w-c",max_slots=1,failure_domain="zone-c",labels={"gpu":"true"})
+    task=Task("gpu-task")
+    task.task_plan.metadata={
+      "fleet_affinity_labels":{"gpu":"true"},
+      "fleet_avoid_failure_domains":["zone-b"],
+    }
+    store.tasks=[task]
+    decision=scheduler.select(["w-b","w-c"],limit=1,now=1000.0)[0]
+    assert decision.worker_id=="w-c"
+
+
+def test_FLEET_FAILURE_RECOVERY_state_survives_sqlite_restart(tmp_path):
+    path=tmp_path/"task-plane.sqlite3"
+    first=Store(str(path)); registry=Registry()
+    first.workers["w-a"]=Worker("w-a"); registry.online.add("w-a")
+    scheduler=FleetScheduler(first,registry,now_fn=lambda:1000.0)
+    scheduler.configure_worker("w-a",max_slots=3,failure_domain="zone-a")
+    scheduler.state.record_worker_failure("w-a","runtime_crash",threshold=1,now=1000.0)
+    first.conn.close()
+
+    second=Store(str(path)); second.workers["w-a"]=Worker("w-a")
+    resumed=FleetScheduler(second,registry,now_fn=lambda:1001.0)
+    assert resumed.state.worker_policy("w-a").max_slots==3
+    assert resumed.state.worker_circuit_open("w-a",1001.0)

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