diff --git a/WORK-RECORDS.md b/WORK-RECORDS.md
index 54097d0..9341fc4 100644
--- a/WORK-RECORDS.md
+++ b/WORK-RECORDS.md
@@ -68,6 +68,7 @@
 | workplan | STATE-WP-0092 | finished | — | workplans/STATE-WP-0092-workplan-flavor-and-depends-on-policy.md |
 | workplan | STATE-WP-0093 | finished | — | workplans/STATE-WP-0093-broadcast-receipts-and-standing-notices.md |
 | workplan | STATE-WP-0094 | active | — | workplans/STATE-WP-0094-routed-gates-and-owner-returns.md |
+| workplan | STATE-WP-0095 | active | — | workplans/STATE-WP-0095-historical-source-receiving.md |
 | task | STATE-WP-ADHOC-2026-06-04-T01 | done | — | workplans/ADHOC-2026-06-04.md |
 | task | STATE-WP-ADHOC-2026-07-01-T01 | done | — | workplans/ADHOC-2026-07-01.md |
 | task | STATE-WP-ADHOC-2026-07-01-T02 | done | — | workplans/ADHOC-2026-07-01.md |
@@ -377,6 +378,10 @@
 | task | STATE-WP-0093-T08 | done | — | workplans/STATE-WP-0093-broadcast-receipts-and-standing-notices.md |
 | task | STATE-WP-0094-T01 | done | — | workplans/STATE-WP-0094-routed-gates-and-owner-returns.md |
 | task | STATE-WP-0094-T02 | done | — | workplans/STATE-WP-0094-routed-gates-and-owner-returns.md |
-| task | STATE-WP-0094-T03 | todo | — | workplans/STATE-WP-0094-routed-gates-and-owner-returns.md |
+| task | STATE-WP-0094-T03 | progress | — | workplans/STATE-WP-0094-routed-gates-and-owner-returns.md |
 | task | STATE-WP-0094-T04 | done | — | workplans/STATE-WP-0094-routed-gates-and-owner-returns.md |
 | task | STATE-WP-0094-T05 | progress | — | workplans/STATE-WP-0094-routed-gates-and-owner-returns.md |
+| task | STATE-WP-0095-T01 | done | — | workplans/STATE-WP-0095-historical-source-receiving.md |
+| task | STATE-WP-0095-T02 | done | — | workplans/STATE-WP-0095-historical-source-receiving.md |
+| task | STATE-WP-0095-T03 | wait | — | workplans/STATE-WP-0095-historical-source-receiving.md |
+| task | STATE-WP-0095-T04 | wait | — | workplans/STATE-WP-0095-historical-source-receiving.md |
diff --git a/api/services/forge_projection.py b/api/services/forge_projection.py
index 032f41d..28dd491 100644
--- a/api/services/forge_projection.py
+++ b/api/services/forge_projection.py
@@ -154,6 +154,7 @@ class DerivedWorkplan:
     flavor: str | None = None
     flavor_promotion_reason: str | None = None
     flavor_promoted_from: str | None = None
+    historical_preservation: bool = False
 
 
 @dataclass
@@ -205,6 +206,7 @@ class DerivedProjection:
                     "flavor_promoted_from": w.flavor_promoted_from,
                     "relative_path": w.relative_path,
                     "archived": w.archived,
+                    "historical_preservation": w.historical_preservation,
                     "tasks": [
                         {
                             "record_id": t.record_id,
@@ -444,7 +446,40 @@ def derive_from_checkout(repo_root: Path, repo_slug: str, commit: str) -> Derive
     if not wp_dir.is_dir():
         proj.records_source_present = False
         return proj
+    # Exact immutable histories are preserved by existing UUID, never allocated.
+    from api.services.historical_projection import bound_sources
+    try:
+        historical = bound_sources(repo_root, repo_slug)
+    except (ValueError, OSError, UnicodeError) as exc:
+        raise ForgeDeriveError(str(exc)) from exc
+    consumed = {entry["path"] for entry, _text in historical}
+    groups = {}
+    for entry, text in historical:
+        meta, body = _split_frontmatter(text)
+        parent_id = entry["parent_record_id"]
+        # Split views carry the parent's existing UUID and repeat its task list.
+        # Only the reviewed archived parent supplies the projected row.
+        if entry["record_id"] != parent_id:
+            continue
+        if repo_slug == "activity-core":
+            tasks = [DerivedTask(record_id=item["id"], uuid=item["state_hub_task_id"],
+                                 title=item.get("title"), status=item.get("status"),
+                                 priority=item.get("priority")) for item in meta["tasks"]]
+        else:
+            tasks = _parse_tasks(body, parent_id)
+            bindings = [yaml.safe_load(m.group(1)) for m in _TASK_BLOCK_RE.finditer(body)]
+            for task, binding in zip(tasks, bindings, strict=True):
+                task.uuid = binding["state_hub_task_id"]
+        groups[parent_id] = DerivedWorkplan(
+            record_id=parent_id, uuid=entry["uuid"], title=meta.get("title"),
+            status=meta.get("status"), relative_path=entry["path"],
+            archived="/archived/" in entry["path"], tasks=tasks,
+            historical_preservation=True,
+        )
+    proj.workplans.extend(groups.values())
     for path in sorted(wp_dir.rglob("*.md")):
+        if path.relative_to(repo_root).as_posix() in consumed:
+            continue
         if path.name.startswith("."):
             continue
         try:
@@ -646,7 +681,11 @@ def diff_against_hub(
     want_wp = {w.record_id.strip().lower(): w for w in derived.workplans}
     have_wp: dict[str, dict[str, Any]] = {}
     want_paths = {_path_key(w.relative_path): k for k, w in want_wp.items()}
+    preserved_by_uuid = {w.uuid: key for key, w in want_wp.items() if w.historical_preservation}
     for w in hub_workplans:
+        if str(w["id"]) in preserved_by_uuid:
+            have_wp[preserved_by_uuid[str(w["id"])]] = w
+            continue
         slug = str(w.get("slug") or "")
         key = slug.strip().lower()
         if key not in want_wp:
@@ -686,6 +725,11 @@ def diff_against_hub(
         uid = str(cur["id"]) if cur else w.uuid
         if not cur:
             continue
+        if w.historical_preservation:
+            if cur.get("status") not in {"done", "finished", "archived", "completed"}:
+                d.differing.append({"kind": "workplan", "record_id": w.record_id, "uuid": uid,
+                                    "changed": {"historical_terminal_status": cur.get("status")}})
+            continue
         changed = {}
         if (cur.get("status") or None) != (w.status or None):
             changed["status"] = {"hub": cur.get("status"), "forge": w.status}
@@ -707,6 +751,18 @@ def diff_against_hub(
             cur = have_wp.get(w.record_id.strip().lower())
             if cur:
                 hub_rows = hub_tasks_by_workplan.get(str(cur["id"]), [])
+        if w.historical_preservation:
+            by_uuid = {str(task["id"]): task for task in hub_rows}
+            for task in w.tasks:
+                current = by_uuid.get(task.uuid)
+                if current is None:
+                    d.missing.append({"kind": "task", "record_id": task.record_id,
+                                      "uuid": task.uuid, "workplan": w.record_id})
+                elif current.get("status") not in {"done", "cancel"}:
+                    d.differing.append({"kind": "task", "record_id": task.record_id,
+                                        "uuid": task.uuid, "changed": {
+                                            "historical_terminal_status": current.get("status")}})
+            continue
         # Match on the canonical record id where the hub has one. Rows created
         # before STATE-WP-0083-T06 fall back to title, which is why those are
         # reported rather than acted on: a renamed heading is indistinguishable
@@ -1312,6 +1368,27 @@ async def reset_repository_projection(
         for task_row in loaded:
             tasks_by_wp.setdefault(task_row.workplan_id, []).append(task_row)
 
+    # Bound histories are receiving support for already existing records only.
+    # Verify every parent and task before any mutation; preserve labels, status,
+    # retirement stamps and all UUIDs. Missing rows cannot authorize recreation.
+    for key, workplan in want.items():
+        if not workplan.historical_preservation:
+            continue
+        row = matched.get(key)
+        local = {str(task.id): task for task in tasks_by_wp.get(row.id, [])} if row else {}
+        missing = [task.uuid for task in workplan.tasks if task.uuid not in local]
+        nonterminal = [task.uuid for task in workplan.tasks
+                       if task.uuid in local and local[task.uuid].status not in {"done", "cancel"}]
+        if (row is None or str(row.id) != workplan.uuid or missing or nonterminal
+                or row.status not in {"done", "finished", "archived", "completed"}):
+            outcome.status = "refused"
+            outcome.refused.append({
+                "reason": "historical binding requires existing parent and task UUIDs",
+                "workplan": workplan.record_id, "uuid": workplan.uuid,
+                "missing_task_uuids": missing, "nonterminal_task_uuids": nonterminal,
+            })
+            return outcome
+
     # Canonical task identity is scoped to the matched current workplan while
     # reconciling. Duplicate rows attached to a retired/displaced workplan are
     # historical evidence and are deliberately not loaded above. Two rows with
@@ -1374,22 +1451,11 @@ async def reset_repository_projection(
                 creating_tasks[task.uuid] = (workplan, task)
 
     if creating_tasks:
-        current_holder_filters = [Workplan.projection_retired_at.is_(None)]
-        if stale:
-            # The caller has already acknowledged these retirements (the
-            # unacknowledged path returned above). Their tasks are preserved as
-            # history, but must not prevent the replacement projection from
-            # claiming its deterministic identities in the same transaction.
-            current_holder_filters.append(
-                Workplan.id.not_in([workplan.id for workplan in stale])
-            )
         task_holders = list(
             (
                 await session.execute(
                     select(Task)
-                    .join(Workplan, Workplan.id == Task.workplan_id)
                     .where(
-                        *current_holder_filters,
                         Task.id.in_(
                             [uuid.UUID(task_id) for task_id in creating_tasks]
                         ),
@@ -1403,7 +1469,7 @@ async def reset_repository_projection(
                 workplan, task = creating_tasks[str(holder.id)]
                 outcome.refused.append(
                     {
-                        "reason": "derived task identifier already belongs to another current workplan",
+                        "reason": "derived task identifier already belongs to another workplan",
                         "kind": "task",
                         "record_id": task.record_id,
                         "uuid": task.uuid,
@@ -1463,6 +1529,10 @@ async def reset_repository_projection(
             outcome.created.append(w.record_id)
             continue
 
+        if w.historical_preservation:
+            outcome.notes.append("Preserved bound historical row and task UUIDs: " + w.record_id)
+            continue
+
         changed = False
         # The title is derived like every other field, and not syncing it left
         # `cust-wp-0010` reading "Domain and Repository Goals" while its file
diff --git a/api/services/historical_projection.py b/api/services/historical_projection.py
new file mode 100644
index 0000000..37c1d9a
--- /dev/null
+++ b/api/services/historical_projection.py
@@ -0,0 +1,90 @@
+"""Exact closed source bindings predating canonical Forge identity.
+
+These are frozen historical source views, not aliases or UUID allocation rules.
+An altered/missing source requires explicit owner review; generic source UUIDs
+remain untrusted. Shared Activity views represent the one archived parent.
+"""
+from pathlib import Path
+from hashlib import sha256
+import json
+
+BINDINGS = json.loads(r'''[
+  {
+    "repo": "activity-core",
+    "path": "workplans/archived/260707-custodian-WP-0003-event-bridge.md",
+    "sha256": "83ea8bcb151e003419b73c59dbd028f4a1ce3d1120a82e4a128a772503ff3b76",
+    "record_id": "custodian-WP-0003",
+    "uuid": "b4eb45a9-69e3-4ab0-b00c-67a53c3117c5",
+    "parent_record_id": "custodian-WP-0003"
+  },
+  {
+    "repo": "activity-core",
+    "path": "workplans/custodian-WP-0001-temporal-backbone.md",
+    "sha256": "020e2eac973f2daa41c1687f8e867da9d521f945547e6aa82c87993b3ff6fc64",
+    "record_id": "custodian-WP-0001",
+    "uuid": "e22516cc-4fdf-4dad-a928-a9e04a9fa7c7",
+    "parent_record_id": "custodian-WP-0001"
+  },
+  {
+    "repo": "activity-core",
+    "path": "workplans/custodian-WP-0002-triggers-ops.md",
+    "sha256": "d8cbc61631c01b6f11a03a63f2fb6275fc853ff7f7459ce65973ed1a233daa01",
+    "record_id": "custodian-WP-0002",
+    "uuid": "3a4f47d9-8bc1-434e-acb4-bed5d4dacda0",
+    "parent_record_id": "custodian-WP-0002"
+  },
+  {
+    "repo": "activity-core",
+    "path": "workplans/custodian-WP-0003a-event-bridge-model-rules-registry.md",
+    "sha256": "ffa6682e9ba3aaf559bd44203d561c90dea69bcf1b1e778209a7b209e6e5cdd3",
+    "record_id": "custodian-WP-0003a",
+    "uuid": "b4eb45a9-69e3-4ab0-b00c-67a53c3117c5",
+    "parent_record_id": "custodian-WP-0003"
+  },
+  {
+    "repo": "activity-core",
+    "path": "workplans/custodian-WP-0003b-event-bridge-parser-workflow-triggers.md",
+    "sha256": "4fc34e0f821e2df51e5c08c7ab34da35b65690381d01f0cfdf55d00c970b414c",
+    "record_id": "custodian-WP-0003b",
+    "uuid": "b4eb45a9-69e3-4ab0-b00c-67a53c3117c5",
+    "parent_record_id": "custodian-WP-0003"
+  },
+  {
+    "repo": "activity-core",
+    "path": "workplans/custodian-WP-0003c-event-bridge-context-integration.md",
+    "sha256": "b6cd62b26515e509670ca39736b6bcec9633beedd46c9bd8f7598c06bac7276b",
+    "record_id": "custodian-WP-0003c",
+    "uuid": "b4eb45a9-69e3-4ab0-b00c-67a53c3117c5",
+    "parent_record_id": "custodian-WP-0003"
+  },
+  {
+    "repo": "activity-core",
+    "path": "workplans/custodian-WP-0004-railiance-ops.md",
+    "sha256": "20df02eea740e109c937130e7f8801a636fc0990ed8c6d2ced55659f8957adb1",
+    "record_id": "custodian-WP-0004",
+    "uuid": "759b1255-aa78-42b5-8fab-c5fcf168f5f4",
+    "parent_record_id": "custodian-WP-0004"
+  },
+  {
+    "repo": "the-custodian",
+    "path": "workplans/ADHOC-2026-08-25-retired.md",
+    "sha256": "344a1293dbb3a67a944eb34ca4a0337487d8f047226027a2107922346fbab2dd",
+    "record_id": "CUST-WP-ADHOC-2026-08-25-RETIRED",
+    "uuid": "1c53d3db-6633-5e2d-987f-10706d73be9e",
+    "parent_record_id": "CUST-WP-ADHOC-2026-08-25-RETIRED"
+  }
+]''')
+
+
+def bound_sources(repo_root: Path, repo_slug: str):
+    entries = [entry for entry in BINDINGS if entry["repo"] == repo_slug]
+    result = []
+    for entry in entries:
+        path = repo_root / entry["path"]
+        if path.is_symlink() or not path.is_file():
+            raise ValueError("Historical source binding missing or symlinked: " + entry["path"])
+        data = path.read_bytes()
+        if sha256(data).hexdigest() != entry["sha256"]:
+            raise ValueError("Historical source binding changed: " + entry["path"])
+        result.append((entry, data.decode("utf-8")))
+    return result
diff --git a/docs/evidence/2026-10-04-historical-receiving-source.json b/docs/evidence/2026-10-04-historical-receiving-source.json
new file mode 100644
index 0000000..1be6867
--- /dev/null
+++ b/docs/evidence/2026-10-04-historical-receiving-source.json
@@ -0,0 +1,112 @@
+{
+  "validation": {
+    "actual_source_cases": 155,
+    "elapsed_seconds": 34.72
+  },
+  "historical_snapshot_proof": {
+    "scope": "offline exact historical receiving proof; no API writes",
+    "repos": {
+      "activity-core": {
+        "parent_uuids": [
+          "3a4f47d9-8bc1-434e-acb4-bed5d4dacda0",
+          "759b1255-aa78-42b5-8fab-c5fcf168f5f4",
+          "b4eb45a9-69e3-4ab0-b00c-67a53c3117c5",
+          "e22516cc-4fdf-4dad-a928-a9e04a9fa7c7"
+        ],
+        "task_uuids": [
+          "3dc749eb-7a73-4199-9c8d-e8f31bac5372",
+          "107115b0-4476-424d-84ff-03992a85d40b",
+          "aa638595-0bca-41ab-b140-b077d05994e1",
+          "a03c823e-5873-4586-912a-6d1370383c7b",
+          "70e7f742-378f-4129-a0f4-f25ccaccffd0",
+          "61cb7426-e0ec-4a52-870d-18a26b3c4f3e",
+          "f28a405d-ddcc-44d8-b443-f26f3ab34eaa",
+          "d9fe3e54-ec1a-4e23-aa76-d8869f4e024d",
+          "47774e01-1026-478e-9a46-7d676bfed45c",
+          "0a74f29f-c07d-4338-be90-e5cf4087261b",
+          "491a6903-8189-43bb-958f-4d16abc84f8e",
+          "f24662ff-4a26-48bd-b997-57e7586c7f11",
+          "e0205c56-1d40-4142-952b-e27ff6a44e1d",
+          "b05f046f-a6ba-4d96-a298-a0bbea067427",
+          "2417912f-845d-489a-ace4-fb9280d3b679",
+          "b7decbb6-ad2b-4fa5-8efc-05a7eb435d76",
+          "e019cb5a-adf0-4a5d-9410-c41810128190",
+          "fb6b3440-47d2-4b0a-97c1-6e780cc497c4",
+          "70a7365f-3042-4770-b3cd-3c6724b0790d",
+          "1da921f5-86a8-488f-a015-402079194e10",
+          "f72bba1a-eb24-496e-9498-f4676facc5c9",
+          "e50550d1-9904-41d7-afd8-492a1f1e91b8",
+          "5a1f7fa3-acb9-4f60-9892-c9eaa120272e",
+          "00231668-95c5-447f-b3d0-1fb8c20b487f",
+          "7abfd375-ea9d-4209-8371-e5664dc2c6c4",
+          "68b6610b-159c-4f1c-92a9-7128efea0961",
+          "9348efea-a7e9-4f92-b866-8fc82cf28fee",
+          "cac1f45a-7391-471a-9566-97cdbd96eb2d",
+          "7f10b5a3-7cad-4914-b603-d57508c85629",
+          "b27e54a1-5dcc-476d-8f4a-c995aea6a8c2",
+          "0eafb60c-f00e-4fd7-a921-7de75fcfe81e",
+          "7bdfc5c2-1f06-4fce-aac3-fae036dcb47e",
+          "766d602d-1b23-4247-a46d-03c0d3b8e498",
+          "ca1bf66f-7094-459c-9abf-5f5c6414c91a",
+          "54c25eae-4fad-42d8-bb15-9c1e7532425e",
+          "a1ed0d8b-df59-4af1-82a1-d01628919689",
+          "cdd349f1-b7ad-4a3a-afa0-ae671a7addb8",
+          "4cd9833c-fbc4-4b10-b6f2-85c028c8c557",
+          "38177fcf-c468-4424-9938-1b01038a386b",
+          "65d77939-6972-4450-993e-23ccb25d9454",
+          "7a182265-4013-4272-8540-cfb4e2079eb3",
+          "229d99ca-2d09-4c96-b3d2-da8b2d14c5b7",
+          "78b9d642-17b1-46c7-8e5f-c0a948821993",
+          "a86dcffe-789a-482e-bc5f-a5ac9db9608e",
+          "a714c511-9f93-4594-b745-9330bc645384",
+          "615dec10-7e63-46b3-80ad-83ff2e51d6ba",
+          "1e06cb43-d0e3-4af5-9216-0877cac2082d",
+          "01b507e3-6420-422f-83ed-90c7b4d49bb7",
+          "673ca7b2-49cc-4314-8be2-9848f686da37",
+          "7ba289e9-fef6-469a-9504-c3664227a2f7",
+          "dac18c7a-a663-4876-ba41-7378094148ab",
+          "e4ba0c93-0940-4d57-aeb6-80d20749ee2b",
+          "24a877f0-1653-4cf2-9e4f-50ed53cbc34c",
+          "c7f5f5c3-2958-4f0c-ab3a-0b0a0374bf67",
+          "95a5edb2-a299-45e1-a7a9-48ecbbce13eb",
+          "7cbcc6db-7c07-4b37-8fd1-dc0a87d93173",
+          "73bf70ef-7969-434d-99d2-7a5787169d94",
+          "2a81f9ba-47cb-480f-a5d3-af74fe238485",
+          "67981de4-b766-4c47-9985-9573297ac464",
+          "7c7a1617-29e9-4ea5-a675-fd75e325c451",
+          "2a2a8d02-19dc-4b09-8792-f855ed48388a",
+          "55ac37b5-5606-42e5-bf11-e8dae3a188b7",
+          "65b3229b-5f03-450d-a263-f2c205be3d28",
+          "83dcd765-5715-499f-be23-f728d0261cfb"
+        ],
+        "parent_count": 4,
+        "bound_task_count": 64,
+        "reset_status": "noop",
+        "no_mutations": true,
+        "diff_clean": true
+      },
+      "the-custodian": {
+        "parent_uuids": [
+          "1c53d3db-6633-5e2d-987f-10706d73be9e"
+        ],
+        "task_uuids": [
+          "d8af3e6b-26ae-580c-bbeb-caa8ad419bec"
+        ],
+        "parent_count": 1,
+        "bound_task_count": 1,
+        "reset_status": "noop",
+        "no_mutations": true,
+        "diff_clean": true
+      }
+    }
+  },
+  "original_preparation_cases": 108,
+  "registrar_preparation_cases": 38,
+  "registrar_applied": false,
+  "receiving_deployed": false,
+  "remaining_owner_workplans": [
+    "STATE-WP-0095",
+    "RMGR-WP-0019"
+  ],
+  "binding_policy": "Exact reviewed full SHA256 frozen histories; no generic source-UUID trust, no relabel/recreate/unretire/move"
+}
diff --git a/docs/owner-return-handoff.md b/docs/owner-return-handoff.md
new file mode 100644
index 0000000..22612ba
--- /dev/null
+++ b/docs/owner-return-handoff.md
@@ -0,0 +1,107 @@
+# Owner-return pickup preparation
+
+STATE-WP-0094-T03 selects the existing Activity Core ActivityDefinition → ops_run
+→ admitted owner harness path for recurring pickup. State Hub does not claim
+ops runs and its workplan launch queue is retired. This selection is a source
+integration boundary, not evidence that a definition or owner harness is admitted.
+
+`scripts/owner_return_handoff.py` consumes the existing consistency checker's
+C-25 JSON. It produces deterministic event identities from repository, task
+record ID and message UUID. Duplicate findings and reordered scans yield the
+same candidates. Eligible candidates are reviewable owner returns, not unblocked
+tasks. The event is suitable for TaskSpec.triggering_event_id; the
+existing Activity Core idempotency key also includes the admitted definition
+and source IDs. Supply the approved ActivityDefinition UUID explicitly.
+
+Before any future dispatch, obtain fresh Activity Core `/ops-runs` rows for the
+configured definition/repository. The planner correlates definition, repository
+and event: open/claimed remain pending, succeeded is completed pickup, and
+failed/expired require owner reconciliation. Duplicate correlated rows fail
+closed. Completed pickup does not imply the task's blocker is resolved or its
+file status should change. Malformed identities and checker collection errors
+fail closed. The module never opens a socket, schedules, dispatches, persists a
+receipt or changes a task. It does not invent a competing receipt ledger.
+
+## Remaining T03 integration
+
+The Activity Core owner must identify an admitted definition, pinned harness
+profile and repository authority for the pickup; accept event/source binding;
+and supply the callable submission/receipt adapter. Use the existing ops-run
+unique key, lease and exact-repeat terminal close semantics. A list snapshot
+alone is not atomic dispatch: actual insertion must retain that server-side
+uniqueness. Message results must be paginated upstream: C-25 currently fetches at
+most 200 replies per counterpart, so a planner cannot claim complete coverage.
+An owner-approved review must precede live activation. No schedule was changed.
+
+References: Activity Core `docs/execution-queue-boundary.md`,
+`docs/ops-run-queue.md`, `src/activity_core/ops_run_queue.py` (TaskSpec key and row
+projection); State Hub `api/services/execution_queue.py` and
+`api/routers/ops_runs.py` (read-only summary).
+
+## T05 graph freshness assessment
+
+The authorized receiving rollout and immutable qualifier/null-clear parity are
+recorded separately in `docs/evidence/2026-10-04-receiving-rollout.json`.
+They do not establish task-edge or scheduled freshness coverage.
+
+Current `api/services/forge_projection.py` neither derives `depends_on` edges nor
+reconciles WorkplanDependency rows. The snapshot returns existing edges, which
+is not evidence that normal Forge receiving maintains them. In contrast,
+`scripts/consistency_check.py` C-20 indexes frontmatter and task-block edges,
+using `from_task_id`; its fix path calls the dependency API. C-40 independently
+reads authoritative source target status. Task-level dependency schema support
+therefore exists while the normal receiving gap remains real.
+
+A bounded next receiving design must define full source edge ownership,
+cross-repository target resolution, missing/ambiguous targets, edge removal,
+transactional failure and repeat noop; it must preserve externally authored
+relations instead of deleting every dependency row. Adding graph receiving
+without that contract would create another destructive projection risk.
+
+Current chart values disable legacy sweep; workstation checkout traversal is
+excluded by ADR-002. Activity Core owns scheduling and Repo Manager owns the
+C-rule engine. No source reading or one successful normal sync establishes
+actual live schedule coverage, freshness SLA or full repository coverage.
+Retain T05 progress until accepted edge receiving and bounded freshness evidence
+include exact runner revision, active definition, per-repository source commit,
+last successful receipt, failed/queued receipts and measured freshness window.
+Historical registrar failures remain explicit coverage holes. Do not revive
+legacy --all --remote sweeping or collapse T03 pickup into T05 index freshness.
+
+## Validation
+
+The staged offline module and regression suite passed 13 tests on 2026-10-04.
+The tests cover pair identity and replay, existing pending/success/failure rows,
+wrong-definition/repository/event receipts, ambiguous duplicate receipts,
+malformed message identity, collection error and legacy workplan-only warnings.
+This proves source planning behavior, not deployed pickup or owner acceptance.
+
+
+## Current eligibility guard (review refinement)
+
+C-25 alone also reports owner replies on tasks that still require a human.
+The first 13-test preparation did not model this independent gate. Review found
+that omission before commit/activation; the refinement requires `current_tasks`
+and an explicit `observed_on` date. Its 30 offline tests pass. No automated
+pickup was dispatched by either preparation.
+
+Callers must build the mapping from fresh authoritative source, keyed by task
+record ID, and bind it to the same repository/source revision used for the scan.
+Each row explicitly includes `target_repo`, `task_record_id`, task `status`,
+`workplan_status`, `workplan_flavor`, boolean `needs_human`, `blocked_on`,
+`depends_on`, `predecessor_statuses`, `wait_until`, `decision_id` and
+`decision_status`. Resolve every task/workplan predecessor and the decision from
+the authoritative current graph; missing targets remain unknown. The offline
+module cannot authenticate a supplied source snapshot, so acquisition/binding
+is a mandatory part of the pending adapter contract. Never manufacture an empty
+predecessor list to compensate for the existing Forge dependency-edge gap.
+
+Missing eligibility fields refuse planning. Human-needed work, unresolved or
+unknown decisions, residual flavor, closed workplans, non-wait tasks, mismatched
+repo/task/owner marker, nonterminal or unknown predecessors and future windows
+produce `stopped`. Even matching successful ops-run history cannot override a
+current safety stop. Residual promotion means an authoritative flavor change;
+a comment or receipt does not promote it. Terminal predecessors/resolved
+non-human decisions and elapsed windows can leave an owner-return wait
+`reviewable`; that disposition permits review preparation only and never clears
+the wait marker, task status, human gate or any other field.
diff --git a/scripts/owner_return_handoff.py b/scripts/owner_return_handoff.py
new file mode 100644
index 0000000..1abea90
--- /dev/null
+++ b/scripts/owner_return_handoff.py
@@ -0,0 +1,149 @@
+"""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"}
+
+
+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")
+    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
+        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, "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)]
diff --git a/tests/test_historical_projection.py b/tests/test_historical_projection.py
new file mode 100644
index 0000000..7b24cdf
--- /dev/null
+++ b/tests/test_historical_projection.py
@@ -0,0 +1,135 @@
+"""Closed historical views preserve identity; unknown sources never gain authority."""
+import hashlib
+import uuid
+from pathlib import Path
+
+import pytest
+
+from api.services import forge_projection as fp
+from api.services import historical_projection as hp
+from tests.test_forge_projection import _FakeSession, _Repo, _Row, _TaskRow, TestExistingWorkplanTasks as ExistingTaskTests
+
+
+def _fixture(tmp_path, monkeypatch):
+    parent = str(uuid.uuid4()); task = str(uuid.uuid4())
+    paths = ['workplans/archived/old.md', 'workplans/split.md']
+    entries = []
+    for index, path in enumerate(paths):
+        rid = 'custodian-WP-0003' + ('a' if index else '')
+        data = (f'---\nid: {rid}\ntype: workplan\nrepo: activity-core\nstatus: done\n'
+                f'state_hub_workstream_id: {parent}\ntasks:\n'
+                f'- id: T34\n  title: Closed task\n  status: done\n  state_hub_task_id: {task}\n---\n').encode()
+        p = tmp_path/path;p.parent.mkdir(parents=True,exist_ok=True);p.write_bytes(data)
+        entries.append(dict(repo='activity-core',path=path,sha256=hashlib.sha256(data).hexdigest(),
+                            record_id=rid,uuid=parent,parent_record_id='custodian-WP-0003'))
+    monkeypatch.setattr(hp,'BINDINGS',entries)
+    return parent,task,paths
+
+
+def test_exact_shared_views_derive_one_preserved_parent(tmp_path, monkeypatch):
+    parent,task,_ = _fixture(tmp_path,monkeypatch)
+    derived = fp.derive_from_checkout(tmp_path,'activity-core','abc')
+    assert len(derived.workplans)==1
+    w = derived.workplans[0]
+    assert w.uuid == parent and w.historical_preservation
+    assert len(w.tasks)==1 and w.tasks[0].uuid==task
+    assert w.tasks[0].record_id=='T34'
+    assert w.uuid != fp.derived_record_uuid(w.record_id)
+
+
+@pytest.mark.parametrize('change',['bytes','missing','symlink'])
+def test_changed_bound_history_refuses_derive(tmp_path,monkeypatch,change):
+    _,_,paths = _fixture(tmp_path,monkeypatch)
+    p=tmp_path/paths[1]
+    if change=='bytes':p.write_text(p.read_text().replace('status: done','status: active'))
+    elif change=='missing':p.unlink()
+    else:
+        data=p.read_bytes();p.unlink();other=tmp_path/'other.md';other.write_bytes(data);p.symlink_to(other)
+    with pytest.raises(fp.ForgeDeriveError,match='Historical source binding'):
+        fp.derive_from_checkout(tmp_path,'activity-core','abc')
+
+
+def test_unknown_repo_cannot_trust_existing_binding(tmp_path,monkeypatch):
+    parent,_,_ = _fixture(tmp_path,monkeypatch)
+    derived=fp.derive_from_checkout(tmp_path,'other-repo','abc')
+    assert all(not w.historical_preservation and w.uuid != parent for w in derived.workplans)
+
+
+@pytest.mark.asyncio
+async def test_historical_receiving_preserves_terminal_tombstone_and_null_task_label(tmp_path,monkeypatch):
+    parent,task,_=_fixture(tmp_path,monkeypatch)
+    derived=fp.derive_from_checkout(tmp_path,'activity-core','abc')
+    row=_Row(slug='old@retired-20260827',status='archived',path='workplans/archived/old.md')
+    row.id=uuid.UUID(parent)
+    from datetime import datetime,timezone
+    row.projection_retired_at=datetime.now(timezone.utc)
+    t=_TaskRow(None,status='done',workplan_id=row.id);t.id=uuid.UUID(task)
+    before_row=dict(row.__dict__);before_task=dict(t.__dict__)
+    session=_FakeSession(_Repo(),[row],task_rows=[t])
+    outcome=await fp.reset_repository_projection(session,'activity-core',derived=derived)
+    assert outcome.status=='noop' and outcome.refused==[]
+    assert row.__dict__==before_row and t.__dict__==before_task
+    assert not session.added and not session.deleted
+
+
+@pytest.mark.asyncio
+async def test_missing_historical_task_refuses_before_mutation(tmp_path,monkeypatch):
+    parent,_,_=_fixture(tmp_path,monkeypatch)
+    derived=fp.derive_from_checkout(tmp_path,'activity-core','abc')
+    row=_Row(slug='old',status='archived',path='workplans/archived/old.md');row.id=uuid.UUID(parent)
+    session=_FakeSession(_Repo(),[row],task_rows=[])
+    outcome=await fp.reset_repository_projection(session,'activity-core',derived=derived)
+    assert outcome.status=='refused'
+    assert outcome.refused[0]['reason'].startswith('historical binding')
+    assert not session.added and not session.deleted
+
+
+@pytest.mark.asyncio
+async def test_task_holder_query_includes_retired_and_stale_parents():
+    derived=ExistingTaskTests()._derived(('DEMO-WP-0001-T01','Task','todo'))
+    current=_Row(slug='demo-wp-0001',status='active',path='workplans/a.md')
+    holder=_TaskRow('DEMO-WP-0001-T01',workplan_id=uuid.uuid4())
+    holder.id=uuid.UUID(derived.workplans[0].tasks[0].uuid)
+    class Session(_FakeSession):
+        statements=[]
+        async def execute(self,statement,*args,**kwargs):
+            self.statements.append(str(statement))
+            return await super().execute(statement,*args,**kwargs)
+    session=Session(_Repo(),[current],task_rows=[],task_collision_rows=[holder])
+    outcome=await fp.reset_repository_projection(session,'demo',derived=derived)
+    assert outcome.status=='refused' and not session.added
+    query=session.statements[-1]
+    assert 'tasks.id IN' in query
+    assert 'projection_retired_at' not in query and 'NOT IN' not in query and 'JOIN' not in query
+
+
+@pytest.mark.asyncio
+async def test_missing_historical_parent_never_recreates(tmp_path,monkeypatch):
+    _fixture(tmp_path,monkeypatch)
+    derived=fp.derive_from_checkout(tmp_path,'activity-core','abc')
+    session=_FakeSession(_Repo(),[])
+    outcome=await fp.reset_repository_projection(session,'activity-core',derived=derived)
+    assert outcome.status=='refused' and not session.added and not session.deleted
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('field',['parent','task'])
+async def test_historical_nonterminal_hub_record_requires_owner_review(tmp_path,monkeypatch,field):
+    parent,task,_=_fixture(tmp_path,monkeypatch)
+    derived=fp.derive_from_checkout(tmp_path,'activity-core','abc')
+    row=_Row(slug='old',status='active' if field=='parent' else 'finished',path='workplans/archived/old.md')
+    row.id=uuid.UUID(parent)
+    t=_TaskRow('preserved-old-label',status='wait' if field=='task' else 'done',workplan_id=row.id)
+    t.id=uuid.UUID(task)
+    session=_FakeSession(_Repo(),[row],task_rows=[t])
+    outcome=await fp.reset_repository_projection(session,'activity-core',derived=derived)
+    assert outcome.status=='refused' and not session.added and not session.deleted
+
+
+def test_historical_diff_uses_uuid_without_relabeling(tmp_path,monkeypatch):
+    parent,task,_=_fixture(tmp_path,monkeypatch)
+    derived=fp.derive_from_checkout(tmp_path,'activity-core','abc')
+    diff=fp.diff_against_hub(derived,[{'id':parent,'slug':'historic@retired-20260827',
+                                    'status':'archived','backing_relative_path':'old-path.md'}],
+        {parent:[{'id':task,'record_id':None,'title':'Old title','status':'done'}]})
+    assert not diff.missing and not diff.stale and not diff.differing
diff --git a/tests/test_owner_return_handoff.py b/tests/test_owner_return_handoff.py
new file mode 100644
index 0000000..eb8e4c4
--- /dev/null
+++ b/tests/test_owner_return_handoff.py
@@ -0,0 +1,197 @@
+import copy
+from datetime import date
+import importlib.util
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+SOURCE = ROOT / "scripts" / "owner_return_handoff.py"
+if not SOURCE.exists():
+    SOURCE = Path(__file__).with_name("owner_return_handoff.py")
+spec = importlib.util.spec_from_file_location("owner_return_handoff", SOURCE)
+module = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(module)
+raw_plan = module.plan_owner_returns
+
+def eligibility(source):
+    return {issue["owner_return"]["task_record_id"]: {
+        "target_repo": source["repo_slug"],
+        "task_record_id": issue["owner_return"]["task_record_id"],
+        "status": "wait", "workplan_status": "active", "workplan_flavor": "implementation",
+        "needs_human": False, "blocked_on": "message-from:ops-warden",
+        "depends_on": [], "predecessor_statuses": {}, "wait_until": None,
+        "decision_id": None, "decision_status": None,
+    } for issue in source["issues"] if "owner_return" in issue}
+
+def plan(source, **kwargs):
+    kwargs.setdefault("current_tasks", eligibility(source))
+    kwargs.setdefault("observed_on", date(2026, 10, 4))
+    return raw_plan(source, **kwargs)
+DEF = "ac000000-0000-0000-0000-000000000001"
+MSG = "ab000000-0000-0000-0000-000000000001"
+RUN = "ad000000-0000-0000-0000-000000000001"
+
+
+def report():
+    return {"repo_slug": "state-hub", "result": "warn", "issues": [
+        {"check_id": "C-25", "file_value": "message-from:ops-warden", "owner_return": {
+            "task_record_id": "STATE-WP-0094-T03", "message_ids": [MSG]}}
+    ]}
+
+
+def receipt(candidate, state):
+    return {"id": RUN, "activity_definition_id": DEF,
+            "target_repo": "state-hub", "state": state,
+            "triggering_event_id": candidate["triggering_event_id"]}
+
+
+def test_repeat_and_issue_order_are_stable_without_mutation():
+    source = report()
+    before = copy.deepcopy(source)
+    source["issues"] *= 2
+    source["issues"][0]["owner_return"]["message_ids"] *= 2
+    result = plan(source, activity_definition_id=DEF)
+    assert len(result) == 1
+    assert result == plan(before, activity_definition_id=DEF)
+    assert result[0]["disposition"] == "reviewable"
+    assert source["issues"][0]["owner_return"]["message_ids"] == [MSG, MSG]
+
+
+def test_each_repo_task_message_has_distinct_identity():
+    source = report()
+    a = plan(source, activity_definition_id=DEF)[0]["triggering_event_id"]
+    for field, value in (("repo_slug", "ops-warden"),):
+        changed = copy.deepcopy(source)
+        changed[field] = value
+        assert plan(changed, activity_definition_id=DEF)[0]["triggering_event_id"] != a
+    changed = report()
+    changed["issues"][0]["owner_return"]["task_record_id"] = "STATE-WP-0094-T05"
+    assert plan(changed, activity_definition_id=DEF)[0]["triggering_event_id"] != a
+    changed["issues"][0]["owner_return"]["message_ids"] = [RUN]
+    assert plan(changed, activity_definition_id=DEF)[0]["triggering_event_id"] != a
+
+
+@pytest.mark.parametrize("state,disposition", [
+    ("open", "pending"), ("claimed", "pending"), ("succeeded", "completed"),
+    ("failed", "review_required"), ("expired", "review_required")])
+def test_existing_supported_receipts_prevent_repeat_ready(state, disposition):
+    candidate = plan(report(), activity_definition_id=DEF)[0]
+    row = receipt(candidate, state)
+    actual = plan(report(), activity_definition_id=DEF, ops_runs=[row])[0]
+    assert actual["disposition"] == disposition
+    assert actual["ops_run_id"] == RUN
+
+
+@pytest.mark.parametrize("field,value", [
+    ("target_repo", "ops-warden"), ("activity_definition_id", RUN),
+    ("triggering_event_id", "another-fire")])
+def test_unrelated_receipts_do_not_suppress(field, value):
+    candidate = plan(report(), activity_definition_id=DEF)[0]
+    row = receipt(candidate, "succeeded")
+    row[field] = value
+    assert plan(report(), activity_definition_id=DEF, ops_runs=[row])[0]["disposition"] == "reviewable"
+
+
+def test_duplicate_correlated_receipts_refuse_ambiguous_replay():
+    candidate = plan(report(), activity_definition_id=DEF)[0]
+    row = receipt(candidate, "succeeded")
+    with pytest.raises(ValueError, match="multiple correlated"):
+        plan(report(), activity_definition_id=DEF, ops_runs=[row, row])
+
+
+def test_missing_id_and_collection_error_fail_closed():
+    source = report()
+    source["issues"][0]["owner_return"]["message_ids"] = ["unknown"]
+    with pytest.raises(ValueError):
+        plan(source, activity_definition_id=DEF)
+    source = report()
+    source["result"] = "error"
+    with pytest.raises(ValueError, match="collection"):
+        plan(source, activity_definition_id=DEF)
+
+
+def test_legacy_workplan_warning_and_other_findings_are_not_tasks():
+    assert plan({"repo_slug": "state-hub", "issues": [
+        {"check_id": "C-25"}, {"check_id": "C-41"}]}, activity_definition_id=DEF) == []
+
+
+@pytest.mark.parametrize("field,value,reason", [
+    ("needs_human", True, "human_gate"),
+    ("workplan_flavor", "residual", "unpromoted_residual"),
+    ("task_record_id", "STATE-WP-0094-T05", "stale_task_identity"),
+    ("target_repo", "ops-warden", "stale_task_identity"),
+    ("blocked_on", "message-from:railiance-platform", "stale_owner_return_marker"),
+    ("status", "done", "task_not_waiting"),
+    ("workplan_status", "finished", "workplan_not_open"),
+    ("wait_until", "2026-10-05", "future_wait_window"),
+])
+def test_current_gates_stop_even_when_matching_run_succeeded(field, value, reason):
+    source = report()
+    tasks = eligibility(source)
+    tasks["STATE-WP-0094-T03"][field] = value
+    row = receipt(plan(source, activity_definition_id=DEF)[0], "succeeded")
+    actual = plan(source, activity_definition_id=DEF, current_tasks=tasks, ops_runs=[row])[0]
+    assert actual["disposition"] == "stopped"
+    assert actual["stop_reason"] == reason
+    assert actual["ops_run_id"] == RUN
+
+
+def test_missing_current_eligibility_refuses_not_ready():
+    with pytest.raises(ValueError, match="missing authoritative"):
+        plan(report(), activity_definition_id=DEF, current_tasks={})
+    tasks = eligibility(report())
+    del tasks["STATE-WP-0094-T03"]["needs_human"]
+    with pytest.raises(ValueError, match="missing authoritative"):
+        plan(report(), activity_definition_id=DEF, current_tasks=tasks)
+
+
+@pytest.mark.parametrize("status", [None, "wait", "progress"])
+def test_unresolved_or_unknown_predecessor_stops(status):
+    tasks = eligibility(report())
+    row = tasks["STATE-WP-0094-T03"]
+    row["depends_on"] = ["WARDEN-WP-0041-T03"]
+    row["predecessor_statuses"] = {"WARDEN-WP-0041-T03": status}
+    assert plan(report(), activity_definition_id=DEF, current_tasks=tasks)[0]["disposition"] == "stopped"
+
+
+def test_terminal_predecessors_and_expired_window_preserve_owner_review():
+    tasks = eligibility(report())
+    row = tasks["STATE-WP-0094-T03"]
+    row["depends_on"] = ["WARDEN-WP-0041-T03"]
+    row["predecessor_statuses"] = {"WARDEN-WP-0041-T03": "done"}
+    row["wait_until"] = "2026-10-04"
+    result = plan(report(), activity_definition_id=DEF, current_tasks=tasks)
+    assert result[0]["disposition"] == "reviewable"
+    assert row["status"] == "wait"
+    assert row["blocked_on"] == "message-from:ops-warden"
+
+
+def test_quoted_false_or_malformed_window_not_authority():
+    tasks = eligibility(report())
+    tasks["STATE-WP-0094-T03"]["needs_human"] = "false"
+    with pytest.raises(ValueError):
+        plan(report(), activity_definition_id=DEF, current_tasks=tasks)
+    tasks = eligibility(report())
+    tasks["STATE-WP-0094-T03"]["wait_until"] = "2026-10-04T00:00:00Z"
+    with pytest.raises(ValueError):
+        plan(report(), activity_definition_id=DEF, current_tasks=tasks)
+
+
+@pytest.mark.parametrize("status", [None, "open"])
+def test_unresolved_decision_stops_without_human_flag(status):
+    tasks = eligibility(report())
+    row = tasks["STATE-WP-0094-T03"]
+    row["decision_id"] = MSG
+    row["decision_status"] = status
+    result = plan(report(), activity_definition_id=DEF, current_tasks=tasks)
+    assert result[0]["stop_reason"] == "decision_not_resolved_or_unknown"
+
+
+def test_resolved_decision_does_not_override_human_gate():
+    tasks = eligibility(report())
+    row = tasks["STATE-WP-0094-T03"]
+    row.update(decision_id=MSG, decision_status="resolved", needs_human=True)
+    assert plan(report(), activity_definition_id=DEF, current_tasks=tasks)[0]["stop_reason"] == "human_gate"
+    row["needs_human"] = False
+    assert plan(report(), activity_definition_id=DEF, current_tasks=tasks)[0]["disposition"] == "reviewable"
diff --git a/workplans/STATE-WP-0094-routed-gates-and-owner-returns.md b/workplans/STATE-WP-0094-routed-gates-and-owner-returns.md
index c99bba3..d828e60 100644
--- a/workplans/STATE-WP-0094-routed-gates-and-owner-returns.md
+++ b/workplans/STATE-WP-0094-routed-gates-and-owner-returns.md
@@ -48,7 +48,7 @@ it does not awaken agents or create duplicate tasks on every checker run.
 
 ```task
 id: STATE-WP-0094-T03
-status: todo
+status: progress
 priority: medium
 state_hub_task_id: "cd73e265-68b0-5dd5-9874-d1d439910a7e"
 ```
@@ -178,3 +178,25 @@ T05 remains progress for actual task-level dependency-edge receiving and
 scheduled freshness coverage; T03 retains runner selection/dedup/receipts.
 Neither the legacy remote scheduler nor a recurring provisioning cadence was
 activated. Evidence: docs/evidence/2026-10-04-receiving-rollout.json.
+
+
+2026-10-04 T03 source preparation: selected ActivityDefinition → Activity Core
+ops_run → admitted owner harness as the supported recurring pickup boundary.
+The offline owner_return_handoff planner derives stable repo/task/message event
+IDs and correlates existing ops-run pending/terminal receipts; 13 staged tests
+pass. No dispatch, receipt write, schedule or task transition is performed.
+An accepted definition/profile/authority and callable owner submission adapter
+remain required, so T03 is progress, not done. Details and T05 graph freshness
+assessment: docs/owner-return-handoff.md. Normal Forge receiving still omits
+dependency edges; C-20 can index them, but runtime freshness coverage and safe
+source-owned edge reconciliation remain live T05 obligations.
+
+
+2026-10-04 T03 eligibility review refinement: C-25 can report tasks carrying
+both owner-return and human gates. Before commit/activation the planner was
+corrected to require current authoritative task/workplan eligibility, resolved
+predecessor/decision states and observation date. Human, residual, stale,
+nonterminal or future gates stop preparation even with prior successful pickup.
+Missing safety metadata refuses planning. Thirty staged tests pass; the initial
+13-test result remains historical preparation evidence, not complete safety
+coverage. T03/T05 stay progress; no dispatch, schedule or source gate clearing.
diff --git a/workplans/STATE-WP-0095-historical-source-receiving.md b/workplans/STATE-WP-0095-historical-source-receiving.md
new file mode 100644
index 0000000..6a79c22
--- /dev/null
+++ b/workplans/STATE-WP-0095-historical-source-receiving.md
@@ -0,0 +1,95 @@
+---
+id: STATE-WP-0095
+type: workplan
+title: "Preserve exact closed historical source bindings in Forge receiving"
+domain: infotech
+repo: state-hub
+status: active
+owner: codex
+created: "2026-10-04"
+updated: "2026-10-04"
+flavor: implementation
+depends_on: []
+---
+
+## Goal
+
+Normal Forge receiving must preserve the seven bound Activity legacy source
+views and the Custodian retired ADHOC source without new identities, source
+rewrites, task movement, deletion, relabeling or unretirement. This implements
+receiving support before RMGR-WP-0019 registrar admission. CUST-WP-0077 owns
+cross-owner coordination; deployment and replay remain explicit owner returns.
+
+## Verify source bindings and shared views
+
+```task
+id: STATE-WP-0095-T01
+status: done
+priority: high
+```
+
+Reviewed seven exact Activity files: four existing workplans and 64 UUID-bound
+closed tasks. The archived custodian-WP-0003 parent and three split_from views
+share one existing parent UUID; every repeated task binding/content agrees.
+Custodian ADHOC-2026-08-25-retired retains parent
+1c53d3db-6633-5e2d-987f-10706d73be9e and task
+d8af3e6b-26ae-580c-bbeb-caa8ad419bec. No identity migration is authorized.
+
+## Implement preservation receiving and global collision refusal
+
+```task
+id: STATE-WP-0095-T02
+status: done
+priority: high
+```
+
+Prepared exact source digest manifest, existing-parent/task verification,
+immutable receiving and UUID-based historical diff. Ordinary canonical sources
+continue deriving their identities. Missing/changed/symlinked source, absent
+bound rows or nonterminal hub histories refuse. Check task UUID holders globally,
+including retired and acknowledged-stale parent workplans, before INSERT.
+
+Offline preview tests and snapshot proof are evidence of source behavior only;
+finish after applying reviewed source and its focused tests to the owning repo.
+
+## Deliver receiving through normal deployment and verify owner return
+
+```task
+id: STATE-WP-0095-T03
+status: wait
+priority: high
+depends_on: [STATE-WP-0095-T02]
+blocked_on: "state-hub owner must deliver the reviewed receiving commit through the normal deployment path and return deployment commit/version evidence before registrar admission"
+```
+
+No direct live code patch. This task is a deployment return, not policy authority
+or permission to resolve identity decisions. Exact receiving fixture acceptance
+and global collision rejection must hold in the delivered version.
+
+## Verify normal historical replay preserves existing identities
+
+```task
+id: STATE-WP-0095-T04
+status: wait
+priority: high
+depends_on: [STATE-WP-0095-T03, RMGR-WP-0019-T03]
+blocked_on: "repo-manager owner RMGR-WP-0019-T02/T03 must return normal historical registrar/replay receipts and before/after preserved UUID verification"
+```
+
+Verify Activity four historical parent UUIDs/64 task UUIDs and Custodian retired
+parent/task unchanged; no historical row creation, movement, deletion,
+relabeling or unretirement. Unrelated stale projections remain separate decisions.
+
+## Offline evidence, 2026-10-04
+
+Prepared receiver suite: Forge projection regression plus historical preservation,
+missing/changed source, terminality, immutable UUID diff, and global holder query.
+Real-source offline snapshot proof preserves four Activity parents/64 tasks and
+one Custodian parent/task without mutations; no API writes or deployment occurred.
+Exact original bindings are maintained in api/services/historical_projection.py.
+
+2026-10-04 actual owning-source validation: 155 focused receiver/qualifier/planner
+cases passed; the independent offline current-source/current-hub-snapshot proof
+retained five existing parents and 65 task UUIDs, no mutations, clean historical
+diff. T02 is source-complete; T03/T04 remain wait for reviewed deployment and
+normal registrar/replay receiving, respectively. Evidence: docs/evidence/2026-10-04-historical-receiving-source.json.