diff --git a/docs/engineering/codex_dgx_worker.md b/docs/engineering/codex_dgx_worker.md
index 6ca2ce5d8..ff313753a 100644
--- a/docs/engineering/codex_dgx_worker.md
+++ b/docs/engineering/codex_dgx_worker.md
@@ -3,7 +3,7 @@
 - **Kind:** Operator runbook and bounded capability description
 - **Lifecycle:** Pilot
 - **Owner:** Michael Witbrock / SAIL
-- **Last reviewed:** 11 September 2026
+- **Last reviewed:** 12 September 2026
 - **Decision surface:** [JVNAUTOSCI-2740](https://naoinstitute.atlassian.net/browse/JVNAUTOSCI-2740)
 
 ## User workflow
@@ -154,6 +154,41 @@ Validate an actual agent-issued shell command before enabling polling.
 
 Install the worker, `codex_von_inbox.py`, and both adjacent `_prompt.md` files
 from the same reviewed revision. Run them with the project's PDM-managed Python environment.
+When these scripts are copied into a release bundle outside a complete checkout,
+pass `--backend-root /path/to/reviewed/Von` to the worker. This selects the
+canonical backend before any backend imports; a missing backend or previously
+loaded imports from another checkout fail with a restart instruction. A checkout
+installation defaults to the checkout containing the worker script.
+
+The selected backend must include task execution-preference hydration on both
+point and list reads. The adapter requires `requested_model` and
+`requested_reasoning_effort` keys even when their values are null. An older reader
+that omits them fails visibly before pickup instead of silently choosing defaults.
+Installing only a newer worker script cannot repair an older canonical reader.
+Do not change global model defaults to compensate for a mismatched installation.
+
+For an authorised upgrade, install the reviewed backend and worker bundle as one
+compatible release, retain the previous backend/bundle selection for rollback,
+and update the scheduled wrapper to pass the chosen backend explicitly. Preserve
+the existing worker lock, state directory, identities, memberships and task
+selectors. Let active work finish before switching the scheduled process; do not
+launch a second controller to test pickup. Before re-enabling the schedule, check
+a labelled isolated task through that adapter's point/list reads, `inputs` and
+`resolve_execution_settings`, including exact task-sourced model/reasoning values.
+This installation change is separate from repository publication and public web
+deployment and requires its own activation authority.
+
+Ordinary conversation `task_create` permits explicit assignment to a represented
+same-organisation coding agent using the continuation route's live membership
+checks. It verifies creator, organisation, assignee, report recipient, execution
+preferences and source link on canonical read-back. A standalone native task does
+not need a project or collection for worker eligibility. Eligibility and creation
+do not prove actual pickup. Unsupported `parent_task_concept_id` on creation is
+rejected before writing; the separately authorised `task_create_subtask` and
+`task_set_parent` routes remain the hierarchy surfaces. When an update recovers
+only some requested fields, the outcome report retains the task link and verified
+fields alongside the unresolved ones.
+
 A JSON config has these fields (paths are operator-selected):
 
 ```json
diff --git a/scripts/codex_von_worker.py b/scripts/codex_von_worker.py
index adcc2f2af..236c7b40a 100755
--- a/scripts/codex_von_worker.py
+++ b/scripts/codex_von_worker.py
@@ -80,6 +80,40 @@ def resolve_execution_settings(config, inputs):
     return resolved
 
 
+def bind_backend_root(backend_root=None):
+    """Select one reviewed backend before imports, including copied script bundles."""
+    root = Path(backend_root or Path(__file__).resolve().parents[1]).resolve()
+    if not (root / "src/backend/services/task_management_service.py").is_file():
+        raise RuntimeError(
+            "Worker backend is missing; pass --backend-root with the "
+            "reviewed Von checkout containing canonical task services"
+        )
+    for name, module in tuple(sys.modules.items()):
+        if name == "src" or name.startswith("src."):
+            filename = getattr(module, "__file__", None)
+            if filename and not Path(filename).resolve().is_relative_to(root):
+                raise RuntimeError(
+                    "Worker backend imports already belong to another "
+                    "checkout; restart with the selected --backend-root"
+                )
+    sys.path.insert(0, str(root))
+    return root
+
+
+def require_task_execution_preferences(task):
+    # Current canonical point and list reads include both keys even when unset.
+    # An older backend omits them. Never turn an incompatible read into defaults.
+    missing = {"requested_model", "requested_reasoning_effort"} - task.keys()
+    if missing:
+        raise RuntimeError(
+            "Canonical task reader lacks execution-preference fields: "
+            + ", ".join(sorted(missing))
+            + ". Install the matching backend and worker release; "
+            "verify --backend-root before resuming pickup."
+        )
+    return task
+
+
 def authorised_task(task, config):
     return (
         task.get("assignee_concept_id") == config["agent_id"]
@@ -118,7 +152,7 @@ def __init__(self, config):
         self.messages = message_service
 
     def task(self, task_id):
-        return self.tasks.get_task(task_id)
+        return require_task_execution_preferences(self.tasks.get_task(task_id))
 
     def native_writer(self, task):
         project_id = task.get("project_concept_id")
@@ -144,7 +178,7 @@ def pending(self):
                 if not authorised_task(task, self.config):
                     continue
                 if self.native_writer(task):
-                    yield task
+                    yield require_task_execution_preferences(task)
                 else:
                     self.send(
                         task,
@@ -161,6 +195,7 @@ def pending(self):
 
     def inputs(self, task):
         """Task comments and direct replies retain their actual authorship."""
+        require_task_execution_preferences(task)
         task_id = task["task_concept_id"]
         comments = []
         offset = 0
@@ -750,6 +785,11 @@ def tick(config, api, lock_fd):
 def main():
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("--config", required=True)
+    parser.add_argument(
+        "--backend-root",
+        help="Reviewed canonical backend checkout; "
+        "required when running a copied worker script bundle",
+    )
     options = parser.parse_args()
     os.umask(0o077)
     config = json.loads(Path(options.config).read_text())
@@ -762,7 +802,7 @@ def main():
         except BlockingIOError:
             print('{"outcome":"already_running"}')
             return
-        sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+        bind_backend_root(options.backend_root)
         from src.backend.integrations.internal_mcp.gateway import (
             INTERNAL_MCP_TRUSTED_LOCAL_OPERATOR_SOURCE,
             bind_internal_mcp_actor_context_source,
diff --git a/src/backend/integrations/internal_mcp/catalogue.py b/src/backend/integrations/internal_mcp/catalogue.py
index 4c6d14ce0..cebd9b948 100644
--- a/src/backend/integrations/internal_mcp/catalogue.py
+++ b/src/backend/integrations/internal_mcp/catalogue.py
@@ -35189,6 +35189,7 @@ def _task_create(**kwargs):
         create_task,
         find_task_by_agent_creation_fingerprint,
         get_task,
+        _task_concept_id_for_agent_creation_fingerprint,
     )
 
     title = kwargs.get("title")
@@ -35214,7 +35215,31 @@ def _task_create(**kwargs):
         return scope_error
     assert actor_scope is not None
 
-    assignee_id = kwargs.get("assignee_id") or kwargs.get("assignee_concept_id")
+    if "parent_task_concept_id" in kwargs:
+        return {
+            **make_error_response(
+                "task_create_parent_unsupported",
+                "task_create does not support parent_task_concept_id. No task was "
+                "created. Omit it for an independent related task; a required child "
+                "relationship needs the separately authorised task_create_subtask "
+                "or task_set_parent capability.",
+            ),
+            "changed": False,
+        }
+    for aliases in (
+        ("assignee_id", "assignee_concept_id"),
+        ("report_to_concept_id", "reports_to_concept_id"),
+    ):
+        values = {str(kwargs[key]).strip() for key in aliases if kwargs.get(key)}
+        if len(values) > 1:
+            return make_error_response(
+                "INVALID_PARAM", f"Conflicting aliases: {', '.join(aliases)}"
+            )
+    assignee_id = str(
+        kwargs.get("assignee_id")
+        or kwargs.get("assignee_concept_id")
+        or actor_scope.user_concept_id
+    ).strip()
     session_id = kwargs.get("originating_session_id") or kwargs.get("session_id")
     created_by = kwargs.get("created_by_concept_id") or kwargs.get("namespace")
     priority = kwargs.get("priority", "medium")
@@ -35246,17 +35271,28 @@ def _task_create(**kwargs):
     actor_id = str(actor_scope.user_concept_id)
     actor_org_id = str(actor_scope.organisation_concept_id or "")
     if (
-        str(assignee_id or "").strip() != actor_id
-        or str(created_by or "").strip() != actor_id
+        str(created_by or "").strip() != actor_id
         or str(org_id or "").strip() != actor_org_id
     ):
         return make_error_response(
             "task_actor_scope_denied",
             (
-                "An ordinary-turn task must be assigned to and created by the "
+                "An ordinary-turn task must be created by the "
                 "authenticated actor in the authenticated organisation."
             ),
         )
+    if not _task_continuation_assignee_allowed(assignee_id, actor_id, actor_org_id):
+        return make_error_response(
+            "task_assignment_scope_denied",
+            "The task creator may assign work to themselves, Von, or a coding "
+            "agent belonging to the same organisation.",
+        )
+    report_to_concept_id = str(report_to_concept_id or "").strip() or None
+    if report_to_concept_id not in (None, actor_id):
+        return make_error_response(
+            "task_report_scope_denied",
+            "Ordinary task creation may report to the authenticated actor.",
+        )
     if len(idempotency_key) > 512:
         return make_error_response(
             "INVALID_PARAM",
@@ -35301,6 +35337,10 @@ def _task_create(**kwargs):
             "request_id": request_id,
             "actor_id": actor_id,
             "organisation_concept_id": actor_org_id,
+            "assignee_concept_id": assignee_id,
+            "report_to_concept_id": report_to_concept_id,
+            "project_concept_id": kwargs.get("project_concept_id"),
+            "collection_concept_ids": kwargs.get("collection_concept_ids"),
             "title": title.strip(),
             "description": description.strip(),
             "originating_session_id": session_id,
@@ -35331,7 +35371,26 @@ def _task_create(**kwargs):
         ).encode("utf-8")
     ).hexdigest()
 
+    assignment_read_back_fields = {
+        "task_concept_id": _task_concept_id_for_agent_creation_fingerprint(
+            creation_fingerprint
+        ),
+        "created_by_concept_id": actor_id,
+        "organisation_concept_id": actor_org_id or None,
+        "assignee_concept_id": assignee_id,
+        "report_to_concept_id": report_to_concept_id,
+        **requested_settings,
+    }
+    if session_id:
+        from ...services.conversation_concept_service import (
+            _generate_conversation_concept_id,
+        )
+
+        assignment_read_back_fields["originating_conversation_id"] = (
+            _generate_conversation_concept_id(session_id)
+        )
     required_read_back_fields = {
+        **assignment_read_back_fields,
         "title": title.strip(),
         "description": description.strip(),
         "status": "pending",
@@ -35380,11 +35439,19 @@ def _idempotent_replay_response(task: dict[str, Any]) -> dict[str, Any]:
         # Existing tasks may since have progressed beyond pending. Verify their
         # durable creation fields without resetting or reassigning them.
         required = ("task_concept_id", "title", "description", "status", "priority")
-        if any(not task.get(field) for field in required):
+        mismatches = {
+            field: {"expected": expected, "actual": task.get(field)}
+            for field, expected in assignment_read_back_fields.items()
+            if task.get(field) != expected
+        }
+        if mismatches or any(not task.get(field) for field in required):
             incomplete = _incomplete_canonical_read_back_response(
                 task_concept_id=str(task.get("task_concept_id") or ""), task=task
             )
             incomplete["changed"] = False
+            incomplete["required_field_mismatches"] = (
+                mismatches or incomplete["required_field_mismatches"]
+            )
             return incomplete
         return {
             **task,
@@ -35409,6 +35476,16 @@ def _creation_receipt(task: dict[str, Any]) -> dict[str, Any]:
             "assignee_concept_id": task.get("assignee_concept_id"),
             "organisation_concept_id": task.get("organisation_concept_id"),
             "task_execution_verified": False,
+            "task_fields": {
+                key: task.get(key)
+                for key in (
+                    "assignee_concept_id",
+                    "report_to_concept_id",
+                    "requested_model",
+                    "requested_reasoning_effort",
+                    "parent_task_concept_id",
+                )
+            },
         }
 
     def _reconcile_after_create_failure() -> dict[str, Any] | None:
@@ -35481,7 +35558,6 @@ def _reconcile_after_create_failure() -> dict[str, Any] | None:
                 return reconciled_task
             raise
         task_concept_id = str(result.get("task_concept_id") or "")
-        required_read_back_fields["task_concept_id"] = task_concept_id
         try:
             canonical_read_back = get_task(task_concept_id)
         except Exception as exc:
@@ -36879,6 +36955,8 @@ def _task_continuation_readback(task_id, fields, task):
         "status": "verified" if not mismatches else "mismatch",
         "verified": not mismatches,
         "task": task,
+        "task_concept_id": task.get("task_concept_id"),
+        "task_fields": {key: task.get(key) for key in fields},
         "mismatched_fields": mismatches,
     }
 
@@ -44918,8 +44996,6 @@ def _build_default_catalogue_task_and_workflow_definitions() -> List[MethodDefin
             output_schema=task_create_output_schema,
             category="write",
             ordinary_turn_trusted_argument_bindings={
-                "assignee_id": "actor_user_concept_id",
-                "assignee_concept_id": "actor_user_concept_id",
                 "created_by_concept_id": "actor_user_concept_id",
                 "organisation_concept_id": "actor_organisation_concept_id",
                 "originating_session_id": "conversation_id",
@@ -44929,12 +45005,10 @@ def _build_default_catalogue_task_and_workflow_definitions() -> List[MethodDefin
             },
             ordinary_turn_fixed_arguments={
                 "session_id": None,
-                "report_to_concept_id": None,
-                "reports_to_concept_id": None,
             },
             ordinary_turn_effect=True,
             description=(
-                "Create a self-assigned Von task (stored as a Vontology concept) for "
+                "Create a Von task (stored as a Vontology concept) for "
                 "the authenticated actor and organisation, linked to the current "
                 "conversation. Identical retries in the same turn reuse the canonical "
                 "task. A durable workflow may instead provide a stable idempotency_key "
@@ -44942,9 +45016,16 @@ def _build_default_catalogue_task_and_workflow_definitions() -> List[MethodDefin
                 "Use this to track work items, action items, or to-dos; creating such "
                 "records can itself be the requested outcome. This operation does not "
                 "execute the recorded work, send mail, or create a calendar event. "
-                "The assignee is the authenticated actor, not a person named in the "
-                "title/description or an 'assignee' argument. Read the returned "
-                "assignee_concept_id when reporting responsibility. "
+                "Assignment defaults to the authenticated actor. Explicit assignee_id "
+                "or assignee_concept_id may select Von or an eligible coding agent in "
+                "the actor's organisation. report_to_concept_id may select the actor. "
+                "Read the returned assignee and execution preferences when reporting "
+                "responsibility. Standalone native coding-agent tasks do not require "
+                "project or collection membership for worker eligibility; eligibility "
+                "is not evidence of pickup or execution. Source conversation linkage "
+                "does not grant transcript access. parent_task_concept_id is unsupported "
+                "here and is rejected before creation; independent related work does "
+                "not require a parent. "
                 "Priority: low, medium, high, critical. Tasks start in 'pending' status and can include "
                 "planning metadata (components, fix versions, sprint values, backlog rank), canonical "
                 "task categories, source semantics, and richer task-detail fields. "
diff --git a/src/backend/services/adaptive_turn_service.py b/src/backend/services/adaptive_turn_service.py
index 0069764ca..daf0f275f 100644
--- a/src/backend/services/adaptive_turn_service.py
+++ b/src/backend/services/adaptive_turn_service.py
@@ -4388,6 +4388,7 @@ def _canonical_effect_readback_receipt(raw_payload: Any) -> dict[str, Any] | Non
             "task_status",
             "verified_outcome",
             "task_execution_verified",
+            "task_fields",
         )
         if key in readback
     }
@@ -4763,6 +4764,62 @@ def _effect_postcondition_identity(
     return hashlib.sha256(_json_bytes(identity)).hexdigest(), identity
 
 
+def _reconcile_task_update_attempts(
+    tool_invocations: Sequence[Mapping[str, Any]],
+    projected: dict[str, dict[str, Any]],
+) -> None:
+    """Match later verified fields on the same task without forgiving omissions."""
+    for index, invocation in enumerate(tool_invocations):
+        if (
+            invocation.get("execution_method", invocation.get("tool"))
+            != "task_update_fields"
+        ):
+            continue
+        state = projected.get(invocation.get("effect_id"), {})
+        if (
+            state.get("effect_status") not in {"failed", "not_started"}
+            or state.get("changed") is not False
+        ):
+            continue
+        arguments = invocation.get("effective_arguments") or {}
+        task_id = arguments.get("task_concept_id") or arguments.get("task_id")
+        fields = arguments.get("fields")
+        if not task_id or not isinstance(fields, Mapping) or not fields:
+            continue
+        recovered = {}
+        # Only the latest successful read-back is current evidence. A later
+        # contradictory update must not be masked by an earlier matching one.
+        for later in tool_invocations[index + 1 :]:
+            later_state = projected.get(later.get("effect_id"), {})
+            receipt = later_state.get("canonical_readback") or {}
+            later_arguments = later.get("effective_arguments") or {}
+            later_task_id = later_arguments.get(
+                "task_concept_id"
+            ) or later_arguments.get("task_id")
+            if (
+                later.get("execution_method", later.get("tool")) != "task_update_fields"
+                or later_task_id != task_id
+                or later_state.get("effect_status") != "succeeded"
+                or receipt.get("verified") is not True
+                or receipt.get("task_concept_id") != task_id
+            ):
+                continue
+            observed = receipt.get("task_fields") or {}
+            for field, expected in fields.items():
+                if field in observed:
+                    if observed[field] == expected:
+                        recovered[field] = later["effect_id"]
+                    else:
+                        recovered.pop(field, None)
+        if recovered:
+            state["recovered_fields"] = sorted(recovered)
+            state["unresolved_fields"] = sorted(set(fields) - recovered.keys())
+            if not state["unresolved_fields"]:
+                state["recovered_by_effect_id"] = list(recovered.values())[-1]
+                state["recovery_status"] = "succeeded"
+                state["reconciliation_basis"] = "later_exact_task_fields_readback"
+
+
 def _reconcile_effect_attempts_by_postcondition(
     *,
     tool_invocations: Sequence[Mapping[str, Any]],
@@ -4815,6 +4872,7 @@ def _reconcile_effect_attempts_by_postcondition(
             if success[0] > index
             else "prior_exact_postcondition_success"
         )
+    _reconcile_task_update_attempts(tool_invocations, projected)
     return projected
 
 
@@ -5935,7 +5993,9 @@ def _build_effect_outcome_report(
             "effect_id": effect_id,
             "tool": tool_name,
             "effect_status": effect_status,
-            "changed": state.get("changed") if isinstance(state.get("changed"), bool) else None,
+            "changed": (
+                state.get("changed") if isinstance(state.get("changed"), bool) else None
+            ),
             "initial_effect_status": state.get("initial_effect_status"),
             "current_outcome_status": state.get("current_outcome_status"),
             # A handler-level canonical read-back already resolves a successful
@@ -5947,9 +6007,7 @@ def _build_effect_outcome_report(
                 or (effect_status == "succeeded" and canonical_readback_verified)
             ),
             "reconciliation_status": state.get("reconciliation_status"),
-            "canonical_readback_present": isinstance(
-                canonical_readback, Mapping
-            ),
+            "canonical_readback_present": isinstance(canonical_readback, Mapping),
             "canonical_readback_verified": canonical_readback_verified,
             "workflow_instance_operational_readback": (
                 workflow_instance_operational_readback
@@ -5957,9 +6015,7 @@ def _build_effect_outcome_report(
             "workflow_instance_readback_verified": (
                 workflow_instance_readback_verified
             ),
-            "workflow_instance_terminal_status": (
-                workflow_instance_terminal_status
-            ),
+            "workflow_instance_terminal_status": (workflow_instance_terminal_status),
             "workflow_id": state.get("workflow_id"),
             "instance_id": state.get("instance_id"),
             "target_ids": target_ids,
@@ -5972,6 +6028,19 @@ def _build_effect_outcome_report(
             "error": error_detail,
             "recovered_by_effect_id": state.get("recovered_by_effect_id"),
             "argument_identity": argument_identity or None,
+            "recovered_fields": state.get("recovered_fields"),
+            "unresolved_fields": state.get("unresolved_fields"),
+            "task_concept_id": (
+                canonical_readback.get("task_concept_id")
+                if isinstance(canonical_readback, Mapping)
+                else None
+            ),
+            "task_fields": (
+                canonical_readback.get("task_fields")
+                if isinstance(canonical_readback, Mapping)
+                and canonical_readback_verified
+                else None
+            ),
         }
         postcondition = _effect_postcondition_identity(invocation)
         if postcondition is not None:
@@ -5992,7 +6061,41 @@ def _build_effect_outcome_report(
         "model_error": "The model did not produce a reliable final answer.",
         "model_non_answer": "The model did not produce a usable final answer.",
     }.get(terminal_status, "This turn did not finish cleanly.")
+    from urllib.parse import quote
+
     lines = ["## Effect outcome report", "", heading]
+    task_records: dict[str, dict[str, Any]] = {}
+    for fact in facts:
+        task_id = fact.get("task_concept_id")
+        if (
+            task_id
+            and fact.get("canonical_readback_verified")
+            and fact.get("effect_status") == "succeeded"
+        ):
+            task_records.setdefault(task_id, {}).update(fact.get("task_fields") or {})
+    for task_id, fields in task_records.items():
+        lines.extend(
+            (
+                "",
+                f"Task [{task_id}](/?concept_id={quote(task_id, safe='')}) is confirmed.",
+            )
+        )
+        for field, label in (
+            ("assignee_concept_id", "Assigned to"),
+            ("report_to_concept_id", "Reports to"),
+            ("requested_model", "Requested model"),
+            ("requested_reasoning_effort", "Requested reasoning effort"),
+            ("parent_task_concept_id", "Parent task"),
+        ):
+            if field in fields:
+                value = fields[field]
+                lines.append(
+                    f"{label}: `{value}`."
+                    if value is not None
+                    else f"{label}: none recorded."
+                )
+        lines.append("These task records do not verify worker pickup or execution.")
+
     confirmed = [
         fact
         for fact in facts
@@ -6106,6 +6209,14 @@ def _build_effect_outcome_report(
             label = f"`{fact['tool']}`"
             status = str(fact.get("effect_status") or "unknown")
             details = [f"status `{status}`"]
+            if fact.get("recovered_fields"):
+                details.append(
+                    "later canonical read-back confirmed fields: "
+                    + ", ".join(fact["recovered_fields"])
+                )
+                details.append(
+                    "still unresolved: " + ", ".join(fact["unresolved_fields"])
+                )
             if int(fact.get("attempt_count") or 1) > 1:
                 details.append(
                     f"{fact['attempt_count']} attempts for this exact postcondition"
diff --git a/tests/backend/test_codex_von_worker.py b/tests/backend/test_codex_von_worker.py
index 989bbf0e3..3614e3026 100644
--- a/tests/backend/test_codex_von_worker.py
+++ b/tests/backend/test_codex_von_worker.py
@@ -457,6 +457,7 @@ def fail_fetch(args, **kwargs):
             "-C",
             str(source),
             "config",
+            "--local",
             "--get",
             "credential.https://github.com.helper",
         ],
@@ -464,3 +465,43 @@ def fail_fetch(args, **kwargs):
         capture_output=True,
     )
     assert untouched.returncode == 1
+
+
+def test_backend_root_rejects_copied_bundle_without_services(tmp_path):
+    with pytest.raises(RuntimeError, match="--backend-root"):
+        worker.bind_backend_root(tmp_path)
+
+
+def test_backend_root_rejects_already_imported_other_checkout(tmp_path):
+    service = tmp_path / "src/backend/services/task_management_service.py"
+    service.parent.mkdir(parents=True)
+    service.write_text("# fixture only\n")
+    with pytest.raises(RuntimeError, match="already belong to another checkout"):
+        worker.bind_backend_root(tmp_path)
+
+
+def test_backend_root_accepts_current_canonical_checkout(monkeypatch):
+    root = Path(__file__).resolve().parents[2]
+    monkeypatch.setattr(sys, "path", list(sys.path))
+    assert worker.bind_backend_root(root) == root
+    assert sys.path[0] == str(root)
+
+
+@pytest.mark.parametrize("missing", ["requested_model", "requested_reasoning_effort"])
+def test_old_backend_read_cannot_silently_launch_worker_defaults(
+    config, monkeypatch, missing
+):
+    row = task(config, requested_model="gpt-6-astra", requested_reasoning_effort="high")
+    row.pop(missing)
+    api = worker.Von(config)
+    monkeypatch.setattr(
+        api.tasks, "search_tasks", lambda **_: {"tasks": [row], "total": 1}
+    )
+    monkeypatch.setattr(api.tasks, "get_task", lambda _: row)
+    monkeypatch.setattr(
+        worker, "launch", lambda *a: pytest.fail("incompatible reader launched work")
+    )
+    with pytest.raises(RuntimeError, match="matching backend and worker release"):
+        worker.tick(config, api, 0)
+    with pytest.raises(RuntimeError, match="execution-preference fields"):
+        api.task(row["task_concept_id"])
diff --git a/tests/backend/test_coding_task_creation_recovery.py b/tests/backend/test_coding_task_creation_recovery.py
new file mode 100644
index 000000000..427cbf99b
--- /dev/null
+++ b/tests/backend/test_coding_task_creation_recovery.py
@@ -0,0 +1,474 @@
+"""JVNAUTOSCI-2750: isolated ordinary-turn → native task → worker replay.
+
+No real task, invitation, message, workflow, or model request is issued.
+"""
+
+from copy import deepcopy
+from importlib.util import module_from_spec, spec_from_file_location
+from pathlib import Path
+
+import mongomock
+import pytest
+
+from src.backend.db.repositories.concepts_repository import ConceptsRepository
+from src.backend.db.repositories.text_value_repository import (
+    TextRelationsRepository,
+    TextValuesRepository,
+)
+from src.backend.integrations.internal_mcp import build_default_catalogue, catalogue
+from src.backend.integrations.internal_mcp.gateway import InternalMCPGateway
+from src.backend.integrations.internal_mcp.transport import InternalMCPTransport
+from src.backend.security.access_control import override_current_actor
+from src.backend.services import (
+    concept_service,
+    conversation_concept_service as conversations,
+)
+from src.backend.services import organisation_membership_service as membership
+from src.backend.services import (
+    task_management_service as tasks,
+    text_value_service as texts,
+)
+from src.backend.services.adaptive_turn_service import (
+    _trusted_tool_payload,
+    _model_visible_input_schema,
+)
+
+
+@pytest.fixture
+def native_store(monkeypatch):
+    db = mongomock.MongoClient()["isolated_coding_task_acceptance_2750"]
+    db.concepts.create_index("concept_id", unique=True)
+    monkeypatch.setattr(ConceptsRepository, "collection", lambda: db.concepts)
+    # The fixture does not replicate the ontology catalogue used for target
+    # visibility. Actor/org query filtering and the assignment checks stay real.
+    monkeypatch.setattr(
+        "src.backend.db.repositories.concepts_repository.sanitize_concept_document",
+        lambda doc: doc,
+    )
+    monkeypatch.setattr(TextRelationsRepository, "collection", lambda: db.relations)
+    monkeypatch.setattr(TextValuesRepository, "collection", lambda: db.values)
+    monkeypatch.setattr(texts, "can_access_concept", lambda _: True)
+    monkeypatch.setattr(texts, "filter_accessible_concept_ids", lambda ids: set(ids))
+    monkeypatch.setattr(texts, "_emit_text_relation_mutation_event", lambda **_: None)
+    monkeypatch.setattr(texts, "_invalidate_stats_for_predicate_change", lambda _: None)
+    monkeypatch.setattr(
+        texts,
+        "_invalidate_workflow_routing_projection_for_text_relation_change",
+        lambda **_: None,
+    )
+    monkeypatch.setattr(
+        tasks, "resolve_task_work_product", lambda _: {"status": "missing"}
+    )
+    monkeypatch.setattr(tasks, "maybe_launch_task_created_workflow", lambda **_: None)
+    monkeypatch.setattr(tasks, "maybe_launch_task_status_workflow", lambda **_: None)
+    monkeypatch.setattr(
+        tasks, "maybe_launch_effort_unit_completed_workflow", lambda **_: None
+    )
+    monkeypatch.setattr(tasks, "ensure_effort_unit_ontology", lambda **_: None)
+    monkeypatch.setattr(
+        concept_service,
+        "get_concept_by_concept_id_exact",
+        lambda cid: {
+            "relationships": {
+                "is_an_instance_of": (
+                    ["#V#coding_agent"] if cid == "#V#worker" else ["#V#person"]
+                )
+            }
+        },
+    )
+    monkeypatch.setattr(
+        membership,
+        "is_user_member_of_organisation",
+        lambda user, org: user in {"#V#alice", "#V#worker"} and org == "#V#lab",
+    )
+    monkeypatch.setattr(
+        tasks,
+        "is_user_member_of_organisation",
+        membership.is_user_member_of_organisation,
+    )
+    monkeypatch.setattr(
+        conversations,
+        "get_or_create_conversation_concept",
+        lambda **kw: conversations._generate_conversation_concept_id(kw["session_id"]),
+    )
+    monkeypatch.setattr(
+        conversations,
+        "get_conversation_concept",
+        lambda cid: {"session_id": "fixture-source", "name": "Isolated acceptance"},
+    )
+    return db
+
+
+@pytest.fixture
+def gateway():
+    return InternalMCPGateway(
+        catalogue=build_default_catalogue(),
+        transport=InternalMCPTransport(),
+        enabled=True,
+    )
+
+
+def payload(gateway, **changes):
+    return _trusted_tool_payload(
+        gateway=gateway,
+        tool_name="task_create",
+        model_payload={
+            "title": "Isolated coding assignment acceptance",
+            "description": "Fixture only; never launch coding work.",
+            "assignee_concept_id": "#V#worker",
+            "report_to_concept_id": "#V#alice",
+            "requested_model": "gpt-6-astra",
+            "requested_reasoning_effort": "high",
+            "created_by_concept_id": "#V#spoof",
+            "organisation_concept_id": "#V#spoof",
+            "originating_session_id": "spoof",
+            "session_id": "spoof",
+            "request_id": "spoof",
+            **changes,
+        },
+        trusted_argument_values={
+            "actor_user_concept_id": "#V#alice",
+            "actor_organisation_concept_id": "#V#lab",
+            "turn_namespace": "#V#alice@lab",
+            "conversation_id": "fixture-source",
+            "turn_id": "fixture-turn",
+        },
+    )
+
+
+def load_worker(path):
+    spec = spec_from_file_location("fixture_worker", path)
+    module = module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def test_ordinary_create_retry_native_readback_and_worker_settings(
+    native_store, gateway, monkeypatch
+):
+    request = payload(gateway)
+    assert request["created_by_concept_id"] == "#V#alice"
+    assert request["organisation_concept_id"] == "#V#lab"
+    assert request["originating_session_id"] == "fixture-source"
+    assert request["session_id"] is None
+    with override_current_actor("#V#alice", "#V#lab"):
+        first = gateway.invoke("task_create", request).payload
+        if not first.get("success") and first.get("task_concept_id"):
+            tasks.get_task(first["task_concept_id"])
+        assert first.get("success"), first
+        repeated = gateway.invoke("task_create", request).payload
+        assert repeated.get("success"), repeated
+        task = tasks.get_task(first["task_concept_id"])
+    assert repeated["task_concept_id"] == task["task_concept_id"]
+    assert repeated["changed"] is False
+    assert native_store.concepts.count_documents({}) == 1
+    assert task["created_by_concept_id"] == "#V#alice"
+    assert task["organisation_concept_id"] == "#V#lab"
+    assert task["assignee_concept_id"] == "#V#worker"
+    assert task["report_to_concept_id"] == "#V#alice"
+    assert task[
+        "originating_conversation_id"
+    ] == conversations._generate_conversation_concept_id("fixture-source")
+    assert task["requested_model"] == "gpt-6-astra"
+    assert task["requested_reasoning_effort"] == "high"
+    assert task["project_concept_id"] is None and task["collection_concept_ids"] == []
+    assert task["parent_task_concept_id"] is None
+    assert first["canonical_readback"]["task_execution_verified"] is False
+
+    worker = load_worker(
+        Path(__file__).resolve().parents[2] / "scripts/codex_von_worker.py"
+    )
+    config = {
+        "agent_id": "#V#worker",
+        "delegator_id": "#V#alice",
+        "organisation_id": "#V#lab",
+    }
+    api = worker.Von(config)
+    monkeypatch.setattr(
+        tasks, "search_tasks", lambda **_: {"tasks": [task], "total": 1}
+    )
+    monkeypatch.setattr(
+        tasks, "list_task_comments", lambda *a, **kw: {"comments": [], "total": 0}
+    )
+    monkeypatch.setattr(api.messages, "get_messages_for_user", lambda *a, **kw: [])
+    assert list(api.pending()) == [task]
+    assert worker.resolve_execution_settings(
+        config, api.inputs(api.task(task["task_concept_id"]))
+    ) == {
+        "model": "gpt-6-astra",
+        "model_source": "task",
+        "reasoning_effort": "high",
+        "reasoning_effort_source": "task",
+    }
+    from src.backend.services import shared_conversation_service as shares
+
+    monkeypatch.setattr(shares, "list_invites_for_user", lambda **_: [])
+    assert api.conversation(task)["available"] is False
+
+
+@pytest.mark.parametrize(
+    "changes,code",
+    [
+        ({"assignee_concept_id": "#V#outsider"}, "task_assignment_scope_denied"),
+        ({"report_to_concept_id": "#V#outsider"}, "task_report_scope_denied"),
+        ({"assignee_id": "#V#alice"}, "INVALID_PARAM"),
+        ({"reports_to_concept_id": "#V#outsider"}, "INVALID_PARAM"),
+        (
+            {"parent_task_concept_id": "#V#required_parent"},
+            "task_create_parent_unsupported",
+        ),
+    ],
+)
+def test_invalid_assignment_or_parent_has_no_partial_creation(
+    native_store, gateway, changes, code
+):
+    with override_current_actor("#V#alice", "#V#lab"):
+        result = gateway.invoke("task_create", payload(gateway, **changes)).payload
+    assert result["error_code"] == code
+    assert native_store.concepts.count_documents({}) == 0
+
+
+@pytest.mark.parametrize(
+    "field,value",
+    [
+        ("assignee_concept_id", "#V#alice"),
+        ("report_to_concept_id", None),
+        ("requested_model", "gpt-5.6-terra"),
+        ("requested_reasoning_effort", "medium"),
+        ("originating_conversation_id", "#V#wrong"),
+        ("task_concept_id", "#V#wrong"),
+    ],
+)
+def test_retry_does_not_verify_wrong_assignment_preferences_or_identity(
+    native_store, gateway, monkeypatch, field, value
+):
+    with override_current_actor("#V#alice", "#V#lab"):
+        first = gateway.invoke("task_create", payload(gateway)).payload
+        assert first.get("success"), first
+        wrong = {**first["canonical_read_back"], field: value}
+        monkeypatch.setattr(
+            tasks, "find_task_by_agent_creation_fingerprint", lambda **_: wrong
+        )
+        result = gateway.invoke("task_create", payload(gateway)).payload
+    assert result["success"] is False
+    assert result["changed"] is False
+    assert field in result["required_field_mismatches"]
+    assert native_store.concepts.count_documents({}) == 1
+
+
+def test_create_schema_exposes_preferences_without_exposing_trusted_creator(gateway):
+    schema = _model_visible_input_schema(gateway.get_method_definition("task_create"))
+    assert {
+        "assignee_id",
+        "assignee_concept_id",
+        "report_to_concept_id",
+        "requested_model",
+        "requested_reasoning_effort",
+    } <= schema["properties"].keys()
+    assert "created_by_concept_id" not in schema["properties"]
+
+
+def test_creation_failed_parent_update_and_assignment_recovery_answer(
+    native_store, gateway, monkeypatch
+):
+    from src.backend.languagemodels.structured_tool_calling.types import (
+        LLMResponse,
+        ToolCall,
+    )
+    from src.backend.services.adaptive_turn_service import execute_adaptive_turn
+    from src.backend.services import turn_execution_record_service
+
+    monkeypatch.setattr(
+        "src.backend.db.mongo_client.get_concepts_collection",
+        lambda: native_store.concepts,
+    )
+    monkeypatch.setattr(
+        turn_execution_record_service,
+        "record_effect_observation_phase",
+        lambda **kw: {
+            "updated": True,
+            "duplicate": False,
+            "phase": kw.get("phase"),
+            "stored_phase": kw.get("observation", {}),
+        },
+    )
+    fields = {
+        "assignee_concept_id": "#V#worker",
+        "report_to_concept_id": "#V#alice",
+        "requested_model": "gpt-6-astra",
+        "requested_reasoning_effort": "high",
+    }
+
+    class IncidentReplay:
+        call = 0
+
+        def generate_with_tools(self, prompt, available_tools, **kwargs):
+            self.call += 1
+            if self.call == 1:
+                name = "task_create"
+                arguments = {
+                    "title": "Isolated recovered assignment",
+                    "description": "Fixture, no real work.",
+                    "assignee_concept_id": "#V#alice",
+                }
+            else:
+                task_id = native_store.concepts.find_one({})["concept_id"]
+                if self.call == 4:
+                    return LLMResponse(
+                        text_response=f"Created {task_id} and assigned it to #V#worker with Astra/high. The requested parent relationship is unresolved. Worker pickup is unverified."
+                    )
+                assert self.call < 4, "unexpected replay model call"
+                name = "task_update_fields"
+                arguments = {
+                    "task_concept_id": task_id,
+                    "fields": {
+                        **fields,
+                        **(
+                            {"parent_task_concept_id": "#V#required_parent"}
+                            if self.call == 2
+                            else {}
+                        ),
+                    },
+                }
+            return LLMResponse(
+                text_response="",
+                tool_calls=[
+                    ToolCall(
+                        tool_name="turn_invoke_capability",
+                        call_id=f"fixture-{self.call}",
+                        payload={"name": name, "arguments": arguments},
+                    )
+                ],
+            )
+
+    result = execute_adaptive_turn(
+        gateway=gateway,
+        prompt="Create a coding task assigned to our coding agent with Astra/high, reporting to me, and link it as a child of the required parent.",
+        context=[],
+        llm_client=IncidentReplay(),
+        model="fixture-no-model-request",
+        user_namespace="#V#alice@lab",
+        user_concept_id="#V#alice",
+        org_concept_id="#V#lab",
+        conversation_id="fixture-source",
+        turn_id="fixture-incident-recovery",
+        turn_budget_seconds=20,
+        final_synthesis_reserve_seconds=2,
+    )
+    assert result.terminal_status == "effect_partially_completed", result.response_text
+    task_id = native_store.concepts.find_one({})["concept_id"]
+    with override_current_actor("#V#alice", "#V#lab"):
+        task = tasks.get_task(task_id)
+    assert task["assignee_concept_id"] == "#V#worker"
+    assert task["requested_reasoning_effort"] == "high"
+    assert task["parent_task_concept_id"] is None
+    authoritative = result.response_text.split("### Model draft")[0]
+    assert f"[{task_id}]" in authoritative
+    assert "#V#worker" in authoritative
+    assert "gpt-6-astra" in authoritative
+    assert "later canonical read-back confirmed fields" in authoritative
+    assert "still unresolved: parent_task_concept_id" in authoritative
+    assert "pickup" in authoritative
+    assert native_store.concepts.count_documents({}) == 1
+
+
+@pytest.mark.parametrize("members", [{"#V#alice"}, {"#V#worker"}])
+def test_create_requires_both_live_memberships(
+    native_store, gateway, monkeypatch, members
+):
+    monkeypatch.setattr(
+        membership,
+        "is_user_member_of_organisation",
+        lambda user, org: user in members and org == "#V#lab",
+    )
+    with override_current_actor("#V#alice", "#V#lab"):
+        result = gateway.invoke("task_create", payload(gateway)).payload
+    assert result["error_code"] == "task_assignment_scope_denied"
+    assert native_store.concepts.count_documents({}) == 0
+
+
+def test_retry_keeps_progressed_task_and_detects_changed_explicit_intent(
+    native_store, gateway
+):
+    with override_current_actor("#V#alice", "#V#lab"):
+        request = payload(gateway, idempotency_key="fixture")
+        first = gateway.invoke("task_create", request).payload
+        assert first.get("success"), first
+        task_id = first["task_concept_id"]
+        tasks.update_task_fields(task_id, fields={"status": "completed"})
+        replay = gateway.invoke("task_create", request).payload
+        assert replay["success"] and replay["status"] == "completed"
+        conflicting = gateway.invoke(
+            "task_create", {**request, "requested_reasoning_effort": "medium"}
+        ).payload
+        assert conflicting["success"] is False and conflicting["changed"] is False
+        assert tasks.get_task(task_id)["status"] == "completed"
+    assert native_store.concepts.count_documents({}) == 1
+
+
+def test_recovery_requires_same_task_and_matching_canonical_fields():
+    from src.backend.services.adaptive_turn_service import (
+        _reconcile_effect_attempts_by_postcondition,
+    )
+
+    failed = {
+        "effect_id": "failed",
+        "tool": "task_update_fields",
+        "effective_arguments": {
+            "task_concept_id": "#V#task",
+            "fields": {"assignee_concept_id": "#V#worker"},
+        },
+    }
+    succeeded = {
+        "effect_id": "success",
+        "tool": "task_update_fields",
+        "effective_arguments": {
+            "task_concept_id": "#V#task",
+            "fields": {"assignee_concept_id": "#V#worker"},
+        },
+    }
+    state = {
+        "failed": {"effect_status": "failed", "changed": False},
+        "success": {
+            "effect_status": "succeeded",
+            "canonical_readback": {
+                "verified": True,
+                "task_concept_id": "#V#task",
+                "task_fields": {"assignee_concept_id": "#V#worker"},
+            },
+        },
+    }
+    reconciled = _reconcile_effect_attempts_by_postcondition(
+        tool_invocations=[failed, succeeded], effect_snapshot=state
+    )
+    assert reconciled["failed"]["recovery_status"] == "succeeded"
+    for changes in (
+        {"task_concept_id": "#V#other"},
+        {"verified": False},
+        {"task_fields": {"assignee_concept_id": "#V#wrong"}},
+    ):
+        changed = deepcopy(state)
+        changed["success"]["canonical_readback"].update(changes)
+        reconciled = _reconcile_effect_attempts_by_postcondition(
+            tool_invocations=[failed, succeeded], effect_snapshot=changed
+        )
+        assert "recovery_status" not in reconciled["failed"]
+    # A subsequent contradictory update invalidates the earlier matching value.
+    changed = deepcopy(state)
+    changed["reassignment"] = {
+        "effect_status": "succeeded",
+        "canonical_readback": {
+            "verified": True,
+            "task_concept_id": "#V#task",
+            "task_fields": {"assignee_concept_id": "#V#different"},
+        },
+    }
+    reconciled = _reconcile_effect_attempts_by_postcondition(
+        tool_invocations=[
+            failed,
+            succeeded,
+            {**succeeded, "effect_id": "reassignment"},
+        ],
+        effect_snapshot=changed,
+    )
+    assert "recovery_status" not in reconciled["failed"]
diff --git a/tests/backend/test_internal_mcp_catalogue_builds.py b/tests/backend/test_internal_mcp_catalogue_builds.py
index 0fa6c011c..33c74bd1d 100644
--- a/tests/backend/test_internal_mcp_catalogue_builds.py
+++ b/tests/backend/test_internal_mcp_catalogue_builds.py
@@ -464,8 +464,6 @@ def test_ordinary_turn_read_projection_follows_capability_authority_metadata():
     task_create_definition = catalogue.get("task_create")
     assert task_create_definition.ordinary_turn_effect is True
     assert task_create_definition.ordinary_turn_trusted_argument_bindings == {
-        "assignee_id": "actor_user_concept_id",
-        "assignee_concept_id": "actor_user_concept_id",
         "created_by_concept_id": "actor_user_concept_id",
         "organisation_concept_id": "actor_organisation_concept_id",
         "originating_session_id": "conversation_id",
diff --git a/tests/backend/test_internal_mcp_direct_message_tools.py b/tests/backend/test_internal_mcp_direct_message_tools.py
index a28edb0af..4ecb83560 100644
--- a/tests/backend/test_internal_mcp_direct_message_tools.py
+++ b/tests/backend/test_internal_mcp_direct_message_tools.py
@@ -242,14 +242,14 @@ def test_message_and_task_identity_arguments_are_server_bound():
         },
         trusted_argument_values=trusted,
     )
-    assert task_payload["assignee_id"] == "#V#user_alice"
-    assert task_payload["assignee_concept_id"] == "#V#user_alice"
+    assert task_payload["assignee_id"] == "#V#spoofed"
+    assert "assignee_concept_id" not in task_payload
     assert task_payload["created_by_concept_id"] == "#V#user_alice"
     assert task_payload["organisation_concept_id"] == "#V#org_test"
     assert task_payload["originating_session_id"] == "conversation-trusted"
     assert task_payload["acting_user_concept_id"] == "#V#user_alice"
     assert task_payload["request_id"] == "turn-trusted"
-    assert task_payload["report_to_concept_id"] is None
+    assert task_payload["report_to_concept_id"] == "#V#outsider"
 
 
 def test_agent_visible_message_and_task_schemas_hide_trusted_identity_fields():
@@ -269,8 +269,6 @@ def test_agent_visible_message_and_task_schemas_hide_trusted_identity_fields():
         "request_id",
         "namespace",
         "author_concept_id",
-        "assignee_id",
-        "assignee_concept_id",
         "created_by_concept_id",
     }
     for tool_name, required_fields in expected_required.items():
@@ -294,6 +292,40 @@ def _owned_task(*, status: str = "pending") -> dict[str, Any]:
     }
 
 
+def _created_task(kwargs):
+    from src.backend.services.task_management_service import (
+        _task_concept_id_for_agent_creation_fingerprint,
+    )
+    from src.backend.services.conversation_concept_service import (
+        _generate_conversation_concept_id,
+    )
+
+    return {
+        **_owned_task(),
+        **{
+            key: kwargs.get(key)
+            for key in (
+                "title",
+                "description",
+                "assignee_concept_id",
+                "created_by_concept_id",
+                "organisation_concept_id",
+                "report_to_concept_id",
+                "requested_model",
+                "requested_reasoning_effort",
+            )
+        },
+        "task_concept_id": _task_concept_id_for_agent_creation_fingerprint(
+            kwargs["agent_creation_fingerprint"]
+        ),
+        "originating_conversation_id": (
+            _generate_conversation_concept_id(kwargs["originating_session_id"])
+            if kwargs.get("originating_session_id")
+            else None
+        ),
+    }
+
+
 def _task_actor_arguments() -> dict[str, str]:
     return {
         "acting_user_concept_id": "#V#user_alice",
@@ -453,11 +485,7 @@ def fake_find(**kwargs):
 
     def fake_create(**kwargs):
         create_calls["count"] += 1
-        task = {
-            **_owned_task(),
-            "requested_model": kwargs.get("requested_model"),
-            "requested_reasoning_effort": kwargs.get("requested_reasoning_effort"),
-        }
+        task = _created_task(kwargs)
         persisted.update(
             {
                 "task": task,
@@ -519,11 +547,7 @@ def fake_find(**kwargs):
 
     def fake_create(**kwargs):
         create_calls["count"] += 1
-        task = {
-            **_owned_task(),
-            "title": kwargs["title"],
-            "description": kwargs["description"],
-        }
+        task = _created_task(kwargs)
         persisted.update(
             {
                 "task": task,
@@ -587,14 +611,7 @@ def test_task_create_actor_scoped_idempotency_key_does_not_cross_actor(monkeypat
 
     def fake_create(**kwargs):
         fingerprints.append(kwargs["agent_creation_fingerprint"])
-        task = {
-            **_owned_task(),
-            "task_concept_id": f"#V#task_{len(fingerprints)}",
-            "title": kwargs["title"],
-            "description": kwargs["description"],
-            "assignee_concept_id": kwargs["assignee_concept_id"],
-            "created_by_concept_id": kwargs["created_by_concept_id"],
-        }
+        task = _created_task(kwargs)
         tasks[task["task_concept_id"]] = task
         return task
 
@@ -635,7 +652,8 @@ def fake_find(**_kwargs):
         find_calls["count"] += 1
         return None if find_calls["count"] == 1 else canonical_task
 
-    def fake_create(**_kwargs):
+    def fake_create(**kwargs):
+        canonical_task.update(_created_task(kwargs))
         raise task_management_service.TaskManagementError("duplicate concept_id")
 
     monkeypatch.setattr(
diff --git a/tests/backend/test_internal_mcp_task_parity_tools.py b/tests/backend/test_internal_mcp_task_parity_tools.py
index fb9436baa..4ef8ea7ce 100644
--- a/tests/backend/test_internal_mcp_task_parity_tools.py
+++ b/tests/backend/test_internal_mcp_task_parity_tools.py
@@ -1024,6 +1024,11 @@ def _fake_add_task_attachment(
 
 def test_task_create_gateway_supports_start_date_and_epic(monkeypatch):
     gateway = _build_gateway()
+    from src.backend.services.task_management_service import (
+        _task_concept_id_for_agent_creation_fingerprint,
+    )
+
+    canonical = {}
     from src.backend.security.access_control import override_current_actor
 
     def _fake_create_task(
@@ -1056,8 +1061,10 @@ def _fake_create_task(
         project_concept_id=None,
         collection_concept_ids=None,
     ):
-        return {
-            "task_concept_id": "#V#task_123",
+        result = {
+            "task_concept_id": _task_concept_id_for_agent_creation_fingerprint(
+                agent_creation_fingerprint
+            ),
             "title": title,
             "description": description,
             "status": "pending",
@@ -1085,6 +1092,8 @@ def _fake_create_task(
             "agent_creation_fingerprint": agent_creation_fingerprint,
             "agent_creation_request_id": agent_creation_request_id,
         }
+        canonical.update(result)
+        return result
 
     monkeypatch.setattr(
         "src.backend.services.task_management_service.create_task",
@@ -1092,13 +1101,7 @@ def _fake_create_task(
     )
     monkeypatch.setattr(
         "src.backend.services.task_management_service.get_task",
-        lambda task_concept_id: {
-            "task_concept_id": task_concept_id,
-            "title": "Task with dates",
-            "description": "Details",
-            "status": "pending",
-            "priority": "medium",
-        },
+        lambda task_concept_id: dict(canonical),
     )
     monkeypatch.setattr(
         "src.backend.services.task_management_service.find_task_by_agent_creation_fingerprint",
@@ -1116,7 +1119,7 @@ def _fake_create_task(
                 "epic_task_concept_id": "#V#task_epic_1",
                 "task_type_ids": ["#V#delegated_task_specification"],
                 "task_source_id": "#V#jira_imported_task_source",
-                "report_to_concept_id": "#V#user_manager",
+                "report_to_concept_id": "#V#user_alice",
                 "task_role": "Communicator",
                 "next_checkpoint": "Tomorrow morning",
                 "progress_signal": "Confirmed by chat",
@@ -1157,19 +1160,13 @@ def _fake_create_task(
         },
     )
     assert material == verified == {"task-create"}
-    assert payload["canonical_read_back"] == {
-        "task_concept_id": "#V#task_123",
-        "title": "Task with dates",
-        "description": "Details",
-        "status": "pending",
-        "priority": "medium",
-    }
+    assert payload["canonical_read_back"] == canonical
     assert payload.get("start_date") == "2026-03-01T10:00:00+00:00"
     assert payload.get("due_date") == "2026-03-05T10:00:00+00:00"
     assert payload.get("epic_task_concept_id") == "#V#task_epic_1"
     assert payload.get("task_type_ids") == ["#V#delegated_task_specification"]
     assert payload.get("task_source_id") == "#V#jira_imported_task_source"
-    assert payload.get("report_to_concept_id") == "#V#user_manager"
+    assert payload.get("report_to_concept_id") == "#V#user_alice"
     assert payload.get("reference_code") == "TASK-001"
     _assert_schema_conformance(gateway, "task_create", payload)
 
@@ -1180,18 +1177,30 @@ def test_task_create_gateway_marks_required_persistence_failure_indeterminate(
     wrong_readback_id,
 ):
     gateway = _build_gateway()
+    from src.backend.services.task_management_service import (
+        _task_concept_id_for_agent_creation_fingerprint,
+    )
+
+    identity = {}
     from src.backend.security.access_control import override_current_actor
 
     monkeypatch.setattr(
         "src.backend.services.task_management_service.create_task",
         lambda **_kwargs: {
-            "task_concept_id": "#V#task_partial",
+            "task_concept_id": identity.setdefault(
+                "id",
+                _task_concept_id_for_agent_creation_fingerprint(
+                    _kwargs["agent_creation_fingerprint"]
+                ),
+            ),
             "title": "Create a task",
             "description": "Persist the required fields.",
             "status": "pending",
             "priority": "medium",
             "required_text_persistence_failures": (
-                [] if wrong_readback_id else [
+                []
+                if wrong_readback_id
+                else [
                     {
                         "predicate": "#V#hasPriority",
                         "exception_type": "TimeoutError",
@@ -1203,11 +1212,16 @@ def test_task_create_gateway_marks_required_persistence_failure_indeterminate(
     monkeypatch.setattr(
         "src.backend.services.task_management_service.get_task",
         lambda task_concept_id: {
-            "task_concept_id": "#V#wrong_task" if wrong_readback_id else task_concept_id,
+            "task_concept_id": (
+                "#V#wrong_task" if wrong_readback_id else task_concept_id
+            ),
             "title": "Create a task",
             "description": "Persist the required fields.",
             "status": "pending",
             "priority": "medium",
+            "created_by_concept_id": "#V#user_alice",
+            "assignee_concept_id": "#V#user_alice",
+            "organisation_concept_id": "#V#org_test",
         },
     )
     monkeypatch.setattr(
@@ -1235,10 +1249,10 @@ def test_task_create_gateway_marks_required_persistence_failure_indeterminate(
     assert payload["error_code"] == "task_canonical_read_back_incomplete"
     assert "canonical_readback" not in payload
     assert payload["effect_status"] == "indeterminate"
-    assert payload["task_concept_id"] == "#V#task_partial"
+    assert payload["task_concept_id"] == identity["id"]
     if wrong_readback_id:
         assert payload["required_field_mismatches"] == {
-            "task_concept_id": {"expected": "#V#task_partial", "actual": "#V#wrong_task"}
+            "task_concept_id": {"expected": identity["id"], "actual": "#V#wrong_task"}
         }
     else:
         assert payload["required_field_mismatches"] == {}