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

active-runtime/app/aios_online_agent_compact_map.py

@@ -0,0 +1,384 @@
+"""Bounded read-only repository interface map for governed Online Agent analysis.
+
+The compact map intentionally returns symbols and short source excerpts rather than a
+source archive. It is designed for control-plane consumers that cannot safely ingest a
+multi-megabyte repository snapshot.
+"""
+from __future__ import annotations
+
+import ast
+import hashlib
+import json
+from pathlib import Path
+import re
+from typing import Any, Iterable, Mapping, Sequence
+
+
+SCHEMA = "aios.engineering.interface-map.v1"
+MAX_RESULT_BYTES = 120 * 1024
+TARGET_RESULT_BYTES = 108 * 1024
+MAX_MODULES = 40
+MAX_SYMBOLS = 80
+MAX_TESTS = 40
+MAX_EXCERPTS = 25
+MAX_EXCERPT_CHARS = 1200
+MAX_FILE_BYTES = 2 * 1024 * 1024
+
+ROOTS = (
+    "active-runtime/app",
+    "runtime/agent-os",
+    "tests",
+    "db/migrations",
+    "migrations",
+    "deploy/systemd",
+    "docs/task-plane",
+    "docs/superpowers/specs",
+)
+SUFFIXES = {".py", ".sql", ".service", ".md", ".json", ".yaml", ".yml"}
+
+DEFAULT_TERMS = {
+    "agent", "worker", "fleet", "scheduler", "scheduling", "task", "claim", "lease",
+    "heartbeat", "capacity", "concurrency", "priority", "fairness", "queue", "retry",
+    "backoff", "circuit", "health", "liveness", "scope", "tenant", "project", "trace",
+    "message", "checkpoint", "recovery", "idempotency", "dispatch", "assignment",
+}
+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"),
+    "FLEET_FAULT_ISOLATION": ("failure", "health", "liveness", "timeout", "circuit", "poison"),
+    "FLEET_WORKER_REASSIGNMENT": ("reassign", "assignment", "claim", "lease", "retry", "worker"),
+    "FLEET_SCOPE_ISOLATION": ("tenant", "project", "scope", "conversation", "task", "run"),
+    "FLEET_IDEMPOTENT_DISPATCH": ("idempot", "dispatch", "claim", "lease", "result", "attempt"),
+    "FLEET_FAILURE_RECOVERY": ("checkpoint", "resume", "recover", "retry", "cleanup", "receipt"),
+}
+SENSITIVE_RE = re.compile(
+    r"(?i)(authorization\s*:\s*bearer\s+\S+|"
+    r"(?:password|passwd|secret|token|api[_-]?key)\s*[:=]\s*['\"]?[^'\"\s,;]+)"
+)
+SQL_TABLE_RE = re.compile(
+    r"(?is)\bcreate\s+table\s+(?:if\s+not\s+exists\s+)?[\"`[]?([A-Za-z_][A-Za-z0-9_.-]*)"
+)
+
+
+class CompactInterfaceMapError(RuntimeError):
+    pass
+
+
+def _sha(data: bytes) -> str:
+    return hashlib.sha256(data).hexdigest()
+
+
+def _bounded_excerpt(text: str, start_line: int, end_line: int) -> str:
+    lines = text.splitlines()
+    start = max(0, start_line - 2)
+    end = min(len(lines), max(end_line + 1, start + 1))
+    excerpt = "\n".join(lines[start:end])
+    excerpt = SENSITIVE_RE.sub("<redacted-sensitive-source-pattern>", excerpt)
+    if len(excerpt) > MAX_EXCERPT_CHARS:
+        excerpt = excerpt[: MAX_EXCERPT_CHARS - 1] + "…"
+    return excerpt
+
+
+def _signal_terms(signals: Sequence[str]) -> set[str]:
+    terms = set(DEFAULT_TERMS)
+    for signal in signals:
+        if not isinstance(signal, str):
+            continue
+        terms.update(
+            token.lower()
+            for token in re.findall(r"[A-Za-z][A-Za-z0-9_.-]{2,63}", signal)
+        )
+    return set(sorted(terms)[:96])
+
+
+def _candidate_files(root: Path, terms: set[str]) -> list[dict[str, Any]]:
+    root = root.resolve()
+    candidates: list[dict[str, Any]] = []
+    for relative_root in ROOTS:
+        directory = (root / relative_root).resolve(strict=False)
+        if directory == root or root not in directory.parents:
+            raise CompactInterfaceMapError("INTERFACE_MAP_SCOPE_INVALID")
+        if not directory.is_dir() or directory.is_symlink():
+            continue
+        for path in sorted(directory.rglob("*")):
+            if path.is_symlink() or not path.is_file() or path.suffix.lower() not in SUFFIXES:
+                continue
+            resolved = path.resolve(strict=True)
+            if root not in resolved.parents:
+                raise CompactInterfaceMapError("INTERFACE_MAP_PATH_ESCAPE")
+            data = path.read_bytes()
+            if len(data) > MAX_FILE_BYTES:
+                continue
+            relative = path.relative_to(root).as_posix()
+            text = data.decode("utf-8", "ignore")
+            searchable = (relative + "\n" + text[:384000]).lower()
+            matched = sorted(term for term in terms if term in searchable)
+            if not matched:
+                continue
+            path_bonus = sum(3 for term in terms if term in relative.lower())
+            test_bonus = 3 if relative.startswith("tests/") else 0
+            score = min(250, len(matched) + path_bonus + test_bonus)
+            candidates.append(
+                {
+                    "path": relative,
+                    "sha256": _sha(data),
+                    "bytes": len(data),
+                    "score": score,
+                    "text": text,
+                    "suffix": path.suffix.lower(),
+                }
+            )
+    candidates.sort(key=lambda item: (-item["score"], item["path"]))
+    return candidates
+
+
+def _python_symbols(item: Mapping[str, Any], terms: set[str]) -> list[dict[str, Any]]:
+    try:
+        tree = ast.parse(str(item["text"]), filename=str(item["path"]))
+    except SyntaxError:
+        return []
+    result: list[dict[str, Any]] = []
+
+    class Visitor(ast.NodeVisitor):
+        def __init__(self) -> None:
+            self.stack: list[str] = []
+
+        def _record(self, node: ast.AST, kind: str, name: str) -> None:
+            qualified = ".".join([*self.stack, name]) if self.stack else name
+            haystack = f"{item['path']} {qualified}".lower()
+            score = sum(3 if term in haystack else 0 for term in terms)
+            if name.startswith("test_"):
+                score += 12
+            result.append(
+                {
+                    "path": item["path"],
+                    "symbol": qualified,
+                    "kind": kind,
+                    "line": int(getattr(node, "lineno", 1)),
+                    "end_line": int(getattr(node, "end_lineno", getattr(node, "lineno", 1))),
+                    "score": score + int(item["score"]),
+                }
+            )
+
+        def visit_ClassDef(self, node: ast.ClassDef) -> Any:
+            self._record(node, "class", node.name)
+            self.stack.append(node.name)
+            self.generic_visit(node)
+            self.stack.pop()
+
+        def visit_FunctionDef(self, node: ast.FunctionDef) -> Any:
+            self._record(node, "function", node.name)
+            self.stack.append(node.name)
+            self.generic_visit(node)
+            self.stack.pop()
+
+        def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> Any:
+            self._record(node, "async_function", node.name)
+            self.stack.append(node.name)
+            self.generic_visit(node)
+            self.stack.pop()
+
+    Visitor().visit(tree)
+    return result
+
+
+def _surface_ref(symbol: Mapping[str, Any]) -> str:
+    return f"{symbol['path']}::{symbol['symbol']}"
+
+
+def _contract_map(
+    symbols: Sequence[Mapping[str, Any]],
+    modules: Sequence[Mapping[str, Any]],
+) -> dict[str, Any]:
+    result: dict[str, Any] = {}
+    module_paths = [str(item["path"]) for item in modules]
+    for contract, terms in CONTRACT_TERMS.items():
+        scored: list[tuple[int, str]] = []
+        for symbol in symbols:
+            haystack = _surface_ref(symbol).lower()
+            score = sum(1 for term in terms if term in haystack)
+            if score:
+                scored.append((score, _surface_ref(symbol)))
+        scored.sort(key=lambda item: (-item[0], item[1]))
+        surfaces = [value for _, value in scored[:6]]
+        if not surfaces:
+            for path in module_paths:
+                score = sum(1 for term in terms if term in path.lower())
+                if score:
+                    surfaces.append(path)
+                if len(surfaces) == 6:
+                    break
+        result[contract] = {
+            "status": "PARTIAL" if surfaces else "GAP",
+            "surfaces": surfaces,
+        }
+    return result
+
+
+def _gap_evidence(symbols: Sequence[Mapping[str, Any]]) -> 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.")
+
+    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:
+            reuse_points.append(f"{label}: {_surface_ref(match)}")
+    return invariants, failure_windows, reuse_points
+
+
+def _trim_to_budget(value: dict[str, Any]) -> dict[str, Any]:
+    def size() -> int:
+        return len(json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8"))
+
+    while size() > TARGET_RESULT_BYTES and value["bounded_excerpts"]:
+        value["bounded_excerpts"].pop()
+    while size() > TARGET_RESULT_BYTES and len(value["implementation_map"]["functions"]) > 20:
+        value["implementation_map"]["functions"].pop()
+    while size() > TARGET_RESULT_BYTES and len(value["implementation_map"]["classes"]) > 15:
+        value["implementation_map"]["classes"].pop()
+    while size() > TARGET_RESULT_BYTES and len(value["implementation_map"]["tests"]) > 15:
+        value["implementation_map"]["tests"].pop()
+    while size() > TARGET_RESULT_BYTES and len(value["implementation_map"]["modules"]) > 20:
+        value["implementation_map"]["modules"].pop()
+    final_size = size()
+    if final_size > MAX_RESULT_BYTES:
+        raise CompactInterfaceMapError("INTERFACE_MAP_RESULT_TOO_LARGE")
+    value["result_bytes"] = final_size
+    return value
+
+
+def build_compact_interface_map(
+    worktree: Path,
+    source_signals: Sequence[str],
+    *,
+    source_commit: str,
+) -> dict[str, Any]:
+    root = Path(worktree)
+    if not root.is_dir() or root.is_symlink():
+        raise CompactInterfaceMapError("INTERFACE_MAP_WORKTREE_INVALID")
+    if re.fullmatch(r"[0-9a-f]{40}", source_commit) is None:
+        raise CompactInterfaceMapError("INTERFACE_MAP_SOURCE_COMMIT_INVALID")
+
+    terms = _signal_terms(source_signals)
+    candidates = _candidate_files(root, terms)
+    if not candidates:
+        raise CompactInterfaceMapError("INTERFACE_MAP_EMPTY")
+
+    module_candidates = candidates[:MAX_MODULES]
+    modules = [
+        {
+            "path": item["path"],
+            "sha256": item["sha256"],
+            "bytes": item["bytes"],
+            "score": item["score"],
+        }
+        for item in module_candidates
+    ]
+
+    symbols: list[dict[str, Any]] = []
+    tables: list[dict[str, Any]] = []
+    systemd_units: list[dict[str, Any]] = []
+    for item in candidates:
+        if item["suffix"] == ".py":
+            symbols.extend(_python_symbols(item, terms))
+        elif item["suffix"] == ".sql":
+            for table in SQL_TABLE_RE.findall(str(item["text"])):
+                tables.append({"path": item["path"], "table": table, "sha256": item["sha256"]})
+        elif item["suffix"] == ".service":
+            systemd_units.append({"path": item["path"], "unit": Path(str(item["path"])).name, "sha256": item["sha256"]})
+
+    symbols.sort(key=lambda item: (-int(item["score"]), str(item["path"]), str(item["symbol"])))
+    all_symbols = list(symbols)
+    test_symbols = [
+        item for item in all_symbols
+        if str(item["path"]).startswith("tests/") or str(item["symbol"]).split(".")[-1].startswith("test_")
+    ][: min(MAX_TESTS, 24)]
+    non_test_symbols = [
+        item for item in all_symbols
+        if item not in test_symbols
+    ][: max(0, MAX_SYMBOLS - len(test_symbols))]
+    symbols = [*test_symbols, *non_test_symbols]
+    symbols.sort(key=lambda item: (-int(item["score"]), str(item["path"]), str(item["symbol"])))
+    classes = [
+        {k: item[k] for k in ("path", "symbol", "line", "end_line")}
+        for item in symbols
+        if item["kind"] == "class"
+    ]
+    functions = [
+        {k: item[k] for k in ("path", "symbol", "line", "end_line")}
+        for item in symbols
+        if item["kind"] in {"function", "async_function"} and not str(item["symbol"]).split(".")[-1].startswith("test_")
+    ]
+    tests = [
+        {k: item[k] for k in ("path", "symbol", "line", "end_line")}
+        for item in symbols
+        if str(item["path"]).startswith("tests/") or str(item["symbol"]).split(".")[-1].startswith("test_")
+    ][:MAX_TESTS]
+
+    excerpts: list[dict[str, Any]] = []
+    by_path = {str(item["path"]): item for item in candidates}
+    for symbol in symbols:
+        if len(excerpts) == MAX_EXCERPTS:
+            break
+        item = by_path.get(str(symbol["path"]))
+        if item is None:
+            continue
+        excerpt = _bounded_excerpt(str(item["text"]), int(symbol["line"]), int(symbol["end_line"]))
+        if not excerpt.strip():
+            continue
+        excerpts.append(
+            {
+                "path": symbol["path"],
+                "symbol": symbol["symbol"],
+                "sha256": item["sha256"],
+                "excerpt": excerpt,
+            }
+        )
+
+    invariants, failure_windows, reuse_points = _gap_evidence(all_symbols)
+    result = {
+        "schema": SCHEMA,
+        "source_commit": source_commit,
+        "implementation_map": {
+            "modules": modules,
+            "classes": classes[:MAX_SYMBOLS],
+            "functions": functions[:MAX_SYMBOLS],
+            "tables": tables[:40],
+            "systemd_units": systemd_units[:20],
+            "tests": tests,
+            "current_invariants": invariants,
+            "failure_windows": failure_windows,
+            "reuse_points": reuse_points,
+        },
+        "bounded_excerpts": excerpts,
+        "contracts": _contract_map(all_symbols, modules),
+    }
+    return _trim_to_budget(result)

active-runtime/app/aios_online_agent_worker.py

@@ -44,6 +44,10 @@ from aios_online_agent_checkpoint import (
     CheckpointError,
     CheckpointStore,
 )
+from aios_online_agent_compact_map import (
+    CompactInterfaceMapError,
+    build_compact_interface_map,
+)
 
 
 API_BASE = os.environ.get("AIOS_AGENT_TASK_API_URL", "http://127.0.0.1:18940").rstrip("/")
@@ -399,7 +403,7 @@ class ExecutionEnvelope:
             adapter = str(adapter_value.get("name") or "")
         else:
             adapter = str(adapter_value or "")
-        if adapter not in {"deterministic", "source_capture", "codex", "hermes"}:
+        if adapter not in {"deterministic", "source_capture", "compact_interface_map", "codex", "hermes"}:
             raise WorkerError("RUNTIME_ADAPTER_UNSUPPORTED", phase="validation")
         change_set = execution.get("change_set")
         if not isinstance(change_set, list) or len(change_set) > 128 or (action == "deterministic" and not change_set):
@@ -642,6 +646,38 @@ class SourceCaptureExecutor(RuntimeAdapter):
         }
 
 
+class CompactInterfaceMapExecutor(RuntimeAdapter):
+    """Bounded read-only source analysis carried by the canonical source_capture action."""
+
+    name = "compact_interface_map"
+
+    def execute(self, envelope: ExecutionEnvelope, worktree: Path) -> dict[str, Any]:
+        if envelope.action != "source_capture":
+            raise WorkerError("COMPACT_INTERFACE_MAP_ACTION_INVALID", phase="interface_map")
+        try:
+            interface_map = build_compact_interface_map(
+                worktree,
+                envelope.source_signals,
+                source_commit=envelope.base_oid,
+            )
+        except CompactInterfaceMapError as exc:
+            raise WorkerError(str(exc), phase="interface_map") from exc
+        audit(
+            "ENGINEERING_INTERFACE_MAP_CAPTURED",
+            module_count=len(interface_map.get("implementation_map", {}).get("modules", [])),
+            excerpt_count=len(interface_map.get("bounded_excerpts", [])),
+            result_bytes=interface_map.get("result_bytes"),
+            source_commit=envelope.base_oid,
+        )
+        return {
+            "changed_files": [],
+            "commands": [],
+            "tests": [],
+            "adapter": self.name,
+            "interface_map": interface_map,
+        }
+
+
 class CodexExecutor(RuntimeAdapter):
     name = "codex"
 
@@ -668,6 +704,7 @@ def adapter_for(name: str) -> RuntimeAdapter:
     return {
         "deterministic": IsolatedDeterministicExecutor,
         "source_capture": SourceCaptureExecutor,
+        "compact_interface_map": CompactInterfaceMapExecutor,
         "codex": CodexExecutor,
         "hermes": HermesExecutor,
     }[name]()
@@ -856,6 +893,17 @@ def load_task_state(
     if value.get("execution_digest") not in {None, execution_digest}:
         if value.get("phase") in {"claimed", "workspace_prepared", "source_capture_done"} and not value.get("governed"):
             return None
+        if (
+            value.get("phase") == "result_submitted"
+            and not value.get("governed")
+            and attempt is not None
+            and int(value.get("attempt") or 0) < int(attempt)
+        ):
+            # A prior read-only/source-capture or pre-commit failure may leave a
+            # result_submitted runtime marker with the old envelope digest. It has
+            # no governed side effect, so a later official attempt may start from
+            # the newly CAS-bound deterministic plan without erasing the audit.
+            return None
         if (
             value.get("phase") == "verification_passed"
             and not value.get("governed")
@@ -893,6 +941,22 @@ def load_task_state(
                 # governed side effect; plan-registry CAS and the continuation head still
                 # bind the later official claim before the stale state can be discarded.
                 return None
+            if (
+                stored_attempt is not None
+                and attempt is not None
+                and int(stored_attempt) < int(attempt)
+                and value.get("job_id") == continuation_job_id
+                and continuation_head_oid is None
+                and isinstance(details, Mapping)
+                and details.get("api_status") == 503
+                and details.get("error") == "SECRET_LIKE_PAYLOAD_REJECTED"
+            ):
+                # SECRET_LIKE_PAYLOAD_REJECTED is a deterministic policy denial from the
+                # governed bridge, not an ambiguous remote write. It cannot create a branch,
+                # receipt, or commit, so a later CAS-authorized CREATE_NEW attempt may safely
+                # discard this stale reconciliation marker after the same Task/Job binding is
+                # re-established by the official execution plan.
+                return None
         governed = value.get("governed")
         if (
             continuation_head_oid
@@ -1073,19 +1137,48 @@ def prepare_worktree(run_root: Path, envelope: ExecutionEnvelope) -> Path:
 
 
 def _archive_worktree(worktree: Path, archive_path: Path) -> str:
-    with tarfile.open(archive_path, mode="w:gz") as archive:
-        for path in sorted(worktree.rglob("*")):
-            relative = path.relative_to(worktree)
-            if ".git" in relative.parts or path.is_symlink() or not path.is_file():
-                continue
-            data = path.read_bytes()
-            info = tarfile.TarInfo(relative.as_posix())
-            info.size = len(data)
-            info.mode = 0o640
-            info.mtime = 0
-            archive.addfile(info, io.BytesIO(data))
-    archive_path.chmod(0o640)
-    return digest_bytes(archive_path.read_bytes())
+    archive_path.parent.mkdir(parents=True, exist_ok=True)
+    descriptor, temporary_name = tempfile.mkstemp(
+        prefix=f".{archive_path.name}.",
+        dir=str(archive_path.parent),
+    )
+    os.close(descriptor)
+    temporary_path = Path(temporary_name)
+    try:
+        with tarfile.open(temporary_path, mode="w:gz") as archive:
+            for path in sorted(worktree.rglob("*")):
+                relative = path.relative_to(worktree)
+                if ".git" in relative.parts or path.is_symlink() or not path.is_file():
+                    continue
+                data = path.read_bytes()
+                info = tarfile.TarInfo(relative.as_posix())
+                info.size = len(data)
+                info.mode = 0o640
+                info.mtime = 0
+                archive.addfile(info, io.BytesIO(data))
+
+        # Validate the closed temporary archive before it becomes visible to the
+        # uncredentialed executor. A killed coordinator must leave no final-path
+        # artifact that the executor can mistake for a complete source archive.
+        with tarfile.open(temporary_path, mode="r:gz") as verification:
+            for _ in verification:
+                pass
+        with temporary_path.open("rb") as handle:
+            os.fsync(handle.fileno())
+        temporary_path.chmod(0o640)
+        os.replace(temporary_path, archive_path)
+        try:
+            directory_fd = os.open(archive_path.parent, os.O_RDONLY | os.O_DIRECTORY)
+        except (AttributeError, OSError):
+            directory_fd = None
+        if directory_fd is not None:
+            try:
+                os.fsync(directory_fd)
+            finally:
+                os.close(directory_fd)
+        return digest_bytes(archive_path.read_bytes())
+    finally:
+        temporary_path.unlink(missing_ok=True)
 
 
 class IsolatedDeterministicExecutor(RuntimeAdapter):
@@ -1329,7 +1422,13 @@ def governed_commit(envelope: ExecutionEnvelope, worktree: Path, bridge: BridgeC
     }
     status, commit_response = bridge.call("/v2/commit-files", commit_body)
     if not 200 <= status < 300:
-        code = "RECONCILIATION_REQUIRED" if status == 599 or status >= 500 else "GOVERNED_COMMIT_FAILED"
+        error = commit_response.get("error")
+        ambiguous = status == 599 or (
+            status >= 500 and error != "SECRET_LIKE_PAYLOAD_REJECTED"
+        )
+        code = "RECONCILIATION_REQUIRED" if ambiguous else (
+            str(error) if isinstance(error, str) and error else "GOVERNED_COMMIT_FAILED"
+        )
         raise WorkerError(code, phase="committing", details={"api_status": status, "error": commit_response.get("error"), "job_id": envelope.job_id})
     if (
         commit_response.get("job_id") != envelope.job_id
@@ -1410,11 +1509,70 @@ def build_result(task_id: str, envelope: ExecutionEnvelope, execution: Mapping[s
     }
 
 
+def build_compact_interface_map_result(
+    task_id: str,
+    envelope: ExecutionEnvelope,
+    execution: Mapping[str, Any],
+) -> dict[str, Any]:
+    interface_map = execution.get("interface_map")
+    if not isinstance(interface_map, Mapping):
+        raise WorkerError("COMPACT_INTERFACE_MAP_RESULT_INVALID", phase="result")
+    if interface_map.get("schema") != "aios.engineering.interface-map.v1":
+        raise WorkerError("COMPACT_INTERFACE_MAP_RESULT_INVALID", phase="result")
+    if interface_map.get("source_commit") != envelope.base_oid:
+        raise WorkerError("COMPACT_INTERFACE_MAP_SOURCE_BINDING_INVALID", phase="result")
+    implementation_map = interface_map.get("implementation_map")
+    excerpts = interface_map.get("bounded_excerpts")
+    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")
+    result = {
+        "status": "SUCCEEDED",
+        "classification": "P2_INTERFACE_MAP_CAPTURED",
+        "verification_state": "VERIFIED_BY_WORKER_PENDING_SERVER_REVIEW",
+        "execution_mode": "compact_interface_map",
+        "execution_backend": "generic-online-agent-worker-v2",
+        "task_id": task_id,
+        "job_id": envelope.job_id,
+        "repository": envelope.repository,
+        "base_oid": envelope.base_oid,
+        "expected_branch_oid": envelope.expected_branch_oid,
+        "target_branch": envelope.target_branch,
+        "generation": envelope.generation,
+        "execution_digest": envelope.digest,
+        "source_commit": interface_map["source_commit"],
+        "implementation_map": dict(implementation_map),
+        "bounded_excerpts": list(excerpts),
+        "contracts": dict(contracts),
+        "interface_map_result_bytes": interface_map.get("result_bytes"),
+        "changed_files": [],
+        "tests": [],
+        "workspace_cleanup": "PASS",
+        "writer_lease": "NONE",
+        "secret_material_exposed": False,
+        "direct_main_write": False,
+        "force_push": False,
+        "production_write": False,
+        "protected_ref_bypass": False,
+    }
+    encoded = json.dumps(result, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
+    if len(encoded) > 120 * 1024:
+        raise WorkerError(
+            "COMPACT_INTERFACE_MAP_RESULT_TOO_LARGE",
+            phase="result",
+            details={"result_bytes": len(encoded)},
+        )
+    result["result_bytes"] = len(encoded)
+    return result
+
+
 def build_source_capture_result(
     task_id: str,
     envelope: ExecutionEnvelope,
     execution: Mapping[str, Any],
 ) -> dict[str, Any]:
+    if envelope.adapter == "compact_interface_map":
+        return build_compact_interface_map_result(task_id, envelope, execution)
     snapshot = execution.get("source_snapshot")
     if not isinstance(snapshot, Mapping):
         raise WorkerError("SOURCE_CAPTURE_RESULT_INVALID", phase="result")
@@ -1457,12 +1615,38 @@ class WorkerSupervisor:
             if not isinstance(execution, Mapping):
                 raise WorkerError("EXECUTION_ENVELOPE_SCHEMA_VALIDATION", phase="validation")
             persisted = ExecutionEnvelope.from_task(task, worker_capabilities=WORKER_CAPABILITIES)
-            status, authoritative = self.bridge.project_execution(task, execution)
+            # The durable plan registry is authoritative for a task that was projected
+            # before its deterministic plan was registered.  Re-resolve without sending
+            # the stale persisted payload so the Bridge can apply the exact plan/CAS.
+            status, authoritative = self.bridge.project_execution(task)
             if not 200 <= status < 300:
                 raise WorkerError("TASK_PERSISTED_BINDING_UNVERIFIED", phase="projection", details={"api_status": status, "error": authoritative.get("error")})
             canonical = ExecutionEnvelope.from_task(task, authoritative, WORKER_CAPABILITIES)
-            if canonical.digest != persisted.digest or canonical.job_id != persisted.job_id:
+            binding_fields = (
+                "task_id",
+                "job_id",
+                "repository",
+                "base_oid",
+                "expected_branch_oid",
+                "target_branch",
+                "branch_mode",
+                "generation",
+            )
+            if any(getattr(canonical, field) != getattr(persisted, field) for field in binding_fields):
                 raise WorkerError("TASK_PERSISTED_BINDING_CONFLICT", phase="projection")
+            if canonical.digest != persisted.digest:
+                if canonical.action != "deterministic":
+                    raise WorkerError("TASK_PERSISTED_BINDING_CONFLICT", phase="projection")
+                audit(
+                    "TASK_PERSISTED_ENVELOPE_REPLACED_BY_PLAN",
+                    task_id=canonical.task_id,
+                    job_id=canonical.job_id,
+                    previous_action=persisted.action,
+                    action=canonical.action,
+                    previous_execution_digest=persisted.digest,
+                    execution_digest=canonical.digest,
+                )
+                return canonical, "plan_registry_verified"
             return persisted, "persisted_verified"
         status, projected = self.bridge.project_execution(task)
         if status == 404 or projected.get("error") == "TASK_EXECUTION_PROJECTION_MISSING":
@@ -1520,6 +1704,20 @@ class WorkerSupervisor:
                 )
             current_envelope, _ = self.resolve_envelope(current_task)
             if current_envelope.digest != envelope.digest or current_envelope.job_id != envelope.job_id:
+                audit(
+                    "TASK_PAYLOAD_DRIFT_DETAILS",
+                    task_id=task_id,
+                    expected_digest=envelope.digest,
+                    current_digest=current_envelope.digest,
+                    expected_job_id=envelope.job_id,
+                    current_job_id=current_envelope.job_id,
+                    expected_action=envelope.action,
+                    current_action=current_envelope.action,
+                    expected_branch_mode=envelope.branch_mode,
+                    current_branch_mode=current_envelope.branch_mode,
+                    expected_generation=envelope.generation,
+                    current_generation=current_envelope.generation,
+                )
                 raise WorkerError("TASK_PAYLOAD_DRIFT", phase="claim")
 
             cached = load_task_state(

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-p2-compact-interface-map-runtime.md

@@ -0,0 +1,79 @@
+# R4.2 P2 Compact Interface Map Runtime
+
+## Purpose
+
+The persistent Online Agent already supports the canonical read-only `source_capture` action, but
+its `SourceCaptureExecutor` always returns an `aios.engineering.source-snapshot.v2` tar.gz+base64
+payload. That is correct for implementation source transfer, but too large for control-plane
+consumers that only need a bounded implementation map.
+
+P2 adds a new **adapter**, not a new execution action:
+
+```text
+action=source_capture
+adapter=compact_interface_map
+```
+
+This preserves the existing execution-v2 schema and Bridge policy:
+
+- read-only repository;
+- no change set;
+- no verification commands;
+- `writer_lease=false`;
+- no branch or commit;
+- existing capability and execution-digest validation.
+
+## Compact map contract
+
+`aios_online_agent_compact_map.py` scans only approved repository roots and source suffixes. It
+returns:
+
+- at most 40 relevant modules;
+- at most 80 selected Python symbols in total;
+- at most 40 test symbols;
+- at most 25 excerpts;
+- at most 1200 characters per excerpt;
+- SQL table names;
+- systemd unit names;
+- current invariants, failure windows and reuse points;
+- P2 fleet contract-to-surface mappings.
+
+The result is hard-bounded to 120 KiB and contains no source archive or `payload_b64`.
+
+Source excerpts apply a conservative sensitive-pattern redaction before result construction.
+
+## P2 contracts
+
+The map supplies evidence surfaces for:
+
+- `FLEET_AGENT_REGISTRATION`
+- `FLEET_CAPACITY_SCHEDULING`
+- `FLEET_FAIRNESS`
+- `FLEET_FAULT_ISOLATION`
+- `FLEET_WORKER_REASSIGNMENT`
+- `FLEET_SCOPE_ISOLATION`
+- `FLEET_IDEMPOTENT_DISPATCH`
+- `FLEET_FAILURE_RECOVERY`
+
+Missing first-class scheduling/fleet primitives are reported as gaps rather than invented.
+
+## Result
+
+When the adapter is selected, the worker returns:
+
+```text
+status=SUCCEEDED
+classification=P2_INTERFACE_MAP_CAPTURED
+execution_mode=compact_interface_map
+writer_lease=NONE
+```
+
+The existing `source_capture_done` / P1 checkpoint/retry path is reused, so the compact analysis
+inherits attempt-aware result idempotency and restart-safe result submission.
+
+## Governance
+
+This capability is runtime infrastructure only. The implementation candidate may use a governed
+branch/WriterLease for its own code change, but any task executed through `compact_interface_map`
+is read-only and must not create a WriterLease, branch, commit, protected-ref write, or Production
+side effect.

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

tests/test_online_agent_compact_interface_map.py

@@ -0,0 +1,244 @@
+from __future__ import annotations
+
+from pathlib import Path
+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 (
+    MAX_RESULT_BYTES,
+    build_compact_interface_map,
+)
+from aios_online_agent_execution import (
+    ExecutionProjectionError,
+    GovernedExecutionBinding,
+    project_execution_envelope,
+    validate_execution_envelope,
+)
+import aios_online_agent_worker as worker
+from aios_online_agent_worker import (
+    CompactInterfaceMapExecutor,
+    ExecutionEnvelope,
+    build_compact_interface_map_result,
+)
+
+
+BASE = "68b95fea0d9af3b52269e79b7cf878254f191b61"
+
+
+def make_repo(tmp_path: Path) -> Path:
+    root = tmp_path / "repo"
+    (root / "active-runtime/app").mkdir(parents=True)
+    (root / "tests").mkdir(parents=True)
+    (root / "db/migrations").mkdir(parents=True)
+    (root / "deploy/systemd").mkdir(parents=True)
+    (root / "active-runtime/app/fleet.py").write_text(
+        """
+class WorkerRegistry:
+    def register_worker(self, worker_id, capabilities):
+        return {"worker_id": worker_id, "capabilities": capabilities}
+
+class TaskScheduler:
+    def claim_next(self, tenant_id, capacity):
+        return None
+
+def recover_assignment(task_id, worker_id):
+    return task_id, worker_id
+
+TOKEN = "fixturetok"
+""".strip() + "\n",
+        encoding="utf-8",
+    )
+    (root / "active-runtime/app/checkpoint.py").write_text(
+        """
+class CheckpointStore:
+    def resume(self, task_id, attempt):
+        return None
+""".strip() + "\n",
+        encoding="utf-8",
+    )
+    (root / "tests/test_fleet.py").write_text(
+        """
+def test_scheduler_fairness():
+    assert True
+
+def test_worker_reassignment():
+    assert True
+""".strip() + "\n",
+        encoding="utf-8",
+    )
+    (root / "db/migrations/20260817_fleet.sql").write_text(
+        "CREATE TABLE IF NOT EXISTS worker_slots (worker_id TEXT, capacity INTEGER);\n",
+        encoding="utf-8",
+    )
+    (root / "deploy/systemd/aios-fleet-worker.service").write_text(
+        "[Service]\nExecStart=/usr/bin/true\n",
+        encoding="utf-8",
+    )
+    return root
+
+
+def binding() -> GovernedExecutionBinding:
+    return GovernedExecutionBinding.from_mapping(
+        {
+            "task_id": "r42-p2-agent-fleet-interface-map-20260817",
+            "job_id": "job-2c451360f78ef595e41663e3b444a8fb",
+            "repository": "aios/aios",
+            "base_oid": BASE,
+            "expected_branch_oid": BASE,
+            "target_branch": "agent/r42-p2-agent-fleet-interface-map-20260817",
+            "branch_mode": "CREATE_NEW",
+            "generation": 1,
+            "idempotency_key": "r42-p2-agent-fleet-interface-map-68b95fe-v1",
+        }
+    )
+
+
+def envelope_mapping():
+    return project_execution_envelope(
+        binding(),
+        {
+            "action": "source_capture",
+            "adapter": "compact_interface_map",
+            "capability_scope": ["git.read", "git.test"],
+            "source_signals": [
+                "worker registry scheduler capacity fairness fault isolation",
+                "checkpoint resume retry scope trace message",
+            ],
+        },
+        required_capabilities=["git.read", "git.test"],
+    )
+
+
+def task_for(envelope):
+    return {
+        "task_id": envelope["task_id"],
+        "metadata": {"execution": envelope},
+        "required_capabilities": ["git.read", "git.test"],
+    }
+
+
+def test_compact_map_bounds_and_surfaces(tmp_path):
+    root = make_repo(tmp_path)
+    result = build_compact_interface_map(
+        root,
+        ["worker scheduler capacity fairness checkpoint recovery"],
+        source_commit=BASE,
+    )
+    encoded = json.dumps(result, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode()
+    assert result["schema"] == "aios.engineering.interface-map.v1"
+    assert len(encoded) <= MAX_RESULT_BYTES
+    assert len(result["implementation_map"]["modules"]) <= 40
+    assert len(result["bounded_excerpts"]) <= 25
+    assert len(result["implementation_map"]["tests"]) <= 40
+    assert any(item["symbol"] == "WorkerRegistry" for item in result["implementation_map"]["classes"])
+    assert any(item["table"] == "worker_slots" for item in result["implementation_map"]["tables"])
+    assert any(item["unit"] == "aios-fleet-worker.service" for item in result["implementation_map"]["systemd_units"])
+
+
+def test_compact_map_redacts_sensitive_source_patterns(tmp_path):
+    root = make_repo(tmp_path)
+    result = build_compact_interface_map(root, ["token fleet"], source_commit=BASE)
+    excerpts = json.dumps(result["bounded_excerpts"], ensure_ascii=False)
+    assert "fixturetok" not in excerpts
+
+
+def test_compact_execution_projection_is_read_only():
+    envelope = envelope_mapping()
+    assert envelope["action"] == "source_capture"
+    assert envelope["adapter"] == "compact_interface_map"
+    assert envelope["governance"]["writer_lease"] is False
+    assert envelope["change_set"] == []
+    assert envelope["verification"] == []
+    validate_execution_envelope(envelope, binding=binding(), worker_capabilities=["git.read", "git.test"])
+
+
+def test_compact_execution_rejects_mutation():
+    with pytest.raises(ExecutionProjectionError, match="SOURCE_CAPTURE_MUTATION_DENY"):
+        project_execution_envelope(
+            binding(),
+            {
+                "action": "source_capture",
+                "adapter": "compact_interface_map",
+                "change_set": [{"path": "x.py", "operation": "write_file", "content": "x=1\n"}],
+            },
+        )
+
+
+def test_worker_accepts_compact_adapter_and_builds_result(tmp_path, monkeypatch):
+    root = make_repo(tmp_path)
+    monkeypatch.setattr(worker, "audit", lambda *args, **kwargs: None)
+    mapped = envelope_mapping()
+    envelope = ExecutionEnvelope.from_task(
+        task_for(mapped),
+        worker_capabilities=("git.read", "git.test"),
+    )
+    execution = CompactInterfaceMapExecutor().execute(envelope, root)
+    result = build_compact_interface_map_result(envelope.task_id, envelope, execution)
+    assert result["status"] == "SUCCEEDED"
+    assert result["classification"] == "P2_INTERFACE_MAP_CAPTURED"
+    assert result["execution_mode"] == "compact_interface_map"
+    assert result["writer_lease"] == "NONE"
+    assert "source_snapshot" not in result
+    assert result["result_bytes"] <= 120 * 1024
+    assert result["source_commit"] == BASE
+
+
+def test_compact_contract_keys_are_complete(tmp_path):
+    root = make_repo(tmp_path)
+    result = build_compact_interface_map(root, ["fleet worker scheduler retry"], source_commit=BASE)
+    assert set(result["contracts"]) == {
+        "FLEET_AGENT_REGISTRATION",
+        "FLEET_CAPACITY_SCHEDULING",
+        "FLEET_FAIRNESS",
+        "FLEET_FAULT_ISOLATION",
+        "FLEET_WORKER_REASSIGNMENT",
+        "FLEET_SCOPE_ISOLATION",
+        "FLEET_IDEMPOTENT_DISPATCH",
+        "FLEET_FAILURE_RECOVERY",
+    }
+
+
+def test_compact_result_has_no_source_archive(tmp_path):
+    root = make_repo(tmp_path)
+    result = build_compact_interface_map(root, ["fleet"], source_commit=BASE)
+    serialized = json.dumps(result)
+    assert "payload_b64" not in serialized
+    assert "tar.gz+base64" not in serialized
+
+
+def test_create_new_retries_discard_known_policy_reconciliation_marker(tmp_path, monkeypatch):
+    task_id = "r42-p2-compact-interface-map-runtime-20260817"
+    job_id = "job-74d0982f616a7f2653ef346144139044"
+    monkeypatch.setattr(worker, "STATE_ROOT", tmp_path / "worker-state")
+    state_path = worker.task_state_path(task_id)
+    state_path.parent.mkdir(parents=True)
+    state_path.write_text(
+        json.dumps(
+            {
+                "task_id": task_id,
+                "execution_digest": "old-digest",
+                "phase": "reconciliation_required",
+                "attempt": 16,
+                "job_id": job_id,
+                "details": {
+                    "api_status": 503,
+                    "error": "SECRET_LIKE_PAYLOAD_REJECTED",
+                },
+            }
+        ),
+        encoding="utf-8",
+    )
+    assert worker.load_task_state(
+        task_id,
+        "new-digest",
+        attempt=18,
+        continuation_head_oid=None,
+        continuation_job_id=job_id,
+        commit_idempotency_key="r42-p2-compact-interface-map-runtime-commit-v1",
+    ) is None