Allow explicitly delegated task images to reach coding workers

docs/engineering/codex_dgx_worker.md

@@ -186,6 +186,31 @@ and five-minute schedule. A selected release is not evidence of actual pickup.
 Local installation and public deployment remain distinct effects; both need
 applicable task or standing authority.
 
+### Task-delegated source images
+
+An operator may explicitly authorise the controller to retrieve task-referenced
+images using the delegator's access when the coding agent cannot read the upload
+directly. Enable `delegated_task_images` in the private worker configuration with
+`enabled: true` and an `authorisation_reference` pointing to the operator's grant.
+This setting defaults off. It applies to any configured coding-agent identity;
+interactive delivery adapters must use the same staging function and copy its
+private files to the executing host before advertising local paths.
+
+Before this handoff, the controller re-reads the canonical native task and checks
+its active status, creator, assignee, report recipient and organisation. Only file
+references in that task's description/evidence/notes, attachment metadata and
+delegator-authored task comments/replies qualify. Other conversation context and
+prior model output cannot extend the eligible set. The existing canonical source
+read still checks the delegator's image access. The coding process receives
+bounded image copies and a manifest with original concept IDs, SHA-256, source
+actor, task and authorisation reference; it receives no owner credentials.
+Reassignment or cancellation prevents a new handoff. Previously staged bytes
+are retained as execution evidence; this does not promise retroactive revocation.
+
+Conversation-image uploader checks remain unchanged. An invitation to the source
+conversation alone is not this delegation. Polling alone does not prove that a
+new run consumed its staged images: verify local copies and the actual run result.
+
 ### Coherent worker/backend releases
 
 `scripts/codex_von_release.py` prepares a complete detached checkout of one

scripts/codex_von_worker.py

@@ -201,6 +201,60 @@ def referenced_tasks(config, api, supplied, task_ids=()):
     }
 
 
+def delegated_task_image_context(config, supplied):
+    """Resolve operator-authorised source images from a fresh assigned task.
+
+    Context from another conversation, a prior result or the model cannot add
+    identifiers to this handoff. Credentials stay in the trusted controller.
+    """
+    policy = config.get("delegated_task_images") or {}
+    if policy.get("enabled") is not True or not policy.get("authorisation_reference"):
+        return {}
+    task_id = supplied.get("task_id") if isinstance(supplied, dict) else None
+    if not task_id:
+        return {}
+    from src.backend.services.file_copy_reference_service import (
+        extract_file_copy_concept_ids_from_text,
+    )
+
+    api = Von(config)
+    task = api.task(task_id)
+    if not (
+        authorised_task(task, config)
+        and task.get("status") in ACTIVE
+        and api.native_writer(task)
+    ):
+        return {}
+    inputs = api.inputs(task)
+    sources = {
+        "description": task.get("description"),
+        "evidence": task.get("evidence"),
+        "notes": task.get("notes"),
+        "attachments": inputs.get("attachments"),
+        "comments": [
+            row
+            for row in inputs.get("comments", [])
+            if row.get("author_concept_id") == config["delegator_id"]
+        ],
+        "replies": [
+            row
+            for row in inputs.get("replies", [])
+            if row.get("sender_id") == config["delegator_id"]
+            and row.get("organisation_concept_id") == config["organisation_id"]
+        ],
+    }
+    return {
+        "task_id": task_id,
+        "source_actor": config["delegator_id"],
+        "recipient_actor": config["agent_id"],
+        "organisation_id": config["organisation_id"],
+        "authorisation_reference": policy["authorisation_reference"],
+        "file_copy_concept_ids": extract_file_copy_concept_ids_from_text(
+            json.dumps(sources, default=str)
+        ),
+    }
+
+
 def stage_file_copy_evidence(config, supplied, run_dir):
     """Controller-only authorised byte reads; expose local copies, never blob URLs."""
     from src.backend.services.computer_file_copy_service import (
@@ -212,6 +266,10 @@ def stage_file_copy_evidence(config, supplied, run_dir):
     )
 
     ids = extract_file_copy_concept_ids_from_text(json.dumps(supplied, default=str))
+    try:
+        delegation = delegated_task_image_context(config, supplied)
+    except Exception as exc:
+        delegation = {"lookup_error_type": type(exc).__name__}
     results = []
     for concept_id in ids[:20]:
         item = {
@@ -222,12 +280,52 @@ def stage_file_copy_evidence(config, supplied, run_dir):
             "scope": "configured worker actor and organisation",
         }
         try:
+            record = None
             result = fetch_file_copy_bytes(
                 file_copy_concept_id=concept_id,
                 user_concept_id=config["agent_id"],
                 organisation_concept_id=config["organisation_id"],
                 max_bytes=5_000_000,
             )
+            if (
+                not result.get("success")
+                and result.get("error") == "not_found"
+                and concept_id in delegation.get("file_copy_concept_ids", [])
+            ):
+                from src.backend.security.access_control import override_current_actor
+
+                # This task-scoped read uses an explicit operator grant. The
+                # coding process receives image copies, never the owner's access.
+                with override_current_actor(
+                    config["delegator_id"], config["organisation_id"]
+                ):
+                    result = fetch_file_copy_bytes(
+                        file_copy_concept_id=concept_id,
+                        user_concept_id=config["delegator_id"],
+                        organisation_concept_id=config["organisation_id"],
+                        max_bytes=5_000_000,
+                    )
+                    if result.get("success"):
+                        if not (result["info"].content_type or "").startswith("image/"):
+                            result = {
+                                "success": False,
+                                "error": "delegated_image_required",
+                            }
+                        else:
+                            record = (
+                                build_file_copy_artifact_record(
+                                    file_copy_concept_id=concept_id
+                                )
+                                or {}
+                            )
+                            item.update(
+                                scope="explicit task-scoped source-image delegation",
+                                source_actor=config["delegator_id"],
+                                delegated_task_id=delegation["task_id"],
+                                authorisation_reference=delegation[
+                                    "authorisation_reference"
+                                ],
+                            )
             if not result.get("success"):
                 item["error_code"] = result.get("error")
                 item["status"] = (
@@ -238,10 +336,11 @@ def stage_file_copy_evidence(config, supplied, run_dir):
             else:
                 data = result["data"]
                 digest = hashlib.sha256(data).hexdigest()
-                record = (
-                    build_file_copy_artifact_record(file_copy_concept_id=concept_id)
-                    or {}
-                )
+                if record is None:
+                    record = (
+                        build_file_copy_artifact_record(file_copy_concept_id=concept_id)
+                        or {}
+                    )
                 expected = record.get("sha256")
                 item.update(
                     sha256=digest,
@@ -277,6 +376,7 @@ def stage_file_copy_evidence(config, supplied, run_dir):
         "results": results,
         "omitted": max(0, len(ids) - 20),
         "source": "canonical actor-scoped file-copy service",
+        "delegation_lookup_error": delegation.get("lookup_error_type"),
     }
 
 

tests/backend/test_codex_von_worker.py

@@ -809,3 +809,92 @@ def fetch(**kwargs):
         assert item["status"] == (
             "checksum_mismatch" if outcome == "mismatch" else "not_found_in_actor_scope"
         )
+
+
+@pytest.mark.parametrize(
+    "case,owner_read,available",
+    [
+        ("allowed", True, True),
+        ("disabled", False, False),
+        ("no_grant", False, False),
+        ("reassigned", False, False),
+        ("other_creator", False, False),
+        ("other_org", False, False),
+        ("cancelled", False, False),
+        ("external_writer", False, False),
+        ("context_only", False, False),
+        ("other_author", False, False),
+        ("not_image", True, False),
+        ("checksum_mismatch", True, False),
+    ],
+)
+def test_source_images_require_explicit_current_task_delegation(
+    config, tmp_path, monkeypatch, case, owner_read, available
+):
+    import hashlib
+    from src.backend.services import computer_file_copy_service as files
+
+    config["delegated_task_images"] = {
+        "enabled": case != "disabled",
+        "authorisation_reference": "" if case == "no_grant" else "operator-grant",
+    }
+    current = task(config, description=FILE_COPY)
+    if case == "reassigned":
+        current["assignee_concept_id"] = "#V#other"
+    if case == "other_creator":
+        current["created_by_concept_id"] = "#V#other"
+    if case == "other_org":
+        current["organisation_concept_id"] = "#V#other"
+    if case == "cancelled":
+        current["status"] = "cancelled"
+    if case in {"context_only", "other_author"}:
+        current["description"] = "A task with no image reference"
+    comments = [{"author_concept_id": "#V#other", "body": FILE_COPY}]
+    api = SimpleNamespace(
+        task=lambda _: current,
+        native_writer=lambda _: case != "external_writer",
+        inputs=lambda _: {"comments": comments},
+    )
+    monkeypatch.setattr(worker, "Von", lambda _: api)
+    calls = []
+    data = b"private task screenshot fixture"
+    digest = hashlib.sha256(data).hexdigest()
+
+    def fetch(**kwargs):
+        calls.append(kwargs["user_concept_id"])
+        if kwargs["user_concept_id"] == config["agent_id"]:
+            return {"success": False, "error": "not_found"}
+        return {
+            "success": True,
+            "data": data,
+            "info": SimpleNamespace(
+                original_filename="screenshot.png",
+                content_type="text/plain" if case == "not_image" else "image/png",
+            ),
+        }
+
+    monkeypatch.setattr(files, "fetch_file_copy_bytes", fetch)
+    monkeypatch.setattr(
+        files,
+        "build_file_copy_artifact_record",
+        lambda **_: {
+            "sha256": "0" * 64 if case == "checksum_mismatch" else digest,
+        },
+    )
+    output = worker.stage_file_copy_evidence(
+        config,
+        {"task_id": current["task_concept_id"], "prior_result": FILE_COPY},
+        tmp_path,
+    )
+    item = output["results"][0]
+    assert (config["delegator_id"] in calls) is owner_read
+    assert item["content_available"] is available
+    if available:
+        assert Path(item["local_path"]).read_bytes() == data
+        assert Path(item["local_path"]).stat().st_mode & 0o777 == 0o600
+        assert item["sha256"] == digest
+        assert item["source_actor"] == config["delegator_id"]
+        assert item["delegated_task_id"] == current["task_concept_id"]
+        assert item["authorisation_reference"] == "operator-grant"
+    else:
+        assert "local_path" not in item