fix(chat): submit explicitly bound inbox work to governed execution

docs/architecture/rfcs/loopx-overall-roadmap-v0.md

@@ -797,6 +797,16 @@ deduplication and return. App settings select and read back the trigger per
 connection. External host-tool permission and sender-bound delegation remain
 separate gaps; receiving a request does not establish execution authority.
 
+An external steward can now select an operator-granted existing task binding
+from its exact source catalog and submit the handoff through the same governed
+delegation/Turn owner. Durable context receipt, bounded launch, receiver
+adoption, task acceptance and original-audience return remain distinct. No
+identity, task, runtime policy or scheduler is provisioned by the handoff. Local
+file/SQLite fixtures qualify dispatch, independent acceptance and return;
+provider/model routing, new-task allocation, operator UI discovery and complete
+stop/result presentation remain separate R3 gates. See
+[bound inbox execution](../../../loopx/capabilities/manager_context/README.md#executing-already-bound-work).
+
 Resolve source context before routing: a short reply retains the exact
 same-conversation parent, with missing and truncated material explicit. TS owns the bounded context projection;
 Lark transport and the private inbox preserve provider ancestry without granting

loopx/capabilities/manager_context/README.md

@@ -23,6 +23,64 @@ Keep the policy private (0600); do not commit it. A read grant without a sender
 grant is insufficient. This does not grant remote delivery, evidence reads,
 worker launch, Todo/lease changes or protected operations.
 
+### Executing already bound work
+
+An operator may separately authorize an external audience to select existing
+governed task bindings. Add an `execution_bindings` list to that exact private
+`sources[channel]` row (retain its independently verified `sender_ids`):
+
+```json
+{"execution_bindings":[{"goal_id":"research","agent_id":"worker",
+  "requester_agent_id":"lead","binding_id":"review"}]}
+```
+
+The Goal must already point at its operator-owned `.loopx/config/` binding file
+through `loopx configure-goal --goal-id research --subagent-execution-config
+.loopx/config/delegations.json --execute`. Provision its real task, validation,
+registered requester, workspace and host using the existing
+[local delegation interface](../../../docs/reference/local-delegation.md).
+Neither registration, a sender grant, read access nor a context brief creates
+this execution grant. No task, role, profile or schedule is provisioned here.
+
+Chat exposes only the authorized Goal/Agent/binding/Todo identities in
+`context_execution.bindings`. For requested work covered by that current task,
+the model may select `execution_binding_id` in its semantic `context_handoff`.
+Assessment-only handoffs omit it. The host rechecks provider provenance, source
+revocation, active membership, Goal configuration, the original Turn and
+canonical preflight, then uses `Delegations.start`. Commands, roots, requester
+identity and host policy cannot come from the model. An unprobed runtime retains
+`runtime_unverified`; it may attempt the existing bounded execution, and is never
+reported as ready, running or complete. Known unavailability or refused task
+admission does not launch. Retry preserves `context-<original-receipt-id>` and
+does not resume, reset or replace a stopped/completed operation.
+
+The original brief and its field/encoded-byte limits remain unchanged. The trusted
+host binds the original inbox request in the existing operation identity and
+receiver bootstrap instructions, outside user-authored fields. Replay preserves
+that binding; it cannot substitute another request.
+
+The receiver independently reads/adopts the original inbox request and returns
+an audience-safe conclusion with the existing `manager-inbox report` path, in
+addition to its peer result. Launch submission, receiver conclusion, canonical
+acceptance and provider delivery remain separate evidence. The existing return
+service replies to the original conversation; it does not start another model
+thread. Plain inbox delivery now says that execution has not started.
+The native postcondition entry retires only its exact operation-owned temporary
+host input before checking a clean delivery worktree; unrelated files and actual
+artifact changes still fail canonical validation. Validation recovery resumes the
+original Turn after its retained host result, without invoking the model again.
+An external conversation is not a native Goal wake owner. The typed wake owner
+settles that separate intent as `no_wake_owner`; the exact original inbox return
+still carries the receiver's conclusion, without a hidden Goal or retry loop.
+
+Remove that source's exact execution grant to prevent later launches/replays.
+Already launched work keeps its original lifecycle: inspect and stop its exact
+operation with `loopx delegation`, rather than assuming `/stop` of the manager
+also stops an independently governed worker. Existing private configuration and
+journals remain local. Automatic allocation of new tasks, operator UI discovery,
+general inbox activation, cross-host readiness and live result/stop presentation
+are still separate product work; this path only selects already configured work.
+
 The existing local operator commands preview, apply and verify exceptions:
 
 ```sh
@@ -63,6 +121,8 @@ retain the original Session, request and message. An App's portfolio read grant
 alone does not grant context delivery, and inbox delivery does not launch a worker.
 Provenance uses the controller's resolved coordination runtime root when present,
 so a separate Chat store cannot split it from the recipient policy and inbox.
+The exact execution catalog and handoff dispatch use that same resolved root;
+private Session/Turn files remain in the existing Chat store.
 Controllers predating that root retain the existing Chat-parent layout.
 Messages outside the shared inbox's source-context bounds remain intact in
 ordinary Chat, but cannot be handed off as inline context; use a scoped artifact

loopx/capabilities/manager_context/execution.py

@@ -0,0 +1,144 @@
+"""Chat selection over explicit operator bindings and the existing Turn owner.
+
+This adapter provisions no work, identity, grants, host profile or scheduler.
+Context receipt, launch observation, receiver adoption and return stay separate.
+"""
+
+from pathlib import Path
+from collections.abc import Callable
+from typing import Any
+
+from ...agent_registry import load_goal_from_registry
+from ...collaboration_mcp import Delegations
+from ...control_plane.collaboration import conversation_scope
+from ...control_plane.collaboration.source_grant_observation import (
+    source_execution_bindings,
+)
+from ...control_plane.effect_runtime import EffectRuntimeRejected
+from ...orchestration import compact_orchestration_policy, normalize_subagent_execution_config
+
+
+def _grants(root: Path, registry: Path, session: dict[str, Any], turn: dict[str, Any]) -> list[dict[str, Any]]:
+    if conversation_scope(session, origin=turn.get("origin", "unknown"))["kind"] != "external_audience":
+        return []
+    result = source_execution_bindings(root, registry, session, turn)
+    return list(result["bindings"])
+
+
+def _service(root: Path, registry: Path, grant: dict[str, Any]) -> tuple[Delegations, dict[str, Any]]:
+    goal = load_goal_from_registry(registry, grant["goal_id"])
+    if not goal:
+        raise ValueError("execution Goal unavailable")
+    configured = compact_orchestration_policy(goal.get("spawn_policy")).get("execution_config")
+    if not configured:
+        raise ValueError("Goal execution configuration unavailable")
+    relative = Path(normalize_subagent_execution_config(configured))
+    workspace = Path(goal["repo"]).resolve()
+    config_root = workspace / ".loopx" / "config"
+    config = (workspace / relative).resolve()
+    if not config.is_relative_to(config_root) or not config.is_file() or config.stat().st_size > 1_000_000:
+        raise ValueError("Goal execution configuration unavailable")
+    service = Delegations(root, registry, grant["goal_id"], grant["requester_agent_id"], config)
+    binding = service.binding(grant["binding_id"], require_active=True)
+    if binding["agent_id"] != grant["agent_id"]:
+        raise ValueError("execution binding recipient changed")
+    return service, binding
+
+
+def catalog(root: Path, registry: Path, session: dict[str, Any], turn: dict[str, Any]) -> dict[str, Any]:
+    """Expose only exact task choices, never commands, paths or host credentials."""
+    try:
+        rows = []
+        for grant in _grants(root, registry, session, turn):
+            _, binding = _service(root, registry, grant)
+            rows.append({key: grant[key] for key in ("goal_id", "agent_id", "binding_id")}
+                        | {"todo_id": binding["todo_id"]})
+        return {"available": True, "bindings": rows}
+    except (OSError, ValueError, KeyError, TypeError, EffectRuntimeRejected):
+        return {"available": False, "bindings": [], "reason": "execution_bindings_unavailable"}
+
+
+def dispatch(root: Path, registry: Path, *, session: dict[str, Any], turn: dict[str, Any],
+             request: dict[str, Any], receipt: dict[str, Any],
+             execution_allowed: Callable[[], bool]) -> dict[str, Any]:
+    """Submit this delivered brief to exactly one separately granted binding.
+
+    Stable operation identity uses the original inbox receipt. No retry invents
+    another operation, and no completed/stopped binding is reset or replaced.
+    The independent receiver must adopt and report through the existing inbox.
+    """
+    selected = request.get("execution_binding_id")
+    if selected is None:
+        return {"submitted": False}
+    try:
+        if not execution_allowed():
+            raise ValueError("source Turn is no longer active")
+        target = {key: request[key] for key in ("goal_id", "agent_id")}
+        if receipt.get("status") != "delivered" or any(receipt.get(key) != value for key, value in target.items()):
+            raise ValueError("original handoff receipt mismatch")
+        grant = next((row for row in _grants(root, registry, session, turn)
+                      if all(row[key] == value for key, value in target.items())
+                      and row["binding_id"] == selected), None)
+        if grant is None:
+            raise ValueError("execution binding not granted to this source")
+        service, binding = _service(root, registry, grant)
+        operation = "context-" + receipt["request_id"]
+        # Recover an existing operation instead of probing a now-completed Todo
+        # and misreporting an accepted result as a refused new launch.
+        if service.path(operation).exists():
+            row = service.read(operation)
+            return {"submitted": True, "operation_id": operation, "todo_id": binding["todo_id"],
+                    "status": row["status"], "replayed": True}
+        preflight = service.inspect(selected)
+        # Match the existing start owner: an unprobed configured runtime may
+        # attempt bounded execution, but is never presented as ready/running.
+        # Known refusal, missing canonical acceptance or an unavailable runtime
+        # must stop before dispatch. Actual launch and acceptance remain owned
+        # by the governed Turn, not this point-in-time preview.
+        if (preflight["state"] not in {"launchable", "runtime_unverified"}
+                or not all(preflight.get(key) is True for key in
+                           ("turn_eligible", "acceptance_ready", "authority_ready"))):
+            return {"submitted": False, "preflight": preflight, "reason": "execution_not_launchable"}
+        # Re-read the source and operator selection after the potentially slow
+        # preview. Delegations.start itself rechecks the exact binding and Turn.
+        if not execution_allowed() or grant not in _grants(root, registry, session, turn):
+            raise ValueError("source execution grant changed before launch")
+        result = service.start(selected, operation, request["brief"],
+                               conversation={"session_id": session["session_id"], "turn_id": turn["turn_id"]},
+                               source_request_id=receipt["request_id"])
+        return {"submitted": True, "operation_id": operation, "todo_id": binding["todo_id"],
+                "status": result["status"], "runtime_readiness": preflight["state"], "replayed": False}
+    except (OSError, ValueError, KeyError, TypeError, EffectRuntimeRejected):
+        return {"submitted": False, "reason": "execution_binding_or_admission_unavailable"}
+
+
+def handoff_message(receipt: dict[str, Any], execution: dict[str, Any]) -> str:
+    prefix = "已将原消息交给 " + str(receipt["agent_id"]) + "。"
+    if execution.get("submitted"):
+        return prefix + "已提交受控执行；受理不代表完成，接收方的处理结论将回到本次对话。"
+    if execution.get("reason"):
+        return prefix + "交接已保存，但执行绑定或任务准入未通过，尚未启动执行。"
+    return prefix + "材料已进入收件箱，尚未启动执行；收到接收方的处理结论后会回到这里。"
+
+
+def handoff_response(root: Path, registry: Path, *, session: dict[str, Any],
+                     turn: dict[str, Any], response: dict[str, Any],
+                     source_authorized: Callable[[], bool],
+                     execution_allowed: Callable[[], bool]) -> dict[str, Any]:
+    """Present context delivery and separately admitted execution on one path."""
+    from . import deliver
+
+    try:
+        if not source_authorized():
+            raise ValueError("manager connection authority is no longer available")
+        receipt = deliver(root, registry, session=session, turn=turn,
+                          request=response["context_handoff"])
+        execution = dispatch(root, registry, session=session, turn=turn,
+                             request=response["context_handoff"], receipt=receipt,
+                             execution_allowed=execution_allowed)
+        return {**response, "proposals": [], "gate": None,
+                "context_handoff_receipt": receipt, "context_execution": execution,
+                "message": handoff_message(receipt, execution)}
+    except (OSError, ValueError):
+        return {**response, "proposals": [], "gate": None,
+                "message": "材料尚未转交：目标绑定、来源授权或持久收件回读未通过。需要修复交接链路；没有改动任务或优先级。"}

loopx/capabilities/manager_context/inspection.py

@@ -164,6 +164,7 @@ def rejected_read_arguments(arguments: dict[str, Any]) -> list[str]:
 def manager_index(context: dict[str, Any]) -> dict[str, Any]:
     """A small directory, never a second mutable progress store."""
     read_tool = CONTEXT_TOOL_NAME if context.get("scope") == "owner_goal" else TOOL_NAME
+    execution = context.get("context_execution")
     return {
         "schema_version": "manager_evidence_index_v1",
         "snapshot_id": context.get("snapshot_id"),
@@ -191,6 +192,8 @@ def manager_index(context: dict[str, Any]) -> dict[str, Any]:
             if row.get("activation_state") != "stopped"
         ],
         "context_delegation": context.get("context_delegation"),
+        **({"context_execution": execution} if isinstance(execution, dict)
+           and (execution.get("bindings") or execution.get("available") is False) else {}),
         "evidence_sources": context.get("evidence_sources", [])[:12],
         "evidence_source_count": len(context.get("evidence_sources", [])),
         "agent_discovery": {"tool": read_tool, "view": "agents", "scope": "permitted_registry",

loopx/chat_agent.py

@@ -327,6 +327,16 @@ def _turn_prompt(
     runtime_profile: str = "restricted",
     project_work: bool = False,
 ) -> str:
+    try:
+        supplied = json.loads(context_summary)
+        choices = supplied.get("context_execution") if isinstance(supplied, dict) else None
+    except (ValueError, TypeError):
+        choices = None
+    execution_guidance = (
+        "When context_execution.bindings supplies an exact existing Todo binding for the requested work, read that Todo and select its binding_id as context_handoff.execution_binding_id to submit governed execution. "
+        "Only select an explicitly cataloged binding that covers this request; registration and context delivery do not authorize execution. For consultation or unrelated/missing task bindings omit execution_binding_id. Never create a hidden Todo, change host settings or reuse a completed/stopped task to obtain launch. "
+        if isinstance(choices, dict) and choices.get("bindings") else ""
+    )
     envelope = {
         "schema_version": CHAT_AGENT_RESPONSE_SCHEMA_VERSION,
         "message": "Complete answer for the operator, at the depth this task needs.",
@@ -404,6 +414,8 @@ def _turn_prompt(
         "A continuation, correction or status question belongs to the established Goal/owner. Preserve its constraints; do not restart, create a duplicate Goal or ask for permission already granted. "
         "For requested work, inspect the supplied Goal directory and relevant work/Agent evidence (using the declared read tool when incomplete). An empty delivery-grant list does not prove there is no existing work. "
         "Use context_handoff for a uniquely relevant, active and currently granted existing owner when the user asks for that work, even without the word delegate. "
+        + execution_guidance
+        +
         "A correction to requested work is authorized context for its existing owner: send the corrected constraints in context_handoff, proposals=[], without asking to approve a Todo edit. Only direct control-plane record/configuration edits use that separate preview path. "
         "Do not redirect a Goal Chat back to its own owner: handle its follow-up in the current conversation. Registration alone is not delivery authority or execution readiness. "
         "Compare ALL plausible existing work items before selecting. A Goal ID, row order, or word overlap is not evidence of user intent. If two active items cover the requested subject and history does not distinguish them, context_handoff MUST be null; ask which in message, with goal_draft=null. "

loopx/chat_coordination.py

@@ -108,6 +108,13 @@ def prepare_turn_context(controller, adapter, session, turn_id, event_sink, *, s
         runtime_root, controller.registry_path, session,
         controller.store.load_turn(session_id, turn_id) or {},
     )
+    from .capabilities.manager_context.execution import catalog
+    execution_catalog = catalog(
+        runtime_root, controller.registry_path, session,
+        controller.store.load_turn(session_id, turn_id) or {},
+    )
+    if execution_catalog["bindings"] or not execution_catalog["available"]:
+        context["context_execution"] = execution_catalog
     if isinstance(adapter, CodexAppServerAdapter):
         from .capabilities.manager_context.inspection import ManagerInspection, manager_index
         from .chat_manager_context import manager_authorization_scope_id