diff --git a/docs/engineering/codex_dgx_worker.md b/docs/engineering/codex_dgx_worker.md
index b748bfe1..16d61eca 100644
--- a/docs/engineering/codex_dgx_worker.md
+++ b/docs/engineering/codex_dgx_worker.md
@@ -672,3 +672,57 @@ writes (registration, one batch, final reference), independently of event count.
 text, refresh, scoped download and clearing on scope change using an isolated
 browser fixture. These tests do not prove live scheduler activation, public
 OAuth, actual provider summaries, or production historical backfill.
+
+### Supervisor repair tasks
+
+An operator-owned `supervision_enabled: true` enables repair-task handoffs in
+an existing controller. It adds no schedule or model poll. Instantiate additional
+agents with that optional flag and their existing agent, delegator and organisation
+bindings; configure the ontology relation through canonical relationship services.
+Do not copy credentials or infer delegation from a supervisor relation.
+
+`report_to_concept_id` is the task override (`#V#reportsto`). When unset,
+`#V#has_supervisor` on the assignee supplies the default for people and agents.
+Canonical task point reads and creation responses expose `reporting_resolution`
+with the selected concept, source and resolution status. Defaults are not copied
+into the explicit field. An ambiguous, inaccessible or cyclic route is reported
+without admitting another coding run. Existing controllers keep their prior
+eligibility policy unless supervision is enabled; enabled controllers still
+require the configured creator/delegator, assignee, organisation and native writer.
+Reporting responsibility no longer substitutes for those execution checks.
+
+Under its existing lock, the controller retains an episode fingerprint before
+creating a canonical repair on behalf of the existing delegator. The repair
+records that attribution, source task, blocker and agent chain in
+`external_references.coding_supervision`, inherits explicit execution preferences,
+and links back with `blocks`/`blocked_by`. The controller reconciles the deterministic
+creation fingerprint after an interrupted acknowledgement. A same-blocker manual
+retry retains the repair episode and refreshes its attempt context; a verified
+recovery followed by a new failure starts a new episode. Self-reports, archives,
+status resets and elapsed time do not supply recovery evidence. Cancelled or
+reassigned repair tasks require reconciliation, not automatic duplication.
+
+The supervisor inspects retained work, repairs within the original scope, and
+returns `repair_receipt_json` containing exact fresh observations. The finishing
+controller checks the current source instructions, scope and assignment, then
+records a canonical attestation. The source admits that attempt-bound receipt once
+through the shared retry policy. A completed repair status alone is insufficient.
+Legacy unclassified failures remain reviewable but require an explicit reviewed
+continuation; the controller does not invent their missing observation contract.
+A human supervisor receives a native task and Von notification, never a coding
+process in the human's identity. Supervisor chains reject repeated identities.
+
+For an upgrade, install the coherent release and update a private Mac adapter
+under its existing lock. The Mac adapter calls `api.reconcile_supervision` for
+waiting attempts and `api.record_supervisor_repair` before applying a successful
+result. Carry `supervision` state to a reserved continuation alongside consumed
+retry tokens. Exclude derived reporting projections and controller-verified repair
+comments from the Mac instruction fingerprint; recheck retry admission at begin.
+Keep old retained results readable when nullable result fields are added.
+
+Before clearing legacy generated `report_to_concept_id` defaults, retain an exact
+review manifest: task ID, old value, original creation provenance, later reporting
+changes and source instruction. Preserve deliberate choices and ambiguous cases.
+Apply only reviewed entries after the eligibility fix is installed, through
+canonical task updates, and read back the effective ontology default. Do not use
+a blanket rewrite based solely on the current field matching the delegator.
diff --git a/scripts/codex_von_release.py b/scripts/codex_von_release.py
index 12f49c3a..d9fdeda1 100644
--- a/scripts/codex_von_release.py
+++ b/scripts/codex_von_release.py
@@ -58,6 +58,15 @@ def verify(target, commit):
         raise runtime.DeploymentError(
             "Incomplete worker release: scripts/codex_von_retry.py"
         )
+    if "codex_von_supervision" in (target / "scripts/codex_von_worker.py").read_text():
+        for dependency in (
+            "scripts/codex_von_supervision.py",
+            "src/backend/services/task_reporting_service.py",
+        ):
+            if not (target / dependency).is_file():
+                raise runtime.DeploymentError(
+                    "Incomplete worker release: " + dependency
+                )
     return {
         "commit": commit,
         "backend_root": str(target),
diff --git a/scripts/codex_von_retry.py b/scripts/codex_von_retry.py
index 4f82c02b..e541451e 100644
--- a/scripts/codex_von_retry.py
+++ b/scripts/codex_von_retry.py
@@ -204,6 +204,20 @@ def admission(config, state, inputs, *, now=None):
                 "reason": receipt["reason"],
                 "receipt": receipt,
             }
+    for row in inputs.get("supervisor_repair_comments", []):
+        receipt = (row.get("source") or {}).get("coding_supervisor_repair")
+        token = "supervisor-comment:" + str(row.get("comment_id"))
+        if (
+            row.get("comment_id")
+            and token not in consumed
+            and verified_repair(blocker, receipt, now)
+        ):
+            return {
+                "kind": "verified_repair",
+                "token": token,
+                "reason": receipt["reason"],
+                "receipt": receipt,
+            }
     return None
 
 
@@ -213,5 +227,10 @@ def consume(state, decision):
     if token:
         state.setdefault("consumed_retry_tokens", []).append(token)
     state["launch_admission"] = decision
+    if state.get("supervision") and decision.get("kind") in {
+        "verified_repair",
+        "probe_recovery",
+    }:
+        state["supervision"]["resolved"] = True
     state.pop("blocker", None)
     state.pop("waiting_report", None)
diff --git a/scripts/codex_von_supervision.py b/scripts/codex_von_supervision.py
new file mode 100644
index 00000000..43dc314f
--- /dev/null
+++ b/scripts/codex_von_supervision.py
@@ -0,0 +1,317 @@
+"""Controller-owned supervisor handoffs, using canonical tasks and retry receipts.
+
+The config is operator-owned. A reporting relationship never grants a coding
+identity permission to impersonate its supervisor. The controller creates a
+bounded repair on behalf of its existing delegator, retaining the source chain.
+"""
+
+import json
+from datetime import UTC, datetime
+
+try:
+    from . import codex_von_retry as retry
+except ImportError:
+    import codex_von_retry as retry
+
+
+def resolution(task):
+    from src.backend.services.task_reporting_service import resolve_task_reporting
+
+    return resolve_task_reporting(task)
+
+
+def source_authorised(config, task):
+    return (
+        task.get("assignee_concept_id") == config["agent_id"]
+        and task.get("created_by_concept_id") == config["delegator_id"]
+        and task.get("organisation_concept_id") == config["organisation_id"]
+        and task.get("status") in {"pending", "in_progress", "blocked"}
+    )
+
+
+def reconcile(config, api, task, state, persist):
+    """Called under the existing controller lock. Persist intents before effects."""
+    if not config.get("supervision_enabled"):
+        return None
+    task = api.task(state["task_id"])
+    if not source_authorised(config, task) or not api.native_writer(task):
+        return None
+    blocker = retry.retain_blocker(config, state)
+    if not blocker:
+        return None
+    # Unclassified legacy failures need retained-evidence review, not invented
+    # observations. Transient failures keep their existing bounded probe first.
+    route = resolution(task)
+    parent = (task.get("external_references") or {}).get("coding_supervision") or {}
+    chain = list(parent.get("agent_chain") or []) + [config["agent_id"]]
+    supervisor = route.get("concept_id")
+    if supervisor in chain:
+        route = dict(route, status="supervision_cycle")
+    if route["status"] != "resolved":
+        content = (
+            f"Supervisor handoff unavailable: {route['status']} ({route['source']}). "
+            f"Configure one accessible reporting recipient for {task['task_concept_id']} "
+            "or its assignee. The source remains waiting; no coding retry was launched."
+        )
+        error_key = retry.digest([state["task_id"], blocker["attempt"], route])
+        if state.get("supervision_error", {}).get("key") != error_key:
+            receipt = api.send(task, error_key + ":supervisor-error", content)
+            state["supervision_error"] = {
+                "key": error_key,
+                "message_id": receipt,
+                "route": route,
+            }
+            persist()
+        return None
+    allowed, _ = api.messages.authorise_direct_message_participants(
+        sender_id=config["agent_id"],
+        recipient_ids=[supervisor],
+        organisation_concept_id=config["organisation_id"],
+    )
+    if not allowed:
+        raise PermissionError("Supervisor is not an accessible organisation member")
+    episode = state.get("supervision")
+    signature = retry.digest(
+        [
+            blocker["key"],
+            blocker["scope"],
+            supervisor,
+            task.get("title"),
+            task.get("description"),
+        ]
+    )
+    if not episode or episode.get("resolved") or episode.get("signature") != signature:
+        episode = {
+            "signature": signature,
+            "key": retry.digest([blocker["binding"], blocker["attempt"], signature]),
+            "failed_attempt": blocker["attempt"],
+            "route": route,
+        }
+        state["supervision"] = episode
+        persist()
+    repair = api.tasks.find_task_by_agent_creation_fingerprint(
+        created_by_concept_id=config["delegator_id"],
+        organisation_concept_id=config["organisation_id"],
+        creation_fingerprint=episode["key"],
+    )
+    if repair is None:
+        # Only selected task data is copied, never the inbox or unrelated chats.
+        provenance = {
+            "source_task_id": state["task_id"],
+            "source_agent": config["agent_id"],
+            "delegator_id": config["delegator_id"],
+            "organisation_id": config["organisation_id"],
+            "episode_key": episode["key"],
+            "agent_chain": chain,
+            "blocker": blocker,
+            "source_instructions": {
+                k: task.get(k)
+                for k in (
+                    "title",
+                    "description",
+                    "requested_model",
+                    "requested_reasoning_effort",
+                )
+            },
+        }
+        context = {
+            "provenance": provenance,
+            "result": state.get("result"),
+            "worktree": state.get("worktree"),
+            "capture": state.get("capture"),
+            "execution_settings": state.get("execution_settings"),
+            "task_evidence": task.get("evidence"),
+            "source_comments": api.inputs(task).get("comments", []),
+            "source_notes": task.get("notes"),
+            "task_links": task.get("task_links"),
+        }
+        description = (
+            "Review the retained blocked work, repair the external prerequisite within the original "
+            "delegation, and verify the repaired condition. Do not repeat the failed coding run. "
+            "If blocked yourself, describe the concrete missing prerequisite for your supervisor.\n\n"
+            "The controller created this repair on behalf of the source delegator under their "
+            "standing supervisor-routing instruction. This does not grant credentials, unrelated "
+            "conversation access, or broader deployment authority.\n\n"
+            "For a classified blocker, return repair_receipt_json containing the original attempt, "
+            "blocker_key, binding, scope, observed_at (after repair), observations matching each "
+            "required_observations value, reason and evidence_reference. Observe the actual repaired "
+            "condition before attesting; completed status or passing unrelated CI is not evidence. "
+            "For an unclassified legacy failure, retain the diagnosis and ask the delegator for a "
+            "reviewed attempt-bound continuation; do not invent required observations.\n\n"
+            + json.dumps(context, indent=2, default=str)
+        )
+        repair = api.tasks.create_task(
+            "Unblock: " + task["title"],
+            description,
+            assignee_concept_id=supervisor,
+            created_by_concept_id=config["delegator_id"],
+            organisation_concept_id=config["organisation_id"],
+            priority=task.get("priority") or "medium",
+            report_to_concept_id=None,
+            requested_model=task.get("requested_model")
+            or config.get("model", "gpt-6-astra"),
+            requested_reasoning_effort=task.get("requested_reasoning_effort") or "high",
+            agent_creation_fingerprint=episode["key"],
+            agent_creation_request_id=episode["key"],
+            initial_external_references={"coding_supervision": provenance},
+        )
+    repair = api.task(repair["task_concept_id"])
+    reference = (repair.get("external_references") or {}).get(
+        "coding_supervision"
+    ) or {}
+    if (
+        repair.get("created_by_concept_id") != config["delegator_id"]
+        or repair.get("organisation_concept_id") != config["organisation_id"]
+        or reference.get("episode_key") != episode["key"]
+    ):
+        raise PermissionError(
+            "Repair task provenance does not match the retained episode"
+        )
+    if repair.get("assignee_concept_id") != supervisor or repair.get("status") not in {
+        "pending",
+        "in_progress",
+        "blocked",
+        "completed",
+    }:
+        raise PermissionError(
+            "Supervisor repair was cancelled or reassigned; reconcile the existing task"
+        )
+    if reference["blocker"]["attempt"] != blocker["attempt"]:
+        repair = api.tasks.refresh_coding_supervision_context(
+            repair["task_concept_id"],
+            episode_key=episode["key"],
+            blocker=blocker,
+            actor_concept_id=config["agent_id"],
+        )
+    episode["task_id"] = repair["task_concept_id"]
+    persist()
+    current = api.task(state["task_id"])
+    if not source_authorised(config, current) or resolution(current) != route:
+        return None
+    if not episode.get("linked"):
+        api.tasks.link_tasks(
+            repair["task_concept_id"],
+            state["task_id"],
+            link_type="blocks",
+            actor_concept_id=config["agent_id"],
+        )
+        current = api.task(state["task_id"])
+        if not any(
+            link.get("target_task_concept_id") == repair["task_concept_id"]
+            for link in current.get("task_links", [])
+        ):
+            raise RuntimeError("Source/repair link read-back failed")
+        episode["linked"] = True
+        persist()
+    content = (
+        f"Blocked task: {state['task_id']}\nSupervisor: {supervisor} ({route['source']})\n"
+        f"Repair task: {repair['task_concept_id']}\n{blocker['reason']}\n"
+        "The source is waiting for verified recovery, without another coding launch."
+    )
+    if not episode.get("message_id"):
+        # Notify Michael once. The assigned supervisor discovers the same canonical
+        # repair via its existing task queue; human supervisors get no coding run.
+        episode["message_id"] = api.send(task, episode["key"] + ":handoff", content)
+        persist()
+    if not episode.get("checkpoint"):
+        api.tasks.update_task_fields(
+            state["task_id"],
+            fields={"next_checkpoint": content},
+            actor_concept_id=config["agent_id"],
+        )
+        if api.task(state["task_id"]).get("next_checkpoint") != content:
+            raise RuntimeError("Supervisor checkpoint read-back failed")
+        episode["checkpoint"] = True
+        persist()
+    return episode
+
+
+def record_repair(config, api, repair_task, result):
+    """The finishing controller attests the supervisor's exact observations.
+
+    This is the existing trusted-producer receipt boundary, not a parser for
+    arbitrary comments or a completion-status inference.
+    """
+    raw = result.get("repair_receipt_json")
+    if (
+        not config.get("supervision_enabled")
+        or not raw
+        or result.get("status") != "completed"
+    ):
+        return None
+    parent = (repair_task.get("external_references") or {}).get(
+        "coding_supervision"
+    ) or {}
+    if not parent or not source_authorised(config, repair_task):
+        raise PermissionError("No current delegated supervisor repair")
+    source = api.task(parent["source_task_id"])
+    if (
+        source.get("status") not in {"blocked", "in_progress", "pending"}
+        or source.get("assignee_concept_id") != parent["source_agent"]
+        or source.get("created_by_concept_id") != config["delegator_id"]
+        or source.get("organisation_concept_id") != config["organisation_id"]
+        or resolution(source).get("concept_id") != config["agent_id"]
+        or any(source.get(k) != v for k, v in parent["source_instructions"].items())
+    ):
+        raise PermissionError("Original task authority or instructions changed")
+    receipt = json.loads(raw)
+    if not retry.verified_repair(parent["blocker"], receipt, datetime.now(UTC)):
+        raise ValueError("Supervisor did not supply matching fresh repair observations")
+    receipt = dict(
+        receipt,
+        supervisor_task_id=repair_task["task_concept_id"],
+        episode_key=parent["episode_key"],
+    )
+    # Reconcile a lost comment acknowledgement before creating another comment.
+    comments = api.tasks.list_task_comments(source["task_concept_id"], limit=500)[
+        "comments"
+    ]
+    for row in comments:
+        if (
+            row.get("author_concept_id") == config["agent_id"]
+            and (row.get("source") or {}).get("coding_supervisor_repair") == receipt
+        ):
+            return row
+    return api.tasks.add_task_comment(
+        source["task_concept_id"],
+        body="Supervisor verified recovery: " + receipt["reason"],
+        author_concept_id=config["agent_id"],
+        source={"coding_supervisor_repair": receipt},
+    )
+
+
+def repair_comments(config, api, task, rows):
+    """Accept only current, linked supervisor attestations; never bare metadata."""
+    if not config.get("supervision_enabled"):
+        return []
+    supervisor = resolution(task).get("concept_id")
+    accepted = []
+    for row in rows:
+        receipt = (row.get("source") or {}).get("coding_supervisor_repair")
+        if row.get("author_concept_id") != supervisor or not isinstance(receipt, dict):
+            continue
+        try:
+            repair = api.task(receipt["supervisor_task_id"])
+            parent = (repair.get("external_references") or {}).get(
+                "coding_supervision"
+            ) or {}
+            if (
+                repair.get("created_by_concept_id") != config["delegator_id"]
+                or repair.get("assignee_concept_id") != supervisor
+                or repair.get("organisation_concept_id") != config["organisation_id"]
+                or parent.get("source_task_id") != task["task_concept_id"]
+                or parent.get("episode_key") != receipt.get("episode_key")
+                or any(
+                    task.get(k) != v
+                    for k, v in parent.get("source_instructions", {}).items()
+                )
+                or not any(
+                    link.get("target_task_concept_id") == repair["task_concept_id"]
+                    for link in task.get("task_links", [])
+                )
+            ):
+                continue
+            accepted.append(row)
+        except (KeyError, api.tasks.TaskNotFoundError):
+            continue
+    return accepted
diff --git a/scripts/codex_von_worker.py b/scripts/codex_von_worker.py
index 4cfe0777..efcd134d 100755
--- a/scripts/codex_von_worker.py
+++ b/scripts/codex_von_worker.py
@@ -31,6 +31,14 @@
     except ImportError:
         from scripts import codex_von_retry as retry_policy
 
+try:
+    from . import codex_von_supervision as supervision
+except ImportError:
+    try:
+        import codex_von_supervision as supervision
+    except ImportError:
+        from scripts import codex_von_supervision as supervision
+
 ACTIVITY_CHECKPOINT_SECONDS = 15
 
 ACTIVE = {"pending", "in_progress", "blocked"}
@@ -43,8 +51,17 @@
         "question": {"type": "string"},
         "deploy_commit": {"type": "string"},
         "blocker": retry_policy.BLOCKER_SCHEMA,
+        "repair_receipt_json": {"type": ["string", "null"]},
     },
-    "required": ["status", "summary", "evidence", "question", "deploy_commit", "blocker"],
+    "required": [
+        "status",
+        "summary",
+        "evidence",
+        "question",
+        "deploy_commit",
+        "blocker",
+        "repair_receipt_json",
+    ],
     "additionalProperties": False,
 }
 
@@ -93,6 +110,7 @@ def capability_context(config, *, read_only, worktree=None):
         "controller_actions": ["reply", "resume_task", "create_task"],
         "deployment_enabled": bool(config.get("deployment_command")),
         "retry_probe_keys": sorted((config.get("retry_probes") or {}).keys()),
+        "supervision_enabled": bool(config.get("supervision_enabled")),
         "limitations": [
             "Use available shell tools for non-secret host/repository facts; missing supplied facts do not mean inspection is unavailable.",
             "Do not access credentials, private databases or live-service mutation routes.",
@@ -173,7 +191,10 @@ def authorised_task(task, config):
         task.get("assignee_concept_id") == config["agent_id"]
         and task.get("organisation_concept_id") == config["organisation_id"]
         and task.get("created_by_concept_id") == config["delegator_id"]
-        and task.get("report_to_concept_id") in (None, config["delegator_id"])
+        and (
+            config.get("supervision_enabled")
+            or task.get("report_to_concept_id") in (None, config["delegator_id"])
+        )
     )
 
 
@@ -443,14 +464,20 @@ def check_task(config, api, task_id):
 
 def validate_result(value):
     # Strict output requires every property; retained old results may omit blocker.
-    fields = set(RESULT_SCHEMA["required"]) - {"blocker"}
-    if not isinstance(value, dict) or set(value) - {"blocker"} not in (
+    optional = {"blocker", "repair_receipt_json"}
+    fields = set(RESULT_SCHEMA["required"]) - optional
+    if not isinstance(value, dict) or set(value) - optional not in (
         fields,
         fields - {"deploy_commit"},
     ):
         raise ValueError("Codex did not return the required result fields")
-    if any(not isinstance(v, str) for k, v in value.items() if k != "blocker"):
+    if any(not isinstance(v, str) for k, v in value.items() if k not in optional):
         raise ValueError("Codex result fields must be text")
+    if value.get("repair_receipt_json") is not None and (
+        not isinstance(value["repair_receipt_json"], str)
+        or not isinstance(json.loads(value["repair_receipt_json"]), dict)
+    ):
+        raise ValueError("Repair receipt must be a JSON object string or null")
     retry_policy.validate_blocker(value.get("blocker"))
     if value["status"] not in RESULT_SCHEMA["properties"]["status"]["enum"]:
         raise ValueError("Unknown Codex outcome")
@@ -518,9 +545,11 @@ def inputs(self, task):
         require_task_execution_preferences(task)
         task_id = task["task_concept_id"]
         comments = []
+        all_comments = []
         offset = 0
         while True:
             page = self.tasks.list_task_comments(task_id, offset=offset, limit=500)
+            all_comments.extend(page["comments"])
             comments.extend(
                 c
                 for c in page["comments"]
@@ -608,12 +637,47 @@ def inputs(self, task):
             "requested_model": task.get("requested_model"),
             "requested_reasoning_effort": task.get("requested_reasoning_effort"),
             "comments": comments,
+            "reporting_resolution": supervision.resolution(task)
+            if self.config.get("supervision_enabled")
+            else None,
+            "report_to_concept_id": task.get("report_to_concept_id"),
+            "supervisor_repair_comments": supervision.repair_comments(
+                self.config, self, task, all_comments
+            ),
             "replies": [
                 message for message in replies if not self.inbox_answered(message)
             ],
             **self.followup_context(task_id),
         }
 
+    def reconcile_supervision(self, task, state, persist):
+        try:
+            return supervision.reconcile(self.config, self, task, state, persist)
+        except Exception as exc:  # noqa: BLE001 - preserve failed handoffs while other tasks progress
+            key = (
+                fingerprint(
+                    [state["task_id"], state.get("attempt"), type(exc).__name__]
+                )
+                + ":supervision-failure"
+            )
+            if state.get("supervision_failure_message", {}).get("key") != key:
+                message_id = self.send(
+                    task,
+                    key,
+                    f"Supervisor handoff for {state['task_id']} could not be reconciled ({type(exc).__name__}). "
+                    "Review the existing repair task, current assignment, organisation membership and reporting configuration. "
+                    "The source remains waiting without another coding launch; retained state is available to the controller.",
+                )
+                state["supervision_failure_message"] = {
+                    "key": key,
+                    "message_id": message_id,
+                }
+                persist()
+            return None
+
+    def record_supervisor_repair(self, task, result):
+        return supervision.record_repair(self.config, self, task, result)
+
     def inbox_answered(self, message):
         if not self.config.get("inbox_enabled"):
             return False
@@ -723,7 +787,7 @@ def send(self, task, key, content):
             recipient_ids=[c["delegator_id"]],
             organisation_concept_id=c["organisation_id"],
             thread_id=task["task_concept_id"],
-            subject=f"Codex DGX: {task['title']}",
+            subject=f"{c.get('display_name', c['agent_id'])}: {task['title']}",
             content=content,
             metadata={"attribution": "Sent by the Codex DGX coding worker"},
         )
@@ -788,6 +852,7 @@ def finish(self, state):
             state["report_task"], state["attempt"] + ":result", state["report_content"]
         )
         if scope_valid:
+            self.record_supervisor_repair(task, result)
             evidence_addition = (
                 f"{result['evidence']}\nVon result: {message_id}\n"
                 f"Execution archive: {capture.get('archive_url', 'unavailable')}\n"
@@ -1163,9 +1228,9 @@ def apply_deployment(config, api, state, state_path, lock_fd):
         receipt
     )
     if success:
-        result[
-            "summary"
-        ] += f" Deployed and verified the public server at {commit[:12]}."
+        result["summary"] += (
+            f" Deployed and verified the public server at {commit[:12]}."
+        )
     else:
         result["status"] = "blocked"
         result["summary"] += (
@@ -1176,9 +1241,9 @@ def apply_deployment(config, api, state, state_path, lock_fd):
         )
     if recover_only:
         result["status"] = "blocked"
-        result[
-            "summary"
-        ] += " The task changed; only the interrupted deployment was reconciled."
+        result["summary"] += (
+            " The task changed; only the interrupted deployment was reconciled."
+        )
     write_json(state_path, state)
 
 
@@ -1398,6 +1463,12 @@ def tick(config, api, lock_fd):
                 # One task's unavailable reporting must not wedge the queue.
                 try:
                     report_waiting(config, api, task, state, path)
+                    if config.get("supervision_enabled"):
+                        api.reconcile_supervision(
+                            task,
+                            state,
+                            lambda path=path, state=state: write_json(path, state),
+                        )
                 except Exception as exc:  # noqa: BLE001 - canonical reporting errors stay isolated to this task
                     state["waiting_report_error"] = type(exc).__name__
                     write_json(path, state)
@@ -1457,9 +1528,9 @@ def tick(config, api, lock_fd):
                 state["result"]["summary"] = (
                     "Coding execution settings rejected: " + str(exc)
                 )
-                state["result"][
-                    "question"
-                ] = "Correct this task's model or reasoning setting, then give an explicit retry reason in the task thread."
+                state["result"]["question"] = (
+                    "Correct this task's model or reasoning setting, then give an explicit retry reason in the task thread."
+                )
             write_json(path, state)
         apply_deployment(config, api, state, path, lock_fd)
         finish_captured(config, api, state, path)
diff --git a/scripts/codex_von_worker_prompt.md b/scripts/codex_von_worker_prompt.md
index 69e0cc51..bef5d3a3 100644
--- a/scripts/codex_von_worker_prompt.md
+++ b/scripts/codex_von_worker_prompt.md
@@ -92,3 +92,14 @@ carry source context, not additional authority. Attachment metadata alone does
 not supply image/file contents. Preserve conditional instructions in a direct
 message assignment: establish whether coding or deployment is needed before
 requesting either. The controller, not the coding run, owns canonical effects.
+
+For a supervisor repair, `assignment.external_references.coding_supervision`
+contains the current source blocker and original delegated scope. It supersedes
+older attempts quoted in the repair description. Inspect retained work and
+observe the actual repaired condition. Return `repair_receipt_json` as the exact
+receipt object serialised to JSON, with `attempt`, `blocker_key`, `binding`,
+`scope`, fresh `observed_at`, `observations`, `reason` and `evidence_reference`.
+Use null when inapplicable or when recovery has not been verified. Completing a
+repair task alone never restarts its source. Do not invent missing observations
+for legacy failures; retain the diagnosis for a reviewed continuation. A human
+supervisor receives a native task and message, not a coding process.
diff --git a/src/backend/services/task_management_service.py b/src/backend/services/task_management_service.py
index 9c42076e..822cca63 100644
--- a/src/backend/services/task_management_service.py
+++ b/src/backend/services/task_management_service.py
@@ -1763,6 +1763,9 @@ def create_task(
             "Task-created workflow launch skipped for %s: %s", task_concept_id, e
         )
 
+    from .task_reporting_service import resolve_task_reporting
+
+    result["reporting_resolution"] = resolve_task_reporting(result)
     return result
 
 
@@ -1779,7 +1782,11 @@ def get_task(task_concept_id: str) -> Dict[str, Any]:
         TaskNotFoundError: If task not found
     """
     task_concept_id, doc = _get_task_doc(task_concept_id)
-    return {**_build_task_response(doc), "current_work_product": resolve_task_work_product(doc)}
+    from .task_reporting_service import resolve_task_reporting
+
+    result = {**_build_task_response(doc), "current_work_product": resolve_task_work_product(doc)}
+    result["reporting_resolution"] = resolve_task_reporting(result)
+    return result
 
 
 def find_task_by_agent_creation_fingerprint(
@@ -5980,3 +5987,56 @@ def get_task_taxonomy() -> Dict[str, Any]:
     "bulk_update_tasks",
     "delete_task",
 ]
+
+
+def refresh_coding_supervision_context(
+    repair_task_id: str,
+    *,
+    episode_key: str,
+    blocker: Dict[str, Any],
+    actor_concept_id: str,
+) -> Dict[str, Any]:
+    """Refresh a controller-created repair after the same unresolved blocker recurs.
+
+    Internal canonical service; not exposed as a model-selectable metadata write.
+    The originating agent can update only its own retained episode's observations.
+    Reporting responsibility does not change the creator/delegator or assignment.
+    """
+    repair_task_id, doc = _get_task_doc(repair_task_id)
+    references = (doc.get("metadata") or {}).get(
+        TASK_METADATA_KEY_EXTERNAL_REFERENCES
+    ) or {}
+    parent = references.get("coding_supervision") or {}
+    if (
+        parent.get("source_agent") != actor_concept_id
+        or parent.get("episode_key") != episode_key
+    ):
+        raise InvalidTaskDataError(
+            "Supervisor episode does not belong to this source controller"
+        )
+    source = get_task(parent["source_task_id"])
+    if (
+        source.get("assignee_concept_id") != actor_concept_id
+        or source.get("created_by_concept_id") != parent.get("delegator_id")
+        or source.get("organisation_concept_id") != parent.get("organisation_id")
+        or source.get("status") not in {"blocked", "pending", "in_progress"}
+        or blocker.get("binding") != parent["blocker"].get("binding")
+        or blocker.get("key") != parent["blocker"].get("key")
+        or blocker.get("scope") != parent["blocker"].get("scope")
+    ):
+        raise InvalidTaskDataError("Source authority or blocking episode changed")
+    updated = dict(parent)
+    updated["blocker"] = blocker
+    updated["prior_attempts"] = list(
+        dict.fromkeys([*parent.get("prior_attempts", []), parent["blocker"]["attempt"]])
+    )
+    ConceptsRepository.update_one(
+        {"concept_id": repair_task_id},
+        {
+            "$set": {
+                "metadata.external_references.coding_supervision": updated,
+                "updated_at": _now(),
+            }
+        },
+    )
+    return get_task(repair_task_id)
diff --git a/src/backend/services/task_reporting_service.py b/src/backend/services/task_reporting_service.py
new file mode 100644
index 00000000..98ad4085
--- /dev/null
+++ b/src/backend/services/task_reporting_service.py
@@ -0,0 +1,57 @@
+"""Resolve reporting responsibility without granting execution or data authority.
+
+Explicit task reports-to wins. An unset value uses the assignee's represented
+supervisor, for people and agents alike. Reads use the caller's actor scope.
+Defaults stay defaults: they are not copied into the task's explicit relation.
+"""
+
+from pymongo.errors import PyMongoError
+
+from src.backend.db.repositories.concepts_repository import ConceptsRepository
+from src.backend.utils.concept_id_utils import ensure_v_concept_prefix
+
+
+def _resolve_task_reporting(task):
+    explicit = task.get("report_to_concept_id")
+    assignee = task.get("assignee_concept_id")
+    source = "task.report_to_concept_id" if explicit else "assignee.#V#has_supervisor"
+    targets = [explicit] if explicit else []
+    if not explicit and assignee:
+        doc = ConceptsRepository.find_one(
+            {"concept_id": assignee}, {"relationships": 1}
+        )
+        if not doc:
+            return {
+                "status": "assignee_inaccessible",
+                "source": source,
+                "concept_id": None,
+            }
+        targets = (doc.get("relationships") or {}).get("#V#has_supervisor", [])
+        if isinstance(targets, str):
+            targets = [targets]
+    targets = list(
+        dict.fromkeys(ensure_v_concept_prefix(v) for v in targets if isinstance(v, str))
+    )
+    status = "resolved" if len(targets) == 1 else "ambiguous" if targets else "unset"
+    target = targets[0] if len(targets) == 1 else None
+    if target == assignee:
+        status = "self_reference"
+    if target and not ConceptsRepository.find_one(
+        {"concept_id": target}, {"concept_id": 1}
+    ):
+        status = "supervisor_inaccessible"
+    return {"status": status, "source": source, "concept_id": target}
+
+
+def resolve_task_reporting(task):
+    # Reporting discovery is useful context, not a reason to turn a successfully
+    # persisted task creation into an ambiguous failure after the write.
+    try:
+        return _resolve_task_reporting(task)
+    except (PyMongoError, RuntimeError) as exc:
+        return {
+            "status": "lookup_failed",
+            "source": "canonical ontology",
+            "concept_id": None,
+            "error_type": type(exc).__name__,
+        }
diff --git a/tests/backend/test_codex_von_supervision.py b/tests/backend/test_codex_von_supervision.py
new file mode 100644
index 00000000..245b864c
--- /dev/null
+++ b/tests/backend/test_codex_von_supervision.py
@@ -0,0 +1,223 @@
+"""Canonical-service/controller acceptance; isolated DB and tiny fixture process.
+
+No live coding worker or provider is launched. The same canonical task/message
+services and DGX loop used in production perform the effects under test.
+"""
+
+import copy
+import json
+from datetime import UTC, datetime
+
+import pytest
+from test_codex_von_retry import canonical_controller as controller_fixture
+from test_codex_von_retry import config as config_fixture
+from test_codex_von_retry import repair_receipt
+
+from scripts import codex_von_supervision as supervision
+from scripts import codex_von_worker as worker
+from src.backend.db.repositories.concepts_repository import ConceptsRepository
+
+canonical_controller = controller_fixture
+config = config_fixture
+
+
+def configure(config, monkeypatch):
+    config["supervision_enabled"] = True
+    config["display_name"] = "Fixture source"
+    for cid, targets in (
+        (config["agent_id"], ["#V#supervisor"]),
+        ("#V#supervisor", [config["delegator_id"]]),
+        (config["delegator_id"], []),
+    ):
+        ConceptsRepository.insert_one(
+            {"concept_id": cid, "relationships": {"#V#has_supervisor": targets}}
+        )
+    # Membership policy itself has dedicated service tests. Here all three
+    # principals are isolated authorised members; test denial separately.
+    from src.backend.services import message_service
+
+    monkeypatch.setattr(
+        message_service,
+        "authorise_direct_message_participants",
+        lambda **kw: (True, None),
+    )
+
+
+def test_lost_create_ack_two_hops_and_verified_continuation(
+    canonical_controller, monkeypatch
+):
+    config, api, create, read_state, launches, fd = canonical_controller
+    configure(config, monkeypatch)
+    source = create()
+    worker.tick(config, api, fd)
+    original_create = api.tasks.create_task
+    failed = False
+
+    def lost_ack(*args, **kwargs):
+        nonlocal failed
+        created = original_create(*args, **kwargs)
+        if kwargs.get("agent_creation_fingerprint") and not failed:
+            failed = True
+            raise OSError("Lost create acknowledgement after canonical effect")
+        return created
+
+    monkeypatch.setattr(api.tasks, "create_task", lost_ack)
+    worker.tick(config, api, fd)
+    assert failed
+    for _ in range(3):
+        worker.tick(config, api, fd)
+    _path, state = read_state(source)
+    episode = state["supervision"]
+    repair = api.task(episode["task_id"])
+    assert repair["assignee_concept_id"] == "#V#supervisor"
+    assert repair["created_by_concept_id"] == config["delegator_id"]
+    assert len(launches()) == 1
+    assert episode["linked"] and episode["message_id"]
+    assert "Repair task:" in api.task(source)["next_checkpoint"]
+    assert api.messages.get_message_for_user(
+        episode["message_id"], config["delegator_id"]
+    )
+    tasks = api.tasks.list_tasks(
+        organisation_concept_id=config["organisation_id"], limit=None
+    )
+    assert (
+        len(
+            [
+                t
+                for t in tasks
+                if (t.get("external_references") or {}).get("coding_supervision")
+            ]
+        )
+        == 1
+    )
+    # The linked repair's completed status alone does not admit the source.
+    api.tasks.update_task_status(repair["task_concept_id"], "completed")
+    worker.tick(config, api, fd)
+    assert len(launches()) == 1
+    api.tasks.update_task_status(repair["task_concept_id"], "in_progress")
+    supconfig = dict(config, agent_id="#V#supervisor")
+    supapi = worker.Von(supconfig)
+    receipt = repair_receipt(state)
+    receipt["observed_at"] = datetime.now(UTC).isoformat()
+    outcome = {"status": "completed", "repair_receipt_json": json.dumps(receipt)}
+    row = supapi.record_supervisor_repair(
+        supapi.task(repair["task_concept_id"]), outcome
+    )
+    again = supapi.record_supervisor_repair(
+        supapi.task(repair["task_concept_id"]), outcome
+    )
+    assert row["comment_id"] == again["comment_id"]
+    worker.tick(config, api, fd)
+    assert len(launches()) == 2
+    assert read_state(source)[1]["launch_admission"]["kind"] == "verified_repair"
+    # A later blocked supervisor goes to the human using the same controller,
+    # not a process launched in the human's identity.
+    supstate = copy.deepcopy(state)
+    supstate.update(task_id=repair["task_concept_id"], attempt="supervisor-failed")
+    supstate.pop("blocker", None)
+    supstate.pop("supervision", None)
+    result = supervision.reconcile(
+        supconfig,
+        supapi,
+        supapi.task(repair["task_concept_id"]),
+        supstate,
+        lambda: None,
+    )
+    human_task = api.task(result["task_id"])
+    assert human_task["assignee_concept_id"] == config["delegator_id"]
+    assert not worker.authorised_task(human_task, config)
+    assert not worker.authorised_task(human_task, supconfig)
+
+
+def test_explicit_reporting_and_cycles_keep_other_work_moving(
+    canonical_controller, monkeypatch
+):
+    config, api, create, read_state, launches, fd = canonical_controller
+    configure(config, monkeypatch)
+    source = create()
+    # An explicit reporting choice is preserved and no longer confused with
+    # execution delegation. Existing creator/org/assignee checks still apply.
+    api.tasks.update_task_fields(
+        source, fields={"report_to_concept_id": config["delegator_id"]}
+    )
+    assert worker.authorised_task(api.task(source), config)
+    assert (
+        supervision.resolution(api.task(source))["concept_id"] == config["delegator_id"]
+    )
+    worker.tick(config, api, fd)
+    worker.tick(config, api, fd)
+    assert (
+        api.task(read_state(source)[1]["supervision"]["task_id"])["assignee_concept_id"]
+        == config["delegator_id"]
+    )
+    other = create("Another task")
+    api.tasks.update_task_fields(
+        other, fields={"report_to_concept_id": config["agent_id"]}
+    )
+    worker.tick(config, api, fd)
+    worker.tick(config, api, fd)
+    assert (
+        read_state(other)[1]["supervision_error"]["route"]["status"]
+        == "supervision_cycle"
+    )
+    third = create("Eligible work despite cycle")
+    worker.tick(config, api, fd)
+    assert [x["task_id"] for x in launches()] == [source, other, third]
+
+
+def test_changed_source_and_nonmember_supervisor_denied(
+    canonical_controller, monkeypatch
+):
+    config, api, create, read_state, launches, fd = canonical_controller
+    configure(config, monkeypatch)
+    source = create()
+    worker.tick(config, api, fd)
+    worker.tick(config, api, fd)
+    _, state = read_state(source)
+    repair = api.task(state["supervision"]["task_id"])
+    supapi = worker.Von(dict(config, agent_id="#V#supervisor"))
+    receipt = repair_receipt(state)
+    receipt["observed_at"] = datetime.now(UTC).isoformat()
+    api.tasks.update_task_fields(
+        source, fields={"description": "Changed user instruction"}
+    )
+    with pytest.raises(PermissionError):
+        supapi.record_supervisor_repair(
+            repair, {"status": "completed", "repair_receipt_json": json.dumps(receipt)}
+        )
+    monkeypatch.setattr(
+        api.messages,
+        "authorise_direct_message_participants",
+        lambda **kw: (False, "membership revoked"),
+    )
+    with pytest.raises(PermissionError):
+        supervision.reconcile(config, api, api.task(source), state, lambda: None)
+    assert len(launches()) == 1
+
+
+def test_same_unresolved_blocker_reuses_repair_after_manual_retry(
+    canonical_controller, monkeypatch
+):
+    cfg, api, create, read_state, launches, fd = canonical_controller
+    configure(cfg, monkeypatch)
+    source = create()
+    worker.tick(cfg, api, fd)
+    worker.tick(cfg, api, fd)
+    _, first = read_state(source)
+    repair_id = first["supervision"]["task_id"]
+    api.tasks.add_task_comment(
+        source,
+        body="/retry-coding "
+        + first["attempt"]
+        + " reviewed retained effects; diagnose further",
+        author_concept_id=cfg["delegator_id"],
+    )
+    worker.tick(cfg, api, fd)
+    worker.tick(cfg, api, fd)
+    _, second = read_state(source)
+    assert len(launches()) == 2
+    assert second["attempt"] != first["attempt"]
+    assert second["supervision"]["task_id"] == repair_id
+    parent = api.task(repair_id)["external_references"]["coding_supervision"]
+    assert parent["blocker"]["attempt"] == second["attempt"]
+    assert first["attempt"] in parent["prior_attempts"]
diff --git a/tests/test_codex_von_release.py b/tests/test_codex_von_release.py
index e31222d6..f32514d6 100644
--- a/tests/test_codex_von_release.py
+++ b/tests/test_codex_von_release.py
@@ -28,6 +28,7 @@ def source(tmp_path):
         "codex_von_worker.py",
         "codex_von_inbox.py",
         "codex_von_retry.py",
+        "codex_von_supervision.py",
         "codex_von_deploy.py",
         "codex_von_release.py",
         "deploy_local_main.py",
@@ -40,6 +41,8 @@ def source(tmp_path):
     service = source / "src/backend/services/task_management_service.py"
     service.parent.mkdir(parents=True)
     service.write_text("RELEASE = 'old'\n")
+    (service.parent / "task_reporting_service.py").write_bytes(
+        (repo / "src/backend/services/task_reporting_service.py").read_bytes())
     preferences = source / "src/backend/utils/task_execution_preferences.py"
     preferences.parent.mkdir(parents=True)
     preferences.write_bytes(