fix(chat): continue private bound Sessions through local source admission

docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md

@@ -730,6 +730,22 @@ packaged UI with synthetic fixtures and qualify the chosen real query separately
 | Execution and control | Existing managed Turn, attached-session/host binding, Chat steering/interrupt | Qualify the exact supported profile. Neither registered nor inbox-acknowledged means running; native steering, next-Turn queue and unsupported must stay distinct |
 | Result | Answer-report, artifact/revision, review/adoption and return owners | Read the stored version; preserve source and independent review. A failed report read retries reading, not a new model run |
 
+The local-input repair uses the existing
+`POST /api/chat/sessions/{session_id}/turns` with `message` and a stable
+`client_turn_id`; `delivery: "queue"` remains optional. For an existing private
+bound-owner Session, a normal send now enters the same queue after fresh
+provider/operator, saved-binding and workspace checks. It retains `origin=web`,
+the original Session/thread and the binding's narrow project or steward audience.
+It does not become a provider event or an unrestricted local portfolio. Group
+input and direct steering retain their source guards. The body cannot supply
+origin, owner or provider claims. The TS binding/source owners decide admission
+and provenance; Python observes identity and commits under the binding fence.
+Steward ingress uses the registered coordination root and existing source policy,
+separately from execution grants. File/SQLite HTTP regressions cover full text,
+replay, FIFO and revocation races. This is a Core/API prerequisite: packaged
+frontend discovery, live shared-Lark adoption, project-recipient collaboration
+and original-conversation result return still require their existing acceptance.
+
 Lark's readable-answer exit also checks rendered emphasis, not only a matching
 Markdown source receipt. Following the [CommonMark delimiter rules](https://spec.commonmark.org/0.31.2/#emphasis-and-strong-emphasis),
 the shared inbox reply adapter normalizes a paired

docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md

@@ -470,6 +470,18 @@ CLI/SSH 导出使用 `goal-portfolio --manager-view todos --goal-id GOAL --todo-
 | 执行与控制 | 现有 managed Turn、attached-session/host binding、Chat steering/interrupt | 验收精确受支持的 profile。已注册或 inbox acknowledged 都不等于正在运行；原生 steering、next-Turn queue 和 unsupported 必须分别表达 |
 | 结果 | Answer-report、artifact/revision、review/adoption 与 return owners | 读取已存版本，保留来源与独立审阅。report 读取失败重试读取，不重新运行模型 |
 
+本机输入修复复用现有 `POST /api/chat/sessions/{session_id}/turns`，传入
+`message` 与稳定的 `client_turn_id`；`delivery: "queue"` 仍可省略。
+对已经存在的本人私聊绑定会话，普通发送在重新核对 provider/本人、原 binding
+与工作区后进入同一队列，保留 `origin=web`、原 Session/thread 和原项目或管家受众，
+不伪装 provider 事件，也不扩大成本机全局 portfolio。群输入与直接 steering
+继续受来源守卫约束；请求体不能声明 origin、本人或 provider 身份。
+TS binding/source owner 决定准入和来源关联，Python 观测身份并在 binding 发布锁内落队列。
+管家 ingress 沿注册的 coordination root 和现有 source policy，执行授权仍独立。
+真实 File/SQLite HTTP 回归覆盖完整文本、重放、FIFO 和撤权竞态。
+此项是 Core/API 前置：打包前端发现、真实共享飞书会话采用、项目接收者协作与原会话
+结果返回仍须完成各自现有验收，不以合成测试宣称整条用户旅程完成。
+
 Lark 可读回复的出口还要检查实际强调呈现，不能把 Markdown 原文读回一致当成视觉验收。
 依照 [CommonMark 分隔符规则](https://spec.commonmark.org/0.31.2/#emphasis-and-strong-emphasis)，
 共享 inbox 回复适配器对“加粗内的末尾标点紧接下一单词”做格式规范化：

loopx/capabilities/manager_context/__init__.py

@@ -59,8 +59,13 @@ def register_ingress(
     message: str,
     source_id: str,
     source_message: str | None = None,
+    origin: str | None = None,
 ) -> None:
-    """Provider-only provenance, persisted before enqueue (never model-authored)."""
+    """Host-verified provenance, persisted before enqueue (never model-authored).
+
+    Existing provider receipts retain their exact shape. Local bound-owner input
+    records its actual web origin rather than impersonating a provider event.
+    """
     if not sender_id or not source_id or not channel.startswith("manager.external."):
         raise ValueError("manager ingress requires exact provider provenance")
     value = dict(
@@ -72,6 +77,10 @@ def register_ingress(
         source_id=source_id,
         source_message=normalize_source_context(message if source_message is None else source_message),
     )
+    if origin is not None:
+        if origin != "web":
+            raise ValueError("unsupported local ingress origin")
+        value["origin"] = origin
     path = (
         _root(runtime_root)
         / "ingress"

loopx/capabilities/native_chat/conversation_bindings.py

@@ -7,6 +7,7 @@
 from __future__ import annotations
 
 from collections.abc import Callable
+from contextlib import contextmanager
 from pathlib import Path
 from typing import Any
 import uuid
@@ -178,7 +179,8 @@ def _stop_revoked_binding(self, binding_id: str) -> None:
     def resolve(self, *, binding_id: str, source_ref: str, sender_ref: str,
                 private_human_message: bool, session_context: dict[str, Any] | None = None,
                 group_ref: str | None = None, topic_ref: str | None = None,
-                group_human_message: bool = False, include_observation: bool = False) -> dict[str, Any]:
+                group_human_message: bool = False, include_observation: bool = False,
+                origin: str | None = None) -> dict[str, Any]:
         current = self.read()
         row = next((item for item in current.get("bindings", []) if item.get("binding_id") == binding_id), None)
         if row is None:
@@ -191,14 +193,9 @@ def resolve(self, *, binding_id: str, source_ref: str, sender_ref: str,
         if next((item for item in current.get("bindings", [])
                  if item.get("binding_id") == binding_id), None) != row:
             raise ValueError("conversation binding changed during identity verification")
-        registry_scope = {}
-        if row.get("goal_scope") == "all_registered":
-            registry_scope["available_goal_ids"] = self.goal_scope_ids(row)
-        selected = self._core("collaboration.conversation.bound_context", {
-            "current": current, "binding_id": binding_id, "source_ref": source_ref, "sender_ref": sender_ref,
-            "private_human_message": private_human_message, "observation": observation,
-            "available_projects": self.projects.available(),
-            **registry_scope,
+        selected = self._resolve_observed(current, row, observation, {
+            "source_ref": source_ref, "sender_ref": sender_ref, "private_human_message": private_human_message,
+            **({"origin": origin} if origin is not None else {}),
             **({"session_context": session_context} if session_context is not None else {}),
             **({"group_ref": group_ref, "topic_ref": topic_ref,
                 "group_human_message": group_human_message} if group_ref is not None or topic_ref is not None or group_human_message else {}),
@@ -208,6 +205,30 @@ def resolve(self, *, binding_id: str, source_ref: str, sender_ref: str,
         # or a substitute for the next admission/delivery owner's fresh check.
         return {**selected, "observation": observation} if include_observation else selected
 
+    def _resolve_observed(self, current, row, observation, source):
+        return self._core("collaboration.conversation.bound_context", {
+            "current": current, "binding_id": row["binding_id"], "observation": observation,
+            "available_projects": self.projects.available(), **source,
+            **({"available_goal_ids": self.goal_scope_ids(row)} if row.get("goal_scope") == "all_registered" else {}),
+        })
+
+    @contextmanager
+    def local_input(self, saved: dict[str, Any]):
+        """Observe outside the publication fence, then retain it through enqueue.
+
+        This is the existing loopback operator's continuation of a private
+        binding. Saved fields select an identity; only fresh Core checks grant it.
+        """
+        source = {"source_ref": saved["source_ref"], "sender_ref": saved["operator_ref"],
+                  "private_human_message": False, "origin": "web", "session_context": saved}
+        observed = self.resolve(binding_id=saved["binding_id"], include_observation=True, **source)
+        with exclusive_file_lock(self.path, operation="admit_local_bound_input"):
+            current = self.read()
+            if next((row for row in current.get("bindings", [])
+                     if row.get("binding_id") == saved["binding_id"]), None) != observed["binding"]:
+                raise ValueError("conversation binding changed before input admission")
+            yield self._resolve_observed(current, observed["binding"], observed["observation"], source)
+
     def goal_scope_ids(self, binding: dict[str, Any]) -> list[str]:
         if binding.get("goal_scope") != "all_registered":
             return list(binding.get("goal_ids", []))

loopx/capabilities/native_chat/project_context.py

@@ -75,11 +75,7 @@ def session_context(self, session: dict[str, Any]) -> dict[str, Any]:
             if self.conversation_bindings is None:
                 raise ValueError("bound steward authority is unavailable")
             selected = self.conversation_bindings.session_context(steward)
-            if session.get("goal_id") != "loopx-manager" or session.get("channel_id") != selected["channel_id"]:
-                raise ValueError("steward audience mismatch")
-            from ...chat_manager import MANAGER_AGENT_OBJECTIVE
-            return {"project": Path(selected["context"]["workspace_path"]), "objective": MANAGER_AGENT_OBJECTIVE,
-                    "title": "Steward"}
+            return self.runtime_context(session, selected)
         saved = session.get("project_context")
         if not isinstance(saved, dict) or session.get("goal_id") is not None:
             raise ValueError("invalid ordinary project Session")
@@ -89,7 +85,18 @@ def session_context(self, session: dict[str, Any]) -> dict[str, Any]:
             selected = self.conversation_bindings.session_context(saved)
         else:
             selected = self.resolve(str(saved.get("project_ref") or ""), session_context=saved)
-        if session.get("channel_id") != selected["channel_id"]:
+        return self.runtime_context(session, selected)
+
+    @staticmethod
+    def runtime_context(session: dict[str, Any], selected: dict[str, Any]) -> dict[str, Any]:
+        """Adapt an already verified Core selection; never repeat provider IO under its fence."""
+        if session.get("steward_context") is not None:
+            if session.get("goal_id") != "loopx-manager" or session.get("channel_id") != selected["channel_id"]:
+                raise ValueError("steward audience mismatch")
+            from ...chat_manager import MANAGER_AGENT_OBJECTIVE
+            return {"project": Path(selected["context"]["workspace_path"]), "objective": MANAGER_AGENT_OBJECTIVE,
+                    "title": "Steward"}
+        if session.get("goal_id") is not None or session.get("channel_id") != selected["channel_id"]:
             raise ValueError("project conversation channel mismatch")
         objective = PROJECT_WORK_OBJECTIVE if selected["context"]["grant"] == "workspace_write" else PROJECT_CONVERSATION_OBJECTIVE
         return {"project": Path(selected["context"]["workspace_path"]),

loopx/capabilities/native_chat/queued_admission.py

@@ -0,0 +1,105 @@
+"""IO admission to the existing same-Session queue; Core owns source authority."""
+from pathlib import Path
+from typing import Any
+
+from ...attached_session import enqueue_attached_agent_turn
+from ...chat_store import CHAT_SESSION_MODE_ATTACHED
+from ...control_plane.collaboration import conversation_scope
+from . import project_context as project_context_policy
+
+
+def enqueue_turn(controller, *, session_id: str, client_turn_id: str, message: str,
+                 attachments: list[dict[str, Any]] | None, work_dir: Path | None, objective: str,
+                 origin: str, external_agent_target: dict | None):
+    session = controller.store.load_session(session_id)
+    if session is None or session.get("status") == "closed":
+        raise KeyError("chat session was not found")
+    bound = session.get("steward_context") or session.get("project_context") or {}
+    if origin == "web" and bound.get("audience") in {"bound_owner", "bound_group"}:
+        if external_agent_target is not None:
+            raise ValueError("Agent target requires its original attached Session")
+        return _enqueue_local_bound(controller, session, client_turn_id, message, attachments)
+    if work_dir is None:
+        raise ValueError("the conversation's runtime context is required")
+    if session.get("steward_context") is not None:
+        if origin not in {"lark", "external"}:
+            raise ValueError("bound steward Chat requires its external source admission")
+        controller.project_contexts.session_context(session)
+    if session.get("project_context") is not None:
+        if conversation_scope(session, origin=origin)["kind"] != "project_workspace":
+            raise ValueError("project grant does not authorize this external audience or source")
+        context = controller.project_contexts.session_context(session)
+        project_context_policy.validate_project_executor_scope(str(session["agent_id"]), session["project_context"], controller.capabilities)
+        work_dir, objective = context["project"], context["objective"]
+    if session.get("session_mode") == CHAT_SESSION_MODE_ATTACHED:
+        if attachments:
+            raise ValueError("attached host session queue does not yet accept attachments")
+        turn, created = enqueue_attached_agent_turn(
+            store=controller.store,
+            registry_path=controller.registry_path,
+            session_id=session_id,
+            client_turn_id=client_turn_id,
+            message=message,
+            origin=origin,
+            external_agent_target=external_agent_target,
+        )
+    else:
+        if external_agent_target is not None:
+            raise ValueError("Agent target requires its original attached Session")
+        turn, created = controller.store.create_queued_turn(
+            session_id,
+            client_turn_id=client_turn_id,
+            message=message,
+            attachments=attachments,
+            origin=origin,
+        )
+        controller.resume_session_queue(
+            session_id=session_id,
+            work_dir=work_dir,
+            objective=objective,
+        )
+    return turn, created
+
+
+def _enqueue_local_bound(controller, session, client_turn_id, message, attachments):
+    contexts = controller.project_contexts
+    bindings = contexts.conversation_bindings
+    if bindings is None:
+        raise ValueError("bound conversation authority is unavailable")
+    saved = session.get("steward_context") or session["project_context"]
+    with bindings.local_input(saved) as selected:
+        fresh = controller.store.load_session(session["session_id"])
+        if fresh is None or fresh.get("status") == "closed":
+            raise KeyError("chat session was not found")
+        for key in ("project_context", "steward_context", "channel_id", "goal_id", "agent_id"):
+            if fresh.get(key) != session.get(key):
+                raise ValueError("bound Session changed before input admission")
+        context = contexts.runtime_context(fresh, selected)
+        if fresh.get("project_context") is not None:
+            project_context_policy.validate_project_executor_scope(str(fresh["agent_id"]),
+                fresh["project_context"], controller.capabilities)
+        else:
+            bindings.ensure_delivery_scope(selected)
+            _record_local_ingress(controller.coordination_runtime_root, fresh, client_turn_id, message)
+        turn, created = controller.store.create_queued_turn(session["session_id"],
+            client_turn_id=client_turn_id, message=message, attachments=attachments, origin="web")
+    # Dispatch after releasing the publication fence. The worker independently
+    # rechecks current scope before effects, including revocation after enqueue.
+    controller.resume_session_queue(session_id=session["session_id"],
+        work_dir=context["project"], objective=context["objective"])
+    return turn, created
+
+
+def _record_local_ingress(runtime_root, session, client_turn_id, message):
+    from ..manager_context import register_ingress
+    from ...control_plane.collaboration.inbox import _hash, normalize_source_context
+
+    try:
+        normalize_source_context(message)
+    except ValueError:
+        # Chat keeps the full input. Oversized Inbox context is not an implicit
+        # truncation or handoff grant, matching the existing provider boundary.
+        return
+    register_ingress(runtime_root, session_id=session["session_id"], client_turn_id=client_turn_id,
+        channel=session["channel_id"], sender_id=session["steward_context"]["operator_ref"],
+        message=message, source_id="web:" + _hash([session["session_id"], client_turn_id]), origin="web")