Allow creator task handoff to coding-agent organisation members

docs/engineering/task_responsibility_continuation.md

@@ -49,7 +49,12 @@ context.
 
 Ordinary chat can use `task_update_fields` to attach the current product, revise
 the checkpoint and evidence, and assign an actor-created task to `#V#von_system`
-with the actor as its report recipient. This reuses the existing task editor
+with the actor as its report recipient. A creator may also assign to a concept
+explicitly typed as `#V#coding_agent` when both the actor and that agent have
+live membership of the authenticated organisation. This supports the
+[DGX coding worker](codex_dgx_worker.md); assignment does not itself grant
+conversation access or enlarge the worker's configured effect authority.
+This reuses the existing task editor
 and product reference. It does not start execution. The bounded ordinary-chat
 projection cannot change the creator or organisation, add an audience, or
 assign work to another person. The general parity editor remains a separate

src/backend/integrations/internal_mcp/catalogue.py

@@ -36872,6 +36872,42 @@ def _task_continuation_readback(task_id, fields, task):
     }
 
 
+def _task_continuation_assignee_allowed(assignee_id, actor_id, organisation_id):
+    """Allow the creator to hand work to a represented coding-agent member."""
+    from ...services.task_execution_service import VON_SYSTEM_CONCEPT_ID
+
+    if assignee_id in (actor_id, VON_SYSTEM_CONCEPT_ID):
+        return True
+    if not organisation_id or not isinstance(assignee_id, str):
+        return False
+
+    from ...services.coding_agent_identity_bootstrap_service import (
+        CODING_AGENT_TYPE_ID,
+    )
+    from ...services.concept_service import (
+        ConceptNotFoundError,
+        get_concept_by_concept_id_exact,
+    )
+    from ...services.organisation_membership_service import (
+        is_user_member_of_organisation,
+    )
+
+    try:
+        concept = get_concept_by_concept_id_exact(assignee_id)
+    except ConceptNotFoundError:
+        return False
+    types = (concept or {}).get("relationships", {}).get("is_an_instance_of", [])
+    if isinstance(types, str):
+        types = [types]
+    if not isinstance(types, list):
+        return False
+    return (
+        CODING_AGENT_TYPE_ID in types
+        and is_user_member_of_organisation(actor_id, organisation_id)
+        and is_user_member_of_organisation(assignee_id, organisation_id)
+    )
+
+
 def _task_update_fields(**kwargs):
     from ...services.task_management_service import (
         InvalidTaskDataError,
@@ -36929,16 +36965,19 @@ def _task_update_fields(**kwargs):
                 "task_continuation_fields_denied",
                 f"Fields outside ordinary task continuation: {unsupported}",
             )
-        from ...services.task_execution_service import VON_SYSTEM_CONCEPT_ID
-
         if "assignee_concept_id" in fields:
             if (
-                fields["assignee_concept_id"] not in (actor_id, VON_SYSTEM_CONCEPT_ID)
-                or current_task.get("created_by_concept_id") != actor_id
+                current_task.get("created_by_concept_id") != actor_id
+                or not _task_continuation_assignee_allowed(
+                    fields["assignee_concept_id"],
+                    actor_id,
+                    actor_scope.organisation_concept_id,
+                )
             ):
                 return make_error_response(
                     "task_assignment_scope_denied",
-                    "The task creator may assign continuing work to themselves or Von.",
+                    "The task creator may assign continuing work to themselves, Von, "
+                    "or a coding agent belonging to the same organisation.",
                 )
         if "report_to_concept_id" in fields and fields["report_to_concept_id"] not in (
             None,
@@ -45122,7 +45161,9 @@ def _build_default_catalogue_task_and_workflow_definitions() -> List[MethodDefin
                 "it to #V#von_system and set report_to_concept_id to themselves. "
                 "This preserves the task and does not launch execution or a schedule. "
                 "Ordinary chat edits only tasks created by or assigned to the "
-                "authenticated actor in the same organisation."
+                "authenticated actor in the same organisation. The creator may "
+                "assign to a represented coding agent with live membership of "
+                "that organisation."
             ),
         ),
         MethodDefinition(

tests/backend/test_task_continuation_mcp.py

@@ -9,7 +9,11 @@
 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 task_management_service
+from src.backend.services import (
+    concept_service,
+    organisation_membership_service,
+    task_management_service,
+)
 from src.backend.services.adaptive_turn_service import (
     _canonical_effect_readback_receipt,
     _canonically_verified_material_effect_ids,
@@ -28,6 +32,9 @@ def task_state(monkeypatch):
         "current_work_product": {"status": "missing"},
     }
     writes = []
+    monkeypatch.setattr(
+        concept_service, "get_concept_by_concept_id_exact", lambda _: None
+    )
     monkeypatch.setattr(task_management_service, "get_task", lambda _: deepcopy(task))
 
     def update(task_id, *, fields, actor_concept_id):
@@ -84,8 +91,13 @@ def test_ordinary_continuation_links_product_and_delegates_without_duplicate_wri
         first = handlers._task_update_fields(**payload)
         repeat = handlers._task_update_fields(**payload)
     assert first["effect_status"] == "succeeded"
-    assert first["canonical_read_back"]["task"]["current_work_product"]["status"] == "ready"
-    assert first["canonical_read_back"]["task"]["assignee_concept_id"] == "#V#von_system"
+    assert (
+        first["canonical_read_back"]["task"]["current_work_product"]["status"]
+        == "ready"
+    )
+    assert (
+        first["canonical_read_back"]["task"]["assignee_concept_id"] == "#V#von_system"
+    )
     receipt = _canonical_effect_readback_receipt(first)
     assert receipt["verified"] is True
     material, verified = _canonically_verified_material_effect_ids(
@@ -213,3 +225,43 @@ def test_continuation_schema_exposes_product_and_hides_authority_switch():
     assert "actor_bound_continuation" not in schema["properties"]
     assert "actor_concept_id" not in schema["properties"]
     assert "current_work_product_concept_id" in definition.description
+
+
+@pytest.mark.parametrize(
+    "types,member_ids,created_by,expected",
+    [
+        (["#V#coding_agent"], {"#V#alice", "#V#worker"}, "#V#alice", True),
+        (["#V#coding_agent"], {"#V#alice"}, "#V#alice", False),
+        (["#V#coding_agent"], {"#V#worker"}, "#V#alice", False),
+        (["#V#person"], {"#V#alice", "#V#worker"}, "#V#alice", False),
+        (["#V#coding_agent"], {"#V#alice", "#V#worker"}, "#V#bob", False),
+    ],
+)
+def test_continuation_uses_live_coding_agent_membership(
+    monkeypatch, task_state, types, member_ids, created_by, expected
+):
+    task, writes = task_state
+    task["created_by_concept_id"] = created_by
+    monkeypatch.setattr(
+        concept_service,
+        "get_concept_by_concept_id_exact",
+        lambda concept_id: {"relationships": {"is_an_instance_of": types}},
+    )
+    monkeypatch.setattr(
+        organisation_membership_service,
+        "is_user_member_of_organisation",
+        lambda user, org: user in member_ids and org == "#V#lab",
+    )
+    with override_current_actor("#V#alice", "#V#lab"):
+        result = handlers._task_update_fields(
+            **_payload(assignee_concept_id="#V#worker", report_to_concept_id="#V#alice")
+        )
+    assert bool(result.get("success")) is expected
+    if expected:
+        assert result["canonical_read_back"]["verified"] is True
+        assert task["assignee_concept_id"] == "#V#worker"
+        assert task["created_by_concept_id"] == "#V#alice"
+        assert len(writes) == 1
+    else:
+        assert result["error_code"] == "task_assignment_scope_denied"
+        assert not writes