ACTIVITY-WP-0043-T06: single-repo owner-return pickup caller in the worker

Precondition 3 of the activation plan. New bounded operation
owner_return_pickup (state-hub-owner-return/pickup), executed by the existing
execute_bounded_operation activity and operator-triggered via the ops console
trigger with confirm_side_effect; no schedule, definition stays disabled.

The caller reads an operator-supplied C-25 checker artefact pinned by sha256
(the checker needs a local checkout and cannot run in the worker), requires a
full-SHA source_revision on the report and every C-25 finding, refuses failed
or incomplete collection, binds build_current_tasks to that SHA (Forge HEAD
verified), runs the State Hub planner vendored byte-identically from
state-hub 150cb0b, and submits only reviewable candidates under
ACTIVITY_CORE_OWNER_RETURN_PICKUP_ENABLED. Evidence is IDs/counts only.

build_current_tasks now refuses a report without source_revision and names
source_revision_reason when it is null.

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

src/activity_core/activities.py

@@ -258,6 +258,13 @@ async def execute_bounded_operation(payload: dict[str, Any]) -> dict[str, Any]:
         from activity_core.context_resolvers.repo_manager import repo_manager_freshness
 
         result = repo_manager_freshness(params)
+    elif spec.operation_id == "owner_return_pickup":
+        from activity_core.owner_return_runner import run_owner_return_pickup
+
+        result = await run_owner_return_pickup(
+            params,
+            session_factory=_get_session_factory() if params.get("submit") is True else None,
+        )
     else:
         raise ApplicationError(
             f"bounded operation {spec.operation_id!r} uses a dedicated activity",

src/activity_core/bounded_operations.py

@@ -73,6 +73,21 @@ _SPECS = (
         credential_route="repo-manager-freshness-runner",
         evidence_mode="repo-manager-freshness-receipt",
     ),
+    # ACTIVITY-WP-0043-T06: single-repo owner-return pickup (C-25 artefact ->
+    # Forge-bound eligibility -> vendored State Hub planner -> adapter).
+    # Operator-triggered only; submission additionally gated by
+    # ACTIVITY_CORE_OWNER_RETURN_PICKUP_ENABLED.
+    BoundedOperationSpec(
+        operation_id="owner_return_pickup",
+        source_type="state-hub-owner-return",
+        query="pickup",
+        resolve_before_execute=False,
+        max_timeout_seconds=300,
+        temporal_max_attempts=1,
+        idempotency="single Temporal attempt; ops_runs.idempotency_key unique (ON CONFLICT DO NOTHING)",
+        credential_route="none (public Forge reads; State Hub decision reads)",
+        evidence_mode="owner-return-pickup-receipts",
+    ),
 )
 
 BOUNDED_OPERATION_REGISTRY = {spec.operation_id: spec for spec in _SPECS}
@@ -177,6 +192,12 @@ def normalize_operation_result(operation_id: str, raw: Any) -> dict[str, Any]:
             "outcomes": outcomes,
             "exit_code": exit_code,
         }
+    if operation_id == "owner_return_pickup":
+        # The runner already projects to counts/identities; keep only its keys.
+        keys = ("kind", "mode", "target_repo", "source_revision", "report_sha256",
+                "observed_on", "planner_input", "planner_output", "receipts",
+                "receipt_counts", "submitted_count")
+        return {key: result.get(key) for key in keys} | {"kind": operation_id}
     raise ValueError(f"no result projection for bounded operation {operation_id!r}")
 
 
@@ -235,6 +256,10 @@ def validate_bounded_operations(
             _validate_backup(params)
         elif spec.operation_id == "repo_manager_freshness":
             _validate_repo_manager_freshness(params)
+        elif spec.operation_id == "owner_return_pickup":
+            from activity_core.owner_return_runner import validate_params
+
+            validate_params(params)
 
 
 def _validate_timeout(spec: BoundedOperationSpec, params: dict[str, Any]) -> None:

src/activity_core/ops_console.py

@@ -41,6 +41,7 @@ SIDE_EFFECT_MARKERS = (
     "forgejo_package_prune",
     "cnpg_option_a_backup",
     "repo_manager_freshness",
+    "owner_return_pickup",
     "apply: true",
     '"apply": true',
     "'apply': true",

src/activity_core/owner_return_eligibility.py

@@ -328,6 +328,16 @@ def build_current_tasks(
     if report is not None:
         if report.get("repo_slug") != target_repo:
             raise EligibilityError("C-25 report is for a different repository")
+        # State Hub 150cb0b: every checker report carries source_revision (a
+        # full SHA, or null + source_revision_reason). A report without the
+        # key predates the contract and is unbound -- refuse it.
+        if "source_revision" not in report:
+            raise EligibilityError("C-25 report has no source_revision (unbound report)")
+        if report["source_revision"] is None:
+            raise EligibilityError(
+                "C-25 report has no source revision: "
+                f"{report.get('source_revision_reason') or 'reason not given'}"
+            )
         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")

src/activity_core/owner_return_runner.py

@@ -0,0 +1,324 @@
+"""Single-repository owner-return pickup caller (ACTIVITY-WP-0043-T06).
+
+Precondition 3 of ``docs/owner-return-pickup-activation-plan.md``: the C-25 →
+planner → adapter chain runs inside the existing Activity Core worker as the
+code-registered bounded operation ``owner_return_pickup`` (ACT-ADR-007),
+operator-triggered only (``POST /ops/automations/{id}/trigger`` with
+``confirm_side_effect``). No schedule exists; the definition stays disabled.
+
+Report acquisition (why an operator-supplied artefact)
+------------------------------------------------------
+State Hub's C-25 checker (``scripts/consistency_check.py``) reads workplan files
+from the repository's registered ``local_path`` and derives ``source_revision``
+from ``git`` in that working tree. The worker has no such checkouts (ADR-002)
+and no checker image, and Repo Manager's C-rule engine does not implement C-25
+yet. So the worker cannot produce the report itself. It accepts a checker JSON
+artefact supplied by the operator at a worker-mounted path, pinned by sha256 in
+the reviewed definition params, and binds it to its revision:
+
+1. ``report.repo_slug`` must equal ``target_repo``; ``report.source_revision``
+   must be a full 40-hex SHA (``null`` + ``source_revision_reason`` refuses);
+   every C-25 finding must carry the same ``repo_slug``/``source_revision``.
+2. ``build_current_tasks`` re-reads the workplans from immutable Forge content
+   at that SHA and refuses unless it is still the default-branch HEAD, so a
+   stale or forged revision cannot be planned from.
+3. Collection must be complete: a failed checker run or any C-25
+   "reply collection ... is incomplete" finding refuses.
+
+The State Hub planner is vendored byte-identically
+(``activity_core.vendored.state_hub_owner_return_handoff``). Existing ops-run
+rows are deliberately *not* passed to it: deduplication stays server-side in
+``submit_owner_returns`` (unique idempotency key), so a re-run reports
+``duplicate`` receipts rather than trusting a snapshot.
+
+Evidence carries counts and identities only (event/message/ops-run IDs, task
+record IDs, dispositions) -- never message bodies or finding text.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import re
+from collections import Counter
+from datetime import date, datetime, timezone
+from pathlib import Path
+from typing import Any, Callable, Mapping
+
+import httpx
+
+from activity_core import owner_return_pickup as orp
+from activity_core.owner_return_eligibility import (
+    DEFAULT_FORGE_API,
+    build_current_tasks,
+    forge_head_status_resolver,
+    state_hub_decision_status,
+)
+from activity_core.vendored import state_hub_owner_return_handoff as planner
+
+OPERATION_ID = "owner_return_pickup"
+MAX_REPORT_BYTES = 5 * 1024 * 1024
+MAX_EVIDENCE_ROWS = 200
+
+_SHA = re.compile(r"^[0-9a-f]{40}$")
+_SHA256 = re.compile(r"^[0-9a-f]{64}$")
+_REPO = re.compile(r"^[a-z0-9][a-z0-9._-]{0,127}$")
+_PREFIX = re.compile(r"^[A-Z][A-Z0-9]*-(?:WP-)?$")
+_INCOMPLETE = re.compile(r"reply collection from \S+ is incomplete")
+
+
+class OwnerReturnPickupRefused(ValueError):
+    """A precondition failed; nothing was submitted."""
+
+
+def validate_params(params: Mapping[str, Any]) -> None:
+    """Admission checks shared by definition parsing and execution."""
+    repo = params.get("target_repo")
+    if not isinstance(repo, str) or not _REPO.fullmatch(repo):
+        raise ValueError("bounded operation owner_return_pickup requires one target_repo")
+    if not isinstance(params.get("report_file"), str) or not params["report_file"].strip():
+        raise ValueError("bounded operation owner_return_pickup requires report_file")
+    digest = params.get("report_sha256")
+    if not isinstance(digest, str) or not _SHA256.fullmatch(digest):
+        raise ValueError("bounded operation owner_return_pickup requires report_sha256 (64 hex)")
+    if not isinstance(params.get("submit"), bool):
+        raise ValueError("bounded operation owner_return_pickup must declare boolean submit explicitly")
+    expected = params.get("expected_source_revision")
+    if expected is not None and (not isinstance(expected, str) or not _SHA.fullmatch(expected)):
+        raise ValueError("expected_source_revision must be a full 40-hex SHA")
+    mapping = params.get("predecessor_repos", {})
+    if not isinstance(mapping, dict) or any(
+        not isinstance(k, str) or not _PREFIX.fullmatch(k)
+        or not isinstance(v, str) or not _REPO.fullmatch(v)
+        for k, v in mapping.items()
+    ):
+        raise ValueError("predecessor_repos must map ID prefixes (e.g. 'STATE-WP-') to repo slugs")
+    for forbidden in ("target_repos", "repos", "command", "argv", "ops_runs"):
+        if forbidden in params:
+            raise ValueError(f"bounded operation owner_return_pickup must not declare {forbidden}")
+
+
+def load_report(path: Path, expected_sha256: str) -> Any:
+    try:
+        raw = path.read_bytes()
+    except OSError as exc:
+        raise OwnerReturnPickupRefused(f"report artefact unreadable: {path}") from exc
+    if len(raw) > MAX_REPORT_BYTES:
+        raise OwnerReturnPickupRefused("report artefact exceeds size bound")
+    if hashlib.sha256(raw).hexdigest() != expected_sha256:
+        raise OwnerReturnPickupRefused("report artefact does not match pinned report_sha256")
+    try:
+        report = json.loads(raw)
+    except json.JSONDecodeError as exc:
+        raise OwnerReturnPickupRefused("report artefact is not JSON") from exc
+    if not isinstance(report, (dict, list)):
+        raise OwnerReturnPickupRefused("report artefact must be a checker JSON object or list")
+    return report
+
+
+def select_repo_report(artefact: Any, target_repo: str) -> dict[str, Any]:
+    """Pick the single repository object (``--repo`` object or ``--all`` list)."""
+    if isinstance(artefact, dict):
+        return artefact
+    matches = [r for r in artefact if isinstance(r, dict) and r.get("repo_slug") == target_repo]
+    if len(matches) != 1:
+        raise OwnerReturnPickupRefused(
+            f"report artefact holds {len(matches)} objects for {target_repo}; exactly one required"
+        )
+    return matches[0]
+
+
+def bind_report(
+    report: Mapping[str, Any], target_repo: str, expected_revision: str | None = None
+) -> dict[str, Any]:
+    """Return the planner-input summary, refusing unbound or incomplete reports."""
+    if report.get("repo_slug") != target_repo:
+        raise OwnerReturnPickupRefused("report is for a different repository")
+    revision = report.get("source_revision")
+    if not isinstance(revision, str) or not _SHA.fullmatch(revision):
+        reason = report.get("source_revision_reason")
+        raise OwnerReturnPickupRefused(
+            "report source_revision is not a full 40-hex SHA"
+            + (f" (reason: {reason})" if isinstance(reason, str) else "")
+        )
+    if expected_revision is not None and revision != expected_revision:
+        raise OwnerReturnPickupRefused("report source_revision differs from expected_source_revision")
+    if report.get("result") == "error":
+        raise OwnerReturnPickupRefused("checker collection failed (result=error)")
+    issues = report.get("issues")
+    if not isinstance(issues, list):
+        raise OwnerReturnPickupRefused("report issues must be a list")
+    findings = 0
+    owner_returns: list[Mapping[str, Any]] = []
+    incomplete: list[str] = []
+    for issue in issues:
+        if not isinstance(issue, Mapping) or issue.get("check_id") != "C-25":
+            continue
+        findings += 1
+        if issue.get("repo_slug") != target_repo or issue.get("source_revision") != revision:
+            raise OwnerReturnPickupRefused("C-25 finding is not bound to the report repository/revision")
+        if "owner_return" in issue:
+            owner_returns.append(issue)
+        elif _INCOMPLETE.search(str(issue.get("message") or "")):
+            incomplete.append(str(issue.get("file_value") or ""))
+    if incomplete:
+        raise OwnerReturnPickupRefused(
+            f"owner-return reply collection incomplete for {len(incomplete)} counterpart(s)"
+        )
+    tasks: list[str] = []
+    messages = 0
+    counterparts: set[str] = set()
+    for issue in owner_returns:
+        context = issue["owner_return"]
+        if isinstance(context, Mapping):
+            task = context.get("task_record_id")
+            if isinstance(task, str) and task not in tasks:
+                tasks.append(task)
+            ids = context.get("message_ids")
+            messages += len(ids) if isinstance(ids, list) else 0
+        marker = str(issue.get("file_value") or "")
+        if marker.startswith("message-from:"):
+            counterparts.add(marker.removeprefix("message-from:"))
+    return {
+        "report_result": report.get("result"),
+        "source_revision": revision,
+        "c25_finding_count": findings,
+        "owner_return_finding_count": len(owner_returns),
+        "task_record_ids": tasks,
+        "message_id_count": messages,
+        "counterparts": sorted(counterparts),
+        "collection_complete": True,
+        "incomplete_counterpart_count": 0,
+    }
+
+
+def _candidate_evidence(candidates: list[dict[str, Any]]) -> dict[str, Any]:
+    return {
+        "candidate_count": len(candidates),
+        "dispositions": dict(Counter(c["disposition"] for c in candidates)),
+        "stop_reasons": dict(Counter(c["stop_reason"] for c in candidates if c["stop_reason"])),
+        "candidates": [
+            {k: c.get(k) for k in ("triggering_event_id", "task_record_id", "message_id",
+                                   "disposition", "stop_reason")}
+            for c in candidates[:MAX_EVIDENCE_ROWS]
+        ],
+    }
+
+
+def _receipt_evidence(receipts: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    keys = ("triggering_event_id", "idempotency_key", "task_record_id", "message_id",
+            "status", "ops_run_id", "state")
+    return [{k: r.get(k) for k in keys} for r in receipts[:MAX_EVIDENCE_ROWS]]
+
+
+async def run_owner_return_pickup(
+    params: Mapping[str, Any],
+    *,
+    session_factory: Callable[[], Any] | None = None,
+    client: httpx.Client | None = None,
+    state_hub_url: str | None = None,
+    forge_api: str = DEFAULT_FORGE_API,
+    today: date | None = None,
+    resolve_path: Callable[[str], Path] | None = None,
+) -> dict[str, Any]:
+    """Plan (and, when ``submit`` and the gate allow, submit) one repository."""
+    validate_params(params)
+    target_repo: str = params["target_repo"]
+    submit: bool = params["submit"]
+    # Fail fast before any read when submission is requested but not gated on.
+    if submit and not orp.pickup_enabled():
+        raise OwnerReturnPickupRefused(f"{orp.ENABLE_ENV} is not enabled")
+    if submit and session_factory is None:
+        raise OwnerReturnPickupRefused("submission requires a database session factory")
+
+    if resolve_path is None:
+        from activity_core.runtime_paths import resolve_runtime_path as resolve_path
+    report = select_repo_report(
+        load_report(resolve_path(params["report_file"]), params["report_sha256"]), target_repo
+    )
+    planner_input = bind_report(report, target_repo, params.get("expected_source_revision"))
+    revision = planner_input["source_revision"]
+    observed_on = today or datetime.now(tz=timezone.utc).date()
+
+    evidence: dict[str, Any] = {
+        "kind": OPERATION_ID,
+        "mode": "submit" if submit else "plan-only",
+        "target_repo": target_repo,
+        "source_revision": revision,
+        "report_sha256": params["report_sha256"],
+        "observed_on": observed_on.isoformat(),
+        "planner_input": planner_input,
+        "planner_output": _candidate_evidence([]),
+        "receipts": [],
+        "receipt_counts": {"inserted": 0, "duplicate": 0, "skipped": 0},
+        "submitted_count": 0,
+    }
+    if not planner_input["task_record_ids"]:
+        return evidence
+
+    own_client = client is None
+    http = client or httpx.Client(timeout=30.0)
+    try:
+        hub = (state_hub_url or _state_hub_url()).rstrip("/")
+        mapping = params.get("predecessor_repos") or {}
+        built = build_current_tasks(
+            target_repo=target_repo,
+            scan_revision=revision,
+            task_record_ids=planner_input["task_record_ids"],
+            client=http,
+            decision_status=state_hub_decision_status(http, hub),
+            external_status=forge_head_status_resolver(http, mapping, forge_api=forge_api)
+            if mapping else None,
+            require_current_head=True,
+            report=report,
+            forge_api=forge_api,
+        )
+    except (ValueError, httpx.HTTPError) as exc:
+        raise OwnerReturnPickupRefused(f"eligibility refused: {exc}") from exc
+    finally:
+        if own_client:
+            http.close()
+    if built.get("source_revision") != revision:
+        raise OwnerReturnPickupRefused("current_tasks are bound to a different revision")
+    planner_input["current_tasks_schema"] = built.get("schema")
+    planner_input["current_tasks_source_revision"] = built["source_revision"]
+    planner_input["current_tasks_count"] = len(built["current_tasks"])
+    planner_input["head_verified"] = True
+
+    try:
+        candidates = planner.plan_owner_returns(
+            report,
+            activity_definition_id=orp.ACTIVITY_DEFINITION_ID,
+            current_tasks=built["current_tasks"],
+            observed_on=observed_on,
+        )
+    except ValueError as exc:
+        raise OwnerReturnPickupRefused(f"planner refused: {exc}") from exc
+    for candidate in candidates:
+        if candidate.get("target_repo") != target_repo or candidate.get(
+            "source_revision", revision
+        ) != revision:
+            raise OwnerReturnPickupRefused("planner candidate is bound to a different repository/revision")
+    evidence["planner_output"] = _candidate_evidence(candidates)
+
+    reviewable = [c for c in candidates if c["disposition"] == "reviewable"]
+    if not submit or not reviewable:
+        return evidence
+    async with session_factory() as session:
+        async with session.begin():
+            receipts = await orp.submit_owner_returns(
+                session, reviewable, collection_complete=planner_input["collection_complete"]
+            )
+    evidence["submitted_count"] = len(reviewable)
+    evidence["receipts"] = _receipt_evidence(receipts)
+    counts = Counter(r["status"] for r in receipts)
+    evidence["receipt_counts"] = {k: counts.get(k, 0) for k in ("inserted", "duplicate", "skipped")}
+    return evidence
+
+
+def _state_hub_url() -> str:
+    import os
+
+    from activity_core.automation_status import DEFAULT_STATE_HUB_URL
+
+    return os.environ.get("STATE_HUB_URL", DEFAULT_STATE_HUB_URL)

src/activity_core/vendored/__init__.py

@@ -0,0 +1,5 @@
+"""Verbatim copies of small upstream modules the worker cannot import at runtime.
+
+Each file is a byte-identical copy; drift is detected by tests when the
+upstream checkout is present. Never edit a vendored file locally -- re-copy it.
+"""

src/activity_core/vendored/state_hub_owner_return_handoff.py

@@ -0,0 +1,159 @@
+"""Offline C-25 → existing Activity Core ops-run handoff planning.
+
+No network, scheduling, dispatch or task mutation. An admitted ActivityDefinition
+and owner harness are prerequisites for a caller to use these candidate IDs.
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+import re
+import uuid
+from collections.abc import Mapping, Sequence
+from datetime import date
+
+_REPO = re.compile(r"^[a-z0-9][a-z0-9-]*$")
+_TASK = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$")
+_STATES = {"open", "claimed", "succeeded", "failed", "expired"}
+_SHA = re.compile(r"^[0-9a-f]{40}$")
+
+
+def _uuid(value: object) -> str:
+    try:
+        return str(uuid.UUID(str(value)))
+    except (ValueError, TypeError, AttributeError):
+        raise ValueError("invalid ops-run/message UUID") from None
+
+
+def _eligibility_stop(current: Mapping, *, repo: str, task: str,
+                      marker: object, observed_on: date) -> str | None:
+    required = {"target_repo", "task_record_id", "status", "workplan_status",
+                "workplan_flavor", "needs_human", "blocked_on", "depends_on",
+                "predecessor_statuses", "wait_until", "decision_id", "decision_status"}
+    if not isinstance(current, Mapping) or not required <= current.keys():
+        raise ValueError("missing authoritative task eligibility fields")
+    if current["target_repo"] != repo or current["task_record_id"] != task:
+        return "stale_task_identity"
+    if type(current["needs_human"]) is not bool:
+        raise ValueError("needs_human must be an authoritative boolean")
+    if current["needs_human"]:
+        return "human_gate"
+    if current["decision_id"] is not None and current["decision_status"] != "resolved":
+        return "decision_not_resolved_or_unknown"
+    if current["workplan_flavor"] == "residual":
+        return "unpromoted_residual"
+    if current["workplan_status"] not in {"proposed", "ready", "active", "blocked", "backlog"}:
+        return "workplan_not_open"
+    if current["status"] != "wait":
+        return "task_not_waiting"
+    if not isinstance(marker, str) or not re.fullmatch(r"message-from:[a-z0-9][a-z0-9-]*", marker):
+        raise ValueError("missing owner-return marker in finding")
+    if current["blocked_on"] != marker:
+        return "stale_owner_return_marker"
+    dependencies = current["depends_on"]
+    statuses = current["predecessor_statuses"]
+    if not isinstance(dependencies, list) or not isinstance(statuses, Mapping):
+        raise ValueError("invalid authoritative predecessor eligibility")
+    if any(not isinstance(item, str) or not item for item in dependencies):
+        raise ValueError("invalid predecessor identity")
+    if any(statuses.get(item) not in {"done", "cancel", "finished", "archived"}
+           for item in dependencies):
+        return "predecessor_not_terminal_or_unknown"
+    wait_until = current["wait_until"]
+    if wait_until is not None:
+        if type(wait_until) is date:
+            until = wait_until
+        elif isinstance(wait_until, str) and re.fullmatch(r"\d{4}-\d{2}-\d{2}", wait_until):
+            until = date.fromisoformat(wait_until)
+        else:
+            raise ValueError("invalid authoritative wait_until date")
+        if until > observed_on:
+            return "future_wait_window"
+    return None
+
+
+def plan_owner_returns(
+    report: Mapping, *, activity_definition_id: str, current_tasks: Mapping,
+    observed_on: date, ops_runs: Sequence[Mapping] = ()
+) -> list[dict]:
+    """Deduplicate by repository, task and message against correlated ops runs.
+
+    Existing open/claimed rows are pending, succeeded rows are completed, and
+    failed/expired rows require owner review. None is dispatched again here.
+    A succeeded automation receipt does not itself close a workplan task.
+    Callers must build current_tasks from fresh authoritative source, resolving
+    every predecessor; checker findings alone contain insufficient safety data.
+    """
+    if not isinstance(current_tasks, Mapping) or type(observed_on) is not date:
+        raise ValueError("authoritative eligibility and observation date required")
+    definition = _uuid(activity_definition_id)
+    repo = report.get("repo_slug")
+    if not isinstance(repo, str) or not _REPO.fullmatch(repo):
+        raise ValueError("invalid repository identity")
+    if report.get("result") == "error":
+        raise ValueError("cannot plan from failed checker collection")
+    # Passed through unchanged so candidates stay bound to the scanned commit;
+    # None (unknown/dirty scan) travels as None and the eligibility builder
+    # refuses it. Never substituted or guessed here.
+    revision = report.get("source_revision")
+    if revision is not None and (not isinstance(revision, str) or not _SHA.fullmatch(revision)):
+        raise ValueError("source_revision must be a full 40-hex commit SHA or null")
+    issues = report.get("issues")
+    if not isinstance(issues, list):
+        raise ValueError("checker issues must be a list")
+    candidates = {}
+    for issue in issues:
+        if not isinstance(issue, Mapping):
+            raise ValueError("invalid checker issue")
+        if issue.get("check_id") != "C-25" or "owner_return" not in issue:
+            continue
+        if issue.get("repo_slug", repo) != repo or issue.get("source_revision", revision) != revision:
+            raise ValueError("C-25 finding is bound to a different repository or revision")
+        context = issue["owner_return"]
+        if not isinstance(context, Mapping):
+            raise ValueError("invalid owner-return context")
+        task = context.get("task_record_id")
+        messages = context.get("message_ids")
+        if not isinstance(task, str) or not _TASK.fullmatch(task):
+            raise ValueError("missing task identity")
+        if not isinstance(messages, list) or not messages:
+            raise ValueError("missing message identities")
+        if task not in current_tasks:
+            raise ValueError("missing authoritative task eligibility")
+        stop = _eligibility_stop(current_tasks[task], repo=repo, task=task,
+                                 marker=issue.get("file_value"), observed_on=observed_on)
+        for raw_message in messages:
+            message = _uuid(raw_message)
+            encoded = json.dumps([repo, task, message], separators=(",", ":")).encode()
+            event = "owner-return:" + hashlib.sha256(encoded).hexdigest()
+            candidates[event] = {
+                "target_repo": repo, "source_revision": revision,
+                "task_record_id": task, "message_id": message,
+                "activity_definition_id": definition, "triggering_event_id": event,
+                "disposition": "stopped" if stop else "reviewable",
+                "stop_reason": stop, "ops_run_id": None,
+            }
+    matched = set()
+    for row in ops_runs:
+        if not isinstance(row, Mapping):
+            raise ValueError("invalid ops-run receipt")
+        event = row.get("triggering_event_id")
+        if event not in candidates or row.get("target_repo") != repo:
+            continue
+        if _uuid(row.get("activity_definition_id")) != definition:
+            continue
+        if event in matched:
+            raise ValueError("multiple correlated ops runs require owner reconciliation")
+        matched.add(event)
+        state = row.get("state")
+        if state not in _STATES:
+            raise ValueError("invalid correlated ops-run state")
+        candidate = candidates[event]
+        candidate["ops_run_id"] = _uuid(row.get("id"))
+        if candidate["disposition"] == "stopped":
+            continue
+        candidate["disposition"] = {
+            "open": "pending", "claimed": "pending", "succeeded": "completed",
+            "failed": "review_required", "expired": "review_required",
+        }[state]
+    return [candidates[key] for key in sorted(candidates)]