feat(online-agent): generalize compact interface-map profiles

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

@@ -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,203 @@ 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
+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
+        )

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-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.

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_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_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},
+        )