Add verified task-project transfer and write-home handoff

docs/design_index.md

@@ -127,6 +127,7 @@ from `AGENTS.md`.
 | [Agent memory and enduring knowledge](engineering/agent_memory_and_enduring_knowledge.md) | Active design guide | Normative only within its stated memory/retrieval scope |
 | [Role-learning convergence](engineering/role_learning_convergence.md) | Active programme design and guide | Canonical programme direction under `AGENTS.md` for claims about enduring role competence; no runtime activation, universal per-PR gate or demonstrated role capability is implied |
 | [Knowledge federation pilot](engineering/knowledge_federation_pilot.md) | Bounded implementation and authority contract | JVNAUTOSCI-2730/2731: automatic scoped discovery, conversation recall and on-demand private artefact copies; opt-in complete-slice exchange between independent stores; source audiences, atomic receipt/read projection, expiry, recovery and later SC448086 admission. Not general database replication or execution federation. |
+| [Task-project home transfer](engineering/task_project_home_transfer.md) | Bounded task-project operations and recovery | Signed project snapshots, preserved audiences and archives, explicit Atlas-to-DGX write-home handoff, current-actor dispatch and interruption recovery. |
 | [Contextual knowledge evolution](engineering/contextual_knowledge_evolution.md) | Active design guide | Advisory under `AGENTS.md` when assertion/context distinctions are material; it does not select a microtheory formalism or prove present implementation |
 | [Assertion and propositional-sentence ontology](engineering/assertion_and_propositional_sentence_ontology.md) | Active design and partial implementation boundary | Canonical model and phased integration plan; the minimal public vocabulary is live, while assertion-record typing, context semantics, general logical storage, and reasoning remain future work |
 | [External conversation import](engineering/external_conversation_import.md) | Active design and implementation boundary | Canonical reference for source-neutral transcript packages, actor-scoped read-only projections, provider adapters, append-only resynchronisation, durable machine-wide batches, and native continuation; verify supported provider export shapes against current fixtures and live evidence |

docs/engineering/task_project_home_transfer.md

@@ -0,0 +1,95 @@
+# Task-project home transfer
+
+This bounded operational slice of the federation programme (JVNAUTOSCI-2615)
+keeps MongoDB and transfers selected native-writer projects. It does not replace
+the broader federation plan or implement general database replication.
+
+## Contract and authority
+
+`writer=von` chooses the tracking application. `task_project_homes` separately
+records the physical write/execution home, epoch, route and state. Private peer
+configuration admits exact projects and audiences; verified identity mappings
+are independent of subscriptions. Task IDs, original authors, Jira aliases,
+comments/history/relationships, project membership and source file bytes survive.
+Accounts, memberships, login bindings, dispatch receipts and execution state do
+not travel with a snapshot. Current canonical assignment creates local dispatch
+authority; a copied assignment cannot start a coding agent.
+
+Canonical task, concept and text-relation mutations obey the project barrier.
+The durable worker checks the home before claiming and again before actions.
+The claim-admission validator can require `task_project_home_v1` to exclude older
+workers. This does not constrain arbitrary administrator database writes.
+Before first registration, identify and drain all previous writers and in-flight
+claims; update active consumers and fence legacy worker claims. An in-flight
+external action is not cancelled merely by changing a home record.
+
+## Transfer order
+
+Use `scripts/transfer_task_project.py` under the explicitly selected environment
+with a private federation config, `VON_TASK_HOME_NODE_ID`, actor and original
+organisation scope. The script is an operator CLI, not a public authority API.
+Each output is owner-only and should be retained with the task's audit evidence.
+
+1. Inventory the exact projects, audiences, identities, collision histories,
+   active execution and source/destination recovery evidence. Verify old records
+   by Jira source identity and history retention, not concept slug or date alone.
+2. Deploy home-aware canonical services and consumers. Require the worker
+   capability at the shared claim boundary after capable workers are available.
+3. `register` the source home with the evidence file. `freeze` the exact epoch
+   only after writers/claims have drained and every retained operation has been
+   reconciled. Failed writes retain operation tokens; elapsed time is not proof
+   their effects stopped.
+4. `capture` a fresh signed snapshot while frozen. An active/preview snapshot
+   cannot be published. Transfer the snapshot and private hash-named file cache
+   through an authenticated operator channel. Keep credentials on their hosts.
+5. At the destination, `preflight` verifies identities/history and records exact
+   current concept and text digests. `stage-files` verifies all file hashes and
+   sizes before and after destination storage. A missing/corrupt file stops the
+   transfer without publishing concepts.
+6. `publish` rechecks the preflight inside one MongoDB transaction, retaining the
+   prior complete view if interrupted. Transactions require a replica set; a
+   standalone server is rejected. The complete snapshot and receipt become
+   visible together with a **prepared**, non-writable, non-executing home.
+7. Canonically read back project/task text, relationships, membership, aliases,
+   audiences and file bytes. Obtain the signed destination `receipt`.
+8. At the source, `release` compares a fresh frozen capture with that receipt,
+   moves the route to a read-only replica and returns a signed release.
+9. At the destination, `activate` requires the exact signed source release,
+   publication receipt, digest and epoch. Only now can canonical writes and
+   deliberate assignment proceed. Verify public search, native create/edit and
+   an explicitly assigned harmless coding-agent canary through actual result.
+10. Repeat for the remaining admitted projects in their own scopes. Report
+    per-project completion and outstanding exceptions, not only pilot success.
+
+## Interruption and recovery
+
+Before publication, retry capture/staging after inspecting the retained state.
+An interrupted transaction leaves the previous destination view intact. A
+committed publication can be replayed idempotently; its receipt is returned
+without overwriting later edits. Preflight rejects concurrent destination text
+or concept changes and requires a new reconciliation.
+
+Between publication and release, both homes remain non-writable. Retry the
+exact receipt after diagnosing the failure. Between release and activation,
+the source remains a read-only replica; retry the exact signed release. There
+is no timeout-based ownership takeover. Stale epochs cannot reactivate a prior
+home. After activation, recover the current destination in place, retaining
+all subsequent writes. Returning ownership is a new explicitly reconciled
+transfer of the latest state; restoring the old source snapshot is prohibited.
+The current CLI does not automate a reverse transfer into an existing home.
+
+Mongo recovery must be tested in a separate authenticated database. A live dump
+is not a whole-database point-in-time snapshot. Preserve the existing data
+volume and take a cold copy before changing a standalone server's topology.
+A topology rollback must retain the latest data volume, not replace it with a
+pre-transfer backup after new work has been accepted. Keep blob bytes with the
+paired database recovery evidence.
+
+## Validation boundary
+
+Targeted tests cover audience/signature denial, retained uncertain operations,
+write/claim fencing, actor-bound dispatch, signed handoff, stale-epoch rejection,
+and real-Mongo transaction interruption/retry with later-write preservation.
+Real project rehearsal and public/canonical read-back are required for a live
+transfer claim. Fixture home certificates are labelled fixture-only and never
+establish that a production source was released.

scripts/codex_von_worker.py

@@ -187,15 +187,29 @@ def require_task_execution_preferences(task):
 
 
 def authorised_task(task, config):
-    return (
-        task.get("assignee_concept_id") == config["agent_id"]
+    if (
+        task.get("assignee_concept_id") != config["agent_id"]
+        or task.get("organisation_concept_id") not in (None, config["organisation_id"])
+    ):
+        return False
+    if not config.get("supervision_enabled") and task.get(
+        "report_to_concept_id"
+    ) not in (None, config["delegator_id"]):
+        return False
+    from src.backend.services.task_dispatch_authority_service import (
+        has_assignment_record,
+        is_authorised_assignment,
+    )
+    if (
+        not task.get("federation")
         and task.get("organisation_concept_id") == config["organisation_id"]
         and task.get("created_by_concept_id") == config["delegator_id"]
-        and (
-            config.get("supervision_enabled")
-            or task.get("report_to_concept_id") in (None, config["delegator_id"])
-        )
-    )
+        and not has_assignment_record(task["task_concept_id"])
+    ):
+        return True
+    if not task.get("dispatch_requested"):
+        return False
+    return is_authorised_assignment(task, config["delegator_id"])
 
 
 def referenced_tasks(config, api, supplied, task_ids=()):
@@ -435,8 +449,8 @@ def check_task(config, api, task_id):
     while True:
         page = api.tasks.search_tasks(
             assignee_concept_id=config["agent_id"],
-            created_by_concept_id=config["delegator_id"],
             organisation_concept_id=config["organisation_id"],
+            organisation_scope_mode="current_plus_unscoped",
             statuses=[task["status"]],
             limit=200,
             offset=offset,
@@ -504,19 +518,27 @@ def task(self, task_id):
     def native_writer(self, task):
         project_id = task.get("project_concept_id")
         if project_id:
-            from src.backend.services.task_project_service import get_task_project
-
-            project = get_task_project(project_id)
-            return bool(project and project.get("writer") == "von")
+            from src.backend.services.task_project_service import task_execution_home
+
+            project = task_execution_home(task["task_concept_id"])
+            home = project.get("write_home") if project else None
+            return bool(
+                project
+                and project.get("writer") == "von"
+                and (
+                    home is None
+                    or home.get("writable_here") is True
+                )
+            )
         return not task.get("is_imported_jira_task", False)
 
     def pending(self):
         offset = 0
         while True:
             page = self.tasks.search_tasks(
                 assignee_concept_id=self.config["agent_id"],
-                created_by_concept_id=self.config["delegator_id"],
                 organisation_concept_id=self.config["organisation_id"],
+                organisation_scope_mode="current_plus_unscoped",
                 statuses=sorted(ACTIVE),
                 limit=200,
                 offset=offset,

scripts/durable_claim_admission.py

@@ -16,10 +16,17 @@
 
 
 def install_admission(
-    db, *, collection="workflow_instances", worker_pattern="^worker_", apply=False
+    db,
+    *,
+    collection="workflow_instances",
+    worker_pattern="^worker_",
+    apply=False,
+    required_capability="durable_task_ownership_v1",
 ):
     options = db[collection].options()
-    rule = claim_admission_validator(worker_pattern)
+    rule = claim_admission_validator(
+        worker_pattern, required_capability=required_capability
+    )
     existing = options.get("validator", {})
     validator = (
         existing
@@ -55,10 +62,14 @@ def install_admission(
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("--apply", action="store_true")
     parser.add_argument("--worker-pattern", default="^worker_")
+    parser.add_argument("--required-capability", default="durable_task_ownership_v1")
     parser.add_argument("--receipt", type=Path, required=True)
     args = parser.parse_args()
     result = install_admission(
-        get_db(), worker_pattern=args.worker_pattern, apply=args.apply
+        get_db(),
+        worker_pattern=args.worker_pattern,
+        apply=args.apply,
+        required_capability=args.required_capability,
     )
     args.receipt.write_text(json.dumps(result, indent=2, default=str))
     print(json.dumps(result, default=str))

scripts/restore_von_db.py

@@ -262,6 +262,28 @@ def _resolve_blob_store_for_restore(backend: str):
             os.environ["VON_BLOB_STORE_BACKEND"] = orig
 
 
+def _uri_for_namespace_restore(uri: str) -> str:
+    """Let --nsFrom/--nsTo select databases while preserving authentication.
+
+    A database path in the URI makes mongorestore treat --dir as a single
+    database directory and skip its database subdirectories with exit code 0.
+    """
+    from urllib.parse import parse_qsl, unquote, urlencode, urlsplit, urlunsplit
+
+    parts = urlsplit(uri)
+    database = unquote(parts.path.lstrip("/"))
+    query = parse_qsl(parts.query, keep_blank_values=True)
+    if (
+        database
+        and parts.username
+        and not any(k.lower() == "authsource" for k, _ in query)
+    ):
+        query.append(("authSource", database))
+    return urlunsplit(
+        (parts.scheme, parts.netloc, "/", urlencode(query), parts.fragment)
+    )
+
+
 def _run_mongorestore(
     *,
     mongo_uri: str,
@@ -302,7 +324,7 @@ def link_or_copy(source, destination):
                 )
         config_path = config_root / "config.yml"
         config_path.write_text(
-            f"uri: {json.dumps(mongo_uri)}\n",
+            f"uri: {json.dumps(_uri_for_namespace_restore(mongo_uri))}\n",
             encoding="utf-8",
         )
         os.chmod(config_path, 0o600)