ACTIVITY-WP-0043-T05: owner-return eligibility builder bound to scan revision

Build the planner's current_tasks from immutable Forge commit content at the
C-25 scan SHA (no checkouts, ADR-002). Fails closed on revision mismatch,
non-head scans, missing/malformed fields and duplicate identities; unresolved
predecessors and decisions stay unknown (None).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 692292@bnt-lap001
Assistant-Session: 89c7d8e8-e1b3-47eb-9411-a0ab227fca2f

docs/owner-return-pickup-activation-plan.md

@@ -21,6 +21,8 @@ STATE-WP-0094-T03. Nothing below has been applied.
 2. A caller that builds `current_tasks` from fresh authoritative source bound to
    the scanned repository revision, resolving every predecessor and decision
    (contract: "Current eligibility guard"). No manufactured empty predecessor lists.
+   Source builder: `activity_core.owner_return_eligibility.build_current_tasks`
+   (ACTIVITY-WP-0043-T05); the C-25 caller must pass the scanned commit SHA.
 3. The caller runs inside the existing supported runner (Custodian dependency
    monitor / established Activity Core worker); no new schedule is introduced.
 4. Founder review of this plan and of the definition/grant above.

src/activity_core/owner_return_eligibility.py

@@ -0,0 +1,474 @@
+"""Eligibility input builder for owner-return pickup (ACTIVITY-WP-0043-T05).
+
+Builds the ``current_tasks`` mapping consumed by State Hub's offline planner
+(``state-hub/scripts/owner_return_handoff.py::plan_owner_returns``) from fresh
+authoritative source, per the contract in ``state-hub/docs/owner-return-handoff.md``
+("Current eligibility guard").
+
+Source binding
+--------------
+The caller supplies ``scan_revision``: the full 40-hex commit SHA that the C-25
+consistency scan read. Every workplan file is read from the Forge at exactly
+that commit (``/contents`` and ``/raw`` with ``ref=<sha>``), i.e. immutable
+commit content -- never a workstation checkout (ADR-002) and never a branch
+name that could move between reads. Before reading, the builder verifies that
+the Forge knows the commit (``/git/commits/<sha>`` echoes the same SHA) and, by
+default, that it is still the default-branch HEAD (``require_current_head``):
+a scan of an older revision is stale and refuses rather than plan from it.
+
+Resolution rules (fail closed)
+------------------------------
+* Missing or malformed authoritative fields (task ``status``, workplan
+  ``status``, non-boolean ``needs_human``, bad ``wait_until``, malformed
+  ``depends_on``) raise :class:`EligibilityError` -- no partial mapping.
+* Duplicate record identities at the revision raise (no winner is chosen).
+* Predecessors: task-block ``depends_on`` plus the owning workplan's
+  frontmatter dependencies (a workplan-level gate applies to all its tasks).
+  Targets are resolved from the same commit; cross-repository targets only
+  through an explicit ``external_status`` resolver. A target that cannot be
+  resolved is recorded as ``None`` (unknown) -- never dropped, never replaced
+  by an empty list.
+* Decisions are hub-native records: ``decision_status`` comes from
+  ``decision_status`` (e.g. State Hub ``GET /decisions/{id}``); lookup failure
+  yields ``None`` (unknown), which the planner stops on.
+
+Nothing here writes, schedules or dispatches. The HTTP client is injected; the
+caller owns any read credential.
+"""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass, field
+from datetime import date
+from typing import Any, Callable, Mapping, Sequence
+from urllib.parse import quote
+
+import httpx
+import yaml
+
+SCHEMA = "activity-core.owner-return-eligibility.v1"
+DEFAULT_FORGE_API = "https://forgejo.coulomb.social/api/v1"
+DEFAULT_FORGE_OWNER = "coulomb"
+WORKPLAN_DIRS = ("workplans", "workplans/archived")
+
+_SHA = re.compile(r"^[0-9a-f]{40}$")
+_REPO = re.compile(r"^[a-z0-9][a-z0-9._-]{0,127}$")
+_TASK_BLOCK_RE = re.compile(r"```task\s*\n(.*?)\n```", re.DOTALL)
+_SHORT_TASK_RE = re.compile(r"^T\d{2,3}$", re.IGNORECASE)
+_DATE = re.compile(r"^\d{4}-\d{2}-\d{2}$")
+_EMPTY_FLAVOR = {"", "~", "null", "none", "nil"}
+
+StatusResolver = Callable[[str], "str | None"]
+
+
+class EligibilityError(ValueError):
+    """Authoritative eligibility could not be established; refuse planning."""
+
+
+# --------------------------------------------------------------------------- #
+# Forge reads pinned to one commit
+# --------------------------------------------------------------------------- #
+
+
+class ForgeCommitReader:
+    """Read-only Forgejo API access pinned to a single immutable commit."""
+
+    def __init__(
+        self,
+        client: httpx.Client,
+        repo: str,
+        *,
+        owner: str = DEFAULT_FORGE_OWNER,
+        api_base: str = DEFAULT_FORGE_API,
+    ) -> None:
+        if not _REPO.fullmatch(repo):
+            raise EligibilityError("invalid repository identity")
+        self._client = client
+        self._repo_url = f"{api_base.rstrip('/')}/repos/{quote(owner)}/{quote(repo)}"
+
+    def _get(self, path: str, **params: str) -> httpx.Response:
+        try:
+            response = self._client.get(f"{self._repo_url}{path}", params=params or None)
+        except httpx.HTTPError as exc:
+            raise EligibilityError(f"forge read failed: {type(exc).__name__}") from exc
+        return response
+
+    def _json(self, path: str, **params: str) -> Any:
+        response = self._get(path, **params)
+        if response.status_code != 200:
+            raise EligibilityError(f"forge read {path} returned HTTP {response.status_code}")
+        try:
+            return response.json()
+        except ValueError as exc:
+            raise EligibilityError(f"forge read {path} returned invalid JSON") from exc
+
+    def verify_commit(self, sha: str) -> None:
+        body = self._json(f"/git/commits/{sha}")
+        if not isinstance(body, Mapping) or body.get("sha") != sha:
+            raise EligibilityError("forge commit does not match scan revision")
+
+    def default_branch_head(self) -> str:
+        repo = self._json("")
+        branch = repo.get("default_branch") if isinstance(repo, Mapping) else None
+        if not isinstance(branch, str) or not branch:
+            raise EligibilityError("forge repository has no default branch")
+        body = self._json(f"/branches/{quote(branch, safe='')}")
+        commit = body.get("commit") if isinstance(body, Mapping) else None
+        head = commit.get("id") if isinstance(commit, Mapping) else None
+        if not isinstance(head, str) or not _SHA.fullmatch(head):
+            raise EligibilityError("forge default-branch head is not a commit SHA")
+        return head
+
+    def list_markdown(self, directory: str, sha: str) -> list[str] | None:
+        """Return ``.md`` file paths in ``directory`` at ``sha``; None if absent."""
+        response = self._get(f"/contents/{quote(directory)}", ref=sha)
+        if response.status_code == 404:
+            return None
+        if response.status_code != 200:
+            raise EligibilityError(f"forge listing {directory} returned HTTP {response.status_code}")
+        entries = response.json()
+        if not isinstance(entries, list):
+            raise EligibilityError(f"forge listing {directory} is not a directory")
+        paths = []
+        for entry in entries:
+            if not isinstance(entry, Mapping):
+                raise EligibilityError("malformed forge directory entry")
+            if entry.get("type") == "file" and str(entry.get("name", "")).endswith(".md"):
+                paths.append(str(entry["path"]))
+        return sorted(paths)
+
+    def read_text(self, path: str, sha: str) -> str:
+        response = self._get(f"/raw/{quote(path)}", ref=sha)
+        if response.status_code != 200:
+            raise EligibilityError(f"forge raw {path} returned HTTP {response.status_code}")
+        return response.text
+
+
+# --------------------------------------------------------------------------- #
+# Source parsing (mirrors State Hub forge_projection semantics, strictly)
+# --------------------------------------------------------------------------- #
+
+
+@dataclass
+class _Task:
+    record_id: str
+    raw_id: str
+    workplan_id: str
+    block: dict[str, Any]
+
+
+@dataclass
+class _Workplan:
+    record_id: str
+    path: str
+    meta: dict[str, Any]
+    tasks: list[_Task] = field(default_factory=list)
+
+
+def _split_frontmatter(text: str, path: str) -> tuple[dict[str, Any], str]:
+    if not text.startswith("---"):
+        raise EligibilityError(f"{path}: missing frontmatter")
+    end = text.find("\n---", 3)
+    if end == -1:
+        raise EligibilityError(f"{path}: unterminated frontmatter")
+    try:
+        meta = yaml.safe_load(text[3:end]) or {}
+    except (yaml.YAMLError, ValueError) as exc:
+        raise EligibilityError(f"{path}: malformed frontmatter") from exc
+    if not isinstance(meta, dict):
+        raise EligibilityError(f"{path}: frontmatter is not a mapping")
+    return meta, text[end + 4 :]
+
+
+def _qualify(task_id: str, workplan_id: str) -> str:
+    return f"{workplan_id}-{task_id.upper()}" if _SHORT_TASK_RE.match(task_id) else task_id
+
+
+def _dependency_list(value: Any, where: str) -> list[str]:
+    if value is None or value == "":
+        return []
+    if isinstance(value, str):
+        items: Sequence[Any] = value.split(",")
+    elif isinstance(value, list):
+        if any(isinstance(item, (dict, list)) for item in value):
+            raise EligibilityError(f"{where}: malformed depends_on")
+        items = [item for item in value if item is not None]
+    else:
+        raise EligibilityError(f"{where}: malformed depends_on")
+    out: list[str] = []
+    for item in items:
+        target = str(item).strip().strip('"').strip("'").strip()
+        if target and target not in out:
+            out.append(target)
+    return out
+
+
+def _workplan_dependencies(meta: Mapping[str, Any], workplan_id: str) -> list[str]:
+    out: list[str] = []
+    for key in ("depends_on", "depends_on_workplans", "depends_on_tasks"):
+        for target in _dependency_list(meta.get(key), f"{workplan_id} {key}"):
+            target = _qualify(target, workplan_id)
+            if target not in out:
+                out.append(target)
+    return out
+
+
+def _parse_workplan(path: str, text: str) -> _Workplan:
+    meta, body = _split_frontmatter(text, path)
+    record_id = str(meta.get("id") or "").strip()
+    if not record_id:
+        raise EligibilityError(f"{path}: workplan has no id")
+    workplan = _Workplan(record_id=record_id, path=path, meta=meta)
+    for match in _TASK_BLOCK_RE.finditer(body):
+        try:
+            block = yaml.safe_load(match.group(1).strip()) or {}
+        except (yaml.YAMLError, ValueError) as exc:
+            raise EligibilityError(f"{path}: malformed task block") from exc
+        if not isinstance(block, dict):
+            continue
+        raw_id = str(block.get("id") or "").strip()
+        if not raw_id:
+            continue
+        workplan.tasks.append(
+            _Task(_qualify(raw_id, record_id), raw_id, record_id, block)
+        )
+    return workplan
+
+
+def _status(value: Any, where: str) -> str:
+    if not isinstance(value, str) or not value.strip():
+        raise EligibilityError(f"{where}: missing authoritative status")
+    return value.strip().lower()
+
+
+def _needs_human(value: Any, where: str) -> bool:
+    if value is None:
+        return False  # same default as the forge projection
+    if isinstance(value, bool):
+        return value
+    text = str(value).strip().lower()
+    if text in {"true", "yes", "1", "on"}:
+        return True
+    if text in {"false", "no", "0", "off", ""}:
+        return False
+    raise EligibilityError(f"{where}: needs_human must be boolean")
+
+
+def _optional_text(value: Any, where: str, name: str) -> str | None:
+    if value is None:
+        return None
+    if not isinstance(value, str):
+        raise EligibilityError(f"{where}: {name} must be text")
+    return value.strip() or None
+
+
+def _wait_until(value: Any, where: str) -> str | None:
+    if value is None or value == "":
+        return None
+    if type(value) is date:
+        return value.isoformat()
+    text = str(value).strip()
+    if _DATE.fullmatch(text):
+        try:
+            return date.fromisoformat(text).isoformat()
+        except ValueError:
+            pass
+    raise EligibilityError(f"{where}: wait_until must be YYYY-MM-DD")
+
+
+def _flavor(value: Any) -> str | None:
+    if value is None:
+        return None
+    text = str(value).strip().lower()
+    return None if text in _EMPTY_FLAVOR else text
+
+
+# --------------------------------------------------------------------------- #
+# Builder
+# --------------------------------------------------------------------------- #
+
+
+def _load_revision(reader: ForgeCommitReader, sha: str) -> list[_Workplan]:
+    paths: list[str] = []
+    for directory in WORKPLAN_DIRS:
+        listed = reader.list_markdown(directory, sha)
+        if listed is None:
+            if directory == WORKPLAN_DIRS[0]:
+                raise EligibilityError("no workplans directory at scan revision")
+            continue
+        paths.extend(listed)
+    return [_parse_workplan(path, reader.read_text(path, sha)) for path in paths]
+
+
+def build_current_tasks(
+    *,
+    target_repo: str,
+    scan_revision: str,
+    task_record_ids: Sequence[str],
+    client: httpx.Client,
+    decision_status: StatusResolver,
+    external_status: StatusResolver | None = None,
+    require_current_head: bool = True,
+    report: Mapping[str, Any] | None = None,
+    forge_owner: str = DEFAULT_FORGE_OWNER,
+    forge_api: str = DEFAULT_FORGE_API,
+) -> dict[str, Any]:
+    """Return ``{"schema", "target_repo", "source_revision", "current_tasks"}``.
+
+    ``current_tasks`` is keyed by the task record IDs requested (as the C-25
+    findings name them) and is the exact ``current_tasks`` argument for
+    ``plan_owner_returns``. Raises :class:`EligibilityError` instead of
+    returning a partial or unbound mapping.
+    """
+    if not isinstance(target_repo, str) or not _REPO.fullmatch(target_repo):
+        raise EligibilityError("invalid repository identity")
+    if not isinstance(scan_revision, str) or not _SHA.fullmatch(scan_revision):
+        raise EligibilityError("scan_revision must be a full 40-hex commit SHA")
+    if report is not None:
+        if report.get("repo_slug") != target_repo:
+            raise EligibilityError("C-25 report is for a different repository")
+        for key in ("source_revision", "commit"):
+            if key in report and report[key] != scan_revision:
+                raise EligibilityError("C-25 report revision does not match scan_revision")
+    requested = list(dict.fromkeys(task_record_ids))
+    if not requested or any(not isinstance(t, str) or not t for t in requested):
+        raise EligibilityError("task_record_ids must be non-empty identities")
+
+    reader = ForgeCommitReader(client, target_repo, owner=forge_owner, api_base=forge_api)
+    reader.verify_commit(scan_revision)
+    if require_current_head and reader.default_branch_head() != scan_revision:
+        raise EligibilityError("scan revision is no longer the default-branch head")
+
+    workplans = _load_revision(reader, scan_revision)
+    by_workplan: dict[str, _Workplan] = {}
+    by_task: dict[str, _Task] = {}
+    by_raw: dict[str, list[_Task]] = {}
+    for workplan in workplans:
+        if workplan.record_id in by_workplan:
+            raise EligibilityError(f"duplicate workplan identity {workplan.record_id}")
+        by_workplan[workplan.record_id] = workplan
+        for task in workplan.tasks:
+            if task.record_id in by_task:
+                raise EligibilityError(f"duplicate task identity {task.record_id}")
+            by_task[task.record_id] = task
+            by_raw.setdefault(task.raw_id, []).append(task)
+
+    def predecessor_status(target: str) -> str | None:
+        if target in by_task:
+            return _status(by_task[target].block.get("status"), target)
+        if target in by_workplan:
+            return _status(by_workplan[target].meta.get("status"), target)
+        if external_status is None:
+            return None
+        try:
+            value = external_status(target)
+        except Exception:  # noqa: BLE001 - unknown, never guessed
+            return None
+        return value.strip().lower() if isinstance(value, str) and value.strip() else None
+
+    rows: dict[str, Any] = {}
+    for key in requested:
+        matches = [by_task[key]] if key in by_task else by_raw.get(key, [])
+        if len(matches) != 1:
+            raise EligibilityError(
+                f"task {key} is {'ambiguous' if matches else 'absent'} at scan revision"
+            )
+        task = matches[0]
+        workplan = by_workplan[task.workplan_id]
+        where = f"{task.record_id}"
+        block = task.block
+        depends_on = _workplan_dependencies(workplan.meta, workplan.record_id)
+        for target in _dependency_list(block.get("depends_on"), f"{where} depends_on"):
+            target = _qualify(target, workplan.record_id)
+            if target not in depends_on:
+                depends_on.append(target)
+        decision_id = _optional_text(block.get("decision_id"), where, "decision_id")
+        resolved_decision: str | None = None
+        if decision_id is not None:
+            try:
+                value = decision_status(decision_id)
+            except Exception:  # noqa: BLE001 - unknown stays unknown
+                value = None
+            resolved_decision = value if isinstance(value, str) and value else None
+        rows[key] = {
+            "target_repo": target_repo,
+            "task_record_id": key,
+            "status": _status(block.get("status"), where),
+            "workplan_status": _status(workplan.meta.get("status"), workplan.record_id),
+            "workplan_flavor": _flavor(workplan.meta.get("flavor")),
+            "needs_human": _needs_human(block.get("needs_human"), where),
+            "blocked_on": _optional_text(block.get("blocked_on"), where, "blocked_on"),
+            "depends_on": depends_on,
+            "predecessor_statuses": {t: predecessor_status(t) for t in depends_on},
+            "wait_until": _wait_until(block.get("wait_until"), where),
+            "decision_id": decision_id,
+            "decision_status": resolved_decision,
+        }
+    return {
+        "schema": SCHEMA,
+        "target_repo": target_repo,
+        "source_revision": scan_revision,
+        "current_tasks": rows,
+    }
+
+
+def state_hub_decision_status(client: httpx.Client, api_base: str) -> StatusResolver:
+    """Resolver reading ``GET {api_base}/decisions/{id}``; failure -> unknown."""
+
+    def resolve(decision_id: str) -> str | None:
+        try:
+            response = client.get(f"{api_base.rstrip('/')}/decisions/{quote(decision_id, safe='')}")
+        except httpx.HTTPError:
+            return None
+        if response.status_code != 200:
+            return None
+        try:
+            status = response.json().get("status")
+        except (ValueError, AttributeError):
+            return None
+        return status if isinstance(status, str) and status else None
+
+    return resolve
+
+
+def forge_head_status_resolver(
+    client: httpx.Client,
+    repo_for_prefix: Mapping[str, str],
+    *,
+    forge_owner: str = DEFAULT_FORGE_OWNER,
+    forge_api: str = DEFAULT_FORGE_API,
+) -> StatusResolver:
+    """Cross-repo predecessor resolver over explicit ID-prefix -> repo mapping.
+
+    Each foreign repository is read once, pinned to its default-branch HEAD
+    commit at first use. Unmapped prefixes and any read failure -> unknown.
+    """
+    cache: dict[str, dict[str, str] | None] = {}
+
+    def load(repo: str) -> dict[str, str] | None:
+        if repo not in cache:
+            try:
+                reader = ForgeCommitReader(client, repo, owner=forge_owner, api_base=forge_api)
+                head = reader.default_branch_head()
+                statuses: dict[str, str] = {}
+                for workplan in _load_revision(reader, head):
+                    statuses[workplan.record_id] = _status(workplan.meta.get("status"), workplan.record_id)
+                    for task in workplan.tasks:
+                        statuses[task.record_id] = _status(task.block.get("status"), task.record_id)
+                cache[repo] = statuses
+            except EligibilityError:
+                cache[repo] = None
+        return cache[repo]
+
+    def resolve(target: str) -> str | None:
+        prefix = next(
+            (p for p in sorted(repo_for_prefix, key=len, reverse=True) if target.startswith(p)),
+            None,
+        )
+        if prefix is None:
+            return None
+        statuses = load(repo_for_prefix[prefix])
+        return None if statuses is None else statuses.get(target)
+
+    return resolve

tests/test_owner_return_eligibility.py

@@ -0,0 +1,309 @@
+"""Tests for the owner-return eligibility input builder (ACTIVITY-WP-0043-T05).
+
+All Forge and State Hub HTTP is served by ``httpx.MockTransport``.
+"""
+
+from __future__ import annotations
+
+import importlib.util
+from datetime import date
+from pathlib import Path
+from urllib.parse import unquote
+
+import httpx
+import pytest
+
+from activity_core import owner_return_eligibility as ore
+from activity_core import owner_return_pickup as orp
+
+REPO = "ops-warden"
+SHA = "a" * 40
+OLD = "b" * 40
+API = ore.DEFAULT_FORGE_API
+BASE = f"{API}/repos/coulomb/{REPO}"
+
+WP1 = """---
+id: WARDEN-WP-0041
+status: active
+flavor: implementation
+depends_on: [WARDEN-WP-0040]
+---
+
+## Owner return
+
+```task
+id: WARDEN-WP-0041-T03
+status: wait
+priority: high
+needs_human: false
+blocked_on: "message-from:state-hub"
+depends_on: [T01, STATE-WP-0094-T03]
+wait_until: "2026-10-01"
+decision_id: "dec-1"
+```
+
+```task
+id: T01
+status: done
+```
+
+```task
+id: WARDEN-WP-0041-T04
+status: wait
+blocked_on: "message-from:state-hub"
+```
+"""
+
+WP0 = """---
+id: WARDEN-WP-0040
+status: finished
+---
+"""
+
+
+def _files(**over):
+    files = {
+        "workplans/WARDEN-WP-0041-owner.md": WP1,
+        "workplans/archived/261001-WARDEN-WP-0040-prior.md": WP0,
+    }
+    files.update(over)
+    return {k: v for k, v in files.items() if v is not None}
+
+
+def _transport(files, *, sha=SHA, head=SHA, seen=None, fail=None):
+    def handler(request: httpx.Request) -> httpx.Response:
+        url = str(request.url).split("?")[0]
+        path = unquote(url)
+        ref = request.url.params.get("ref")
+        if seen is not None:
+            seen.append((path, ref))
+        if fail and fail in path:
+            return httpx.Response(500)
+        if path == BASE:
+            return httpx.Response(200, json={"default_branch": "main"})
+        if path == f"{BASE}/branches/main":
+            return httpx.Response(200, json={"commit": {"id": head}})
+        if path.startswith(f"{BASE}/git/commits/"):
+            asked = path.rsplit("/", 1)[1]
+            return httpx.Response(200, json={"sha": sha if asked == sha else "c" * 40})
+        if path.startswith(f"{BASE}/contents/"):
+            if ref != sha:
+                return httpx.Response(404)
+            directory = path[len(f"{BASE}/contents/"):]
+            entries = [
+                {"type": "file", "name": p.rsplit("/", 1)[1], "path": p}
+                for p in files if p.rsplit("/", 1)[0] == directory
+            ]
+            if not entries and not any(p.startswith(directory + "/") for p in files):
+                return httpx.Response(404)
+            if directory == "workplans":
+                entries.append({"type": "dir", "name": "archived", "path": "workplans/archived"})
+            return httpx.Response(200, json=entries)
+        if path.startswith(f"{BASE}/raw/"):
+            name = path[len(f"{BASE}/raw/"):]
+            if ref != sha or name not in files:
+                return httpx.Response(404)
+            return httpx.Response(200, text=files[name])
+        return httpx.Response(404)
+
+    return httpx.MockTransport(handler)
+
+
+def _build(files=None, *, decisions=None, external=None, **kw):
+    transport = kw.pop("transport", None) or _transport(files or _files())
+    client = httpx.Client(transport=transport)
+    decisions = {"dec-1": "resolved"} if decisions is None else decisions
+    return ore.build_current_tasks(
+        target_repo=REPO,
+        scan_revision=kw.pop("scan_revision", SHA),
+        task_record_ids=kw.pop("task_record_ids", ["WARDEN-WP-0041-T03"]),
+        client=client,
+        decision_status=decisions.get,
+        external_status=external,
+        **kw,
+    )
+
+
+def test_builds_complete_row_bound_to_revision():
+    seen: list = []
+    out = _build(transport=_transport(_files(), seen=seen),
+                 external={"STATE-WP-0094-T03": "done"}.get)
+    assert out["schema"] == ore.SCHEMA
+    assert out["source_revision"] == SHA
+    row = out["current_tasks"]["WARDEN-WP-0041-T03"]
+    assert row == {
+        "target_repo": REPO,
+        "task_record_id": "WARDEN-WP-0041-T03",
+        "status": "wait",
+        "workplan_status": "active",
+        "workplan_flavor": "implementation",
+        "needs_human": False,
+        "blocked_on": "message-from:state-hub",
+        "depends_on": ["WARDEN-WP-0040", "WARDEN-WP-0041-T01", "STATE-WP-0094-T03"],
+        "predecessor_statuses": {
+            "WARDEN-WP-0040": "finished",
+            "WARDEN-WP-0041-T01": "done",
+            "STATE-WP-0094-T03": "done",
+        },
+        "wait_until": "2026-10-01",
+        "decision_id": "dec-1",
+        "decision_status": "resolved",
+    }
+    content_reads = [(p, r) for p, r in seen if "/contents/" in p or "/raw/" in p]
+    assert content_reads and all(ref == SHA for _, ref in content_reads)
+
+
+def test_unresolvable_predecessor_and_decision_stay_unknown():
+    out = _build(decisions={})
+    row = out["current_tasks"]["WARDEN-WP-0041-T03"]
+    assert row["predecessor_statuses"]["STATE-WP-0094-T03"] is None
+    assert "STATE-WP-0094-T03" in row["depends_on"]
+    assert row["decision_status"] is None
+
+
+def test_external_resolver_exception_is_unknown_not_fatal():
+    def boom(_):
+        raise RuntimeError("down")
+    row = _build(external=boom)["current_tasks"]["WARDEN-WP-0041-T03"]
+    assert row["predecessor_statuses"]["STATE-WP-0094-T03"] is None
+
+
+def test_task_without_dependencies_has_explicit_workplan_gate_only():
+    row = _build(task_record_ids=["WARDEN-WP-0041-T04"])["current_tasks"]["WARDEN-WP-0041-T04"]
+    assert row["depends_on"] == ["WARDEN-WP-0040"]
+    assert row["decision_id"] is None and row["wait_until"] is None
+    assert row["needs_human"] is False
+
+
+@pytest.mark.parametrize("revision", ["main", "a" * 39, "A" * 40, None])
+def test_refuses_non_sha_revision(revision):
+    with pytest.raises(ore.EligibilityError):
+        _build(scan_revision=revision)
+
+
+def test_refuses_when_scan_is_not_current_head():
+    with pytest.raises(ore.EligibilityError, match="no longer the default-branch head"):
+        _build(transport=_transport(_files(), head=OLD))
+
+
+def test_stale_scan_allowed_only_when_head_check_disabled():
+    out = _build(transport=_transport(_files(), head=OLD), require_current_head=False)
+    assert out["source_revision"] == SHA
+
+
+def test_refuses_unknown_commit():
+    with pytest.raises(ore.EligibilityError, match="does not match"):
+        _build(transport=_transport(_files(), sha=OLD, head=SHA))
+
+
+def test_refuses_report_revision_mismatch():
+    report = {"repo_slug": REPO, "source_revision": OLD, "issues": []}
+    with pytest.raises(ore.EligibilityError, match="revision"):
+        _build(report=report)
+    with pytest.raises(ore.EligibilityError, match="different repository"):
+        _build(report={"repo_slug": "other", "issues": []})
+
+
+@pytest.mark.parametrize(
+    "old,new,match",
+    [
+        ("status: wait\npriority: high", "priority: high", "missing authoritative status"),
+        ("needs_human: false", "needs_human: maybe", "needs_human"),
+        ('wait_until: "2026-10-01"', 'wait_until: "soon"', "wait_until"),
+        ("depends_on: [T01, STATE-WP-0094-T03]", "depends_on: {a: b}", "depends_on"),
+        ("id: WARDEN-WP-0041\nstatus: active", "id: WARDEN-WP-0041", "missing authoritative status"),
+    ],
+)
+def test_missing_or_malformed_fields_fail_closed(old, new, match):
+    files = _files(**{"workplans/WARDEN-WP-0041-owner.md": WP1.replace(old, new, 1)})
+    with pytest.raises(ore.EligibilityError, match=match):
+        _build(files)
+
+
+def test_absent_and_duplicate_tasks_refuse():
+    with pytest.raises(ore.EligibilityError, match="absent"):
+        _build(task_record_ids=["WARDEN-WP-0041-T09"])
+    dup = WP1.replace("id: WARDEN-WP-0041-T04", "id: WARDEN-WP-0041-T03")
+    with pytest.raises(ore.EligibilityError, match="duplicate task"):
+        _build(_files(**{"workplans/WARDEN-WP-0041-owner.md": dup}))
+
+
+def test_forge_failure_refuses():
+    with pytest.raises(ore.EligibilityError):
+        _build(transport=_transport(_files(), fail="/raw/"))
+
+
+def test_missing_workplans_directory_refuses():
+    with pytest.raises(ore.EligibilityError, match="no workplans"):
+        _build(transport=_transport({}))
+
+
+def test_state_hub_decision_resolver():
+    def handler(request):
+        if request.url.path.endswith("/decisions/dec-1"):
+            return httpx.Response(200, json={"id": "dec-1", "status": "open"})
+        return httpx.Response(404)
+    resolve = ore.state_hub_decision_status(
+        httpx.Client(transport=httpx.MockTransport(handler)), "http://hub"
+    )
+    assert resolve("dec-1") == "open"
+    assert resolve("dec-2") is None
+
+
+def test_forge_head_resolver_pins_foreign_head_and_maps_prefix():
+    foreign = "d" * 40
+    files = {"workplans/STATE-WP-0094-x.md": "---\nid: STATE-WP-0094\nstatus: active\n---\n"
+             "```task\nid: STATE-WP-0094-T03\nstatus: progress\n```\n"}
+    fbase = f"{API}/repos/coulomb/state-hub"
+
+    def handler(request):
+        path = unquote(str(request.url).split("?")[0])
+        ref = request.url.params.get("ref")
+        if path == fbase:
+            return httpx.Response(200, json={"default_branch": "main"})
+        if path == f"{fbase}/branches/main":
+            return httpx.Response(200, json={"commit": {"id": foreign}})
+        if path == f"{fbase}/contents/workplans" and ref == foreign:
+            return httpx.Response(200, json=[{"type": "file", "name": "STATE-WP-0094-x.md",
+                                              "path": "workplans/STATE-WP-0094-x.md"}])
+        if path.startswith(f"{fbase}/raw/") and ref == foreign:
+            return httpx.Response(200, text=files[path[len(f"{fbase}/raw/"):]])
+        return httpx.Response(404)
+
+    resolve = ore.forge_head_status_resolver(
+        httpx.Client(transport=httpx.MockTransport(handler)), {"STATE-WP-": "state-hub"}
+    )
+    assert resolve("STATE-WP-0094-T03") == "progress"
+    assert resolve("STATE-WP-0094") == "active"
+    assert resolve("STATE-WP-0094-T99") is None
+    assert resolve("OTHER-WP-0001") is None
+
+
+_PLANNER = Path("/home/worsch/state-hub/scripts/owner_return_handoff.py")
+
+
+@pytest.mark.skipif(not _PLANNER.is_file(), reason="state-hub planner not checked out")
+def test_output_is_accepted_by_state_hub_planner():
+    spec = importlib.util.spec_from_file_location("owner_return_handoff", _PLANNER)
+    planner = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(planner)
+    msg = "11111111-2222-3333-4444-555555555555"
+    report = {"repo_slug": REPO, "result": "warn", "issues": [{
+        "check_id": "C-25", "file_value": "message-from:state-hub",
+        "owner_return": {"task_record_id": "WARDEN-WP-0041-T03", "message_ids": [msg]},
+    }]}
+    ok = _build(external={"STATE-WP-0094-T03": "done"}.get, report=report)
+    plan = planner.plan_owner_returns(
+        report, activity_definition_id=orp.ACTIVITY_DEFINITION_ID,
+        current_tasks=ok["current_tasks"], observed_on=date(2026, 10, 4),
+    )
+    assert [c["disposition"] for c in plan] == ["reviewable"]
+    assert plan[0]["triggering_event_id"] == orp.owner_return_event_id(
+        REPO, "WARDEN-WP-0041-T03", msg)
+    unknown = _build(report=report)
+    plan = planner.plan_owner_returns(
+        report, activity_definition_id=orp.ACTIVITY_DEFINITION_ID,
+        current_tasks=unknown["current_tasks"], observed_on=date(2026, 10, 4),
+    )
+    assert plan[0]["disposition"] == "stopped"
+    assert plan[0]["stop_reason"] == "predecessor_not_terminal_or_unknown"

workplans/ACTIVITY-WP-0043-state-hub-owner-return-pickup.md

@@ -85,5 +85,26 @@ blocking_reason: "Activation is a separately reviewed founder action"
 state_hub_task_id: "254526ce-5a76-57b3-b7d5-a874518c7b40"
 ```
 
-Plan: `docs/owner-return-pickup-activation-plan.md`. Requires T03 and an
+Plan: `docs/owner-return-pickup-activation-plan.md`. Requires T03 and T05
-eligibility-source binding (fresh authoritative `current_tasks`).
+(eligibility-source binding of fresh authoritative `current_tasks`).
+
+## Eligibility input builder bound to the scan revision
+
+```task
+id: ACTIVITY-WP-0043-T05
+status: done
+priority: high
+```
+
+`src/activity_core/owner_return_eligibility.py::build_current_tasks` builds the
+planner's `current_tasks` from immutable Forge commit content at the exact
+40-hex `scan_revision` of the C-25 scan (`/contents` and `/raw` with
+`ref=<sha>`; no checkouts, ADR-002). It verifies the commit and, by default,
+that it is still the default-branch head; refuses on any revision mismatch,
+missing/malformed field, duplicate identity or absent task. Predecessors are
+task-block plus workplan-frontmatter dependencies, resolved in the same commit
+or via an explicit cross-repo resolver (`forge_head_status_resolver`, HEAD
+pinned per foreign repo); unresolved targets stay `None`. Decision status comes
+from an injected resolver (`state_hub_decision_status`); failure stays unknown.
+Tests: `tests/test_owner_return_eligibility.py` (HTTP via MockTransport,
+including acceptance by the State Hub planner).