diff --git a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md index 9a3776ad9d..2971947c31 100644 --- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md +++ b/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 diff --git a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md index 6f200219e6..79004c0794 100644 --- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md +++ b/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 回复适配器对“加粗内的末尾标点紧接下一单词”做格式规范化： diff --git a/loopx/capabilities/manager_context/__init__.py b/loopx/capabilities/manager_context/__init__.py index 9e394ab620..d0124ff332 100644 --- a/loopx/capabilities/manager_context/__init__.py +++ b/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" diff --git a/loopx/capabilities/native_chat/conversation_bindings.py b/loopx/capabilities/native_chat/conversation_bindings.py index 7845361562..76760e0536 100644 --- a/loopx/capabilities/native_chat/conversation_bindings.py +++ b/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", [])) diff --git a/loopx/capabilities/native_chat/project_context.py b/loopx/capabilities/native_chat/project_context.py index fa44f78f61..06c17f9afa 100644 --- a/loopx/capabilities/native_chat/project_context.py +++ b/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"]), diff --git a/loopx/capabilities/native_chat/queued_admission.py b/loopx/capabilities/native_chat/queued_admission.py new file mode 100644 index 0000000000..a6dfcb5693 --- /dev/null +++ b/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") diff --git a/loopx/chat_runtime.py b/loopx/chat_runtime.py index 354fe7fa98..937306c717 100644 --- a/loopx/chat_runtime.py +++ b/loopx/chat_runtime.py @@ -1347,54 +1347,17 @@ def enqueue_turn( client_turn_id: str, message: str, attachments: list[AttachmentPayload] | None = None, - work_dir: Path, + work_dir: Path | None, objective: str, origin: str = "external", external_agent_target: dict[str, Any] | None = None, ) -> tuple[dict[str, Any], bool]: """Persist a bounded same-Session Turn and dispatch it in FIFO order.""" - session = self.store.load_session(session_id) - if session is None or session.get("status") == "closed": - raise KeyError("chat session was not found") - if session.get("steward_context") is not None: - if origin not in {"lark", "external"}: - raise ValueError("bound steward Chat requires its external source admission") - self.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 = self.project_contexts.session_context(session) - project_context_policy.validate_project_executor_scope(str(session["agent_id"]), session["project_context"], self.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=self.store, - registry_path=self.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 = self.store.create_queued_turn( - session_id, - client_turn_id=client_turn_id, - message=message, - attachments=attachments, - origin=origin, - ) - self.resume_session_queue( - session_id=session_id, - work_dir=work_dir, - objective=objective, - ) - return turn, created + from .capabilities.native_chat.queued_admission import enqueue_turn + return enqueue_turn(self, session_id=session_id, client_turn_id=client_turn_id, + message=message, attachments=attachments, work_dir=work_dir, objective=objective, + origin=origin, external_agent_target=external_agent_target) def resume_session_queue( self, diff --git a/loopx/chat_server.py b/loopx/chat_server.py index 02432b3763..77a5ff98d5 100644 --- a/loopx/chat_server.py +++ b/loopx/chat_server.py @@ -825,21 +825,30 @@ def _session_turn(self, session_id: str) -> None: except (ValueError, TypeError, OverflowError) as exc: self._send_error(str(exc), turn_replay_safe=True, delivery_state="not_delivered") return + bound = session.get("steward_context") or session.get("project_context") or {} + bound_input = bound.get("audience") in {"bound_owner", "bound_group"} try: - context = self._session_context(session) - runtime_objective = str(context["objective"] or context["title"]) + # Bound local input resolves its context in the source admission + # owner, once, outside the grant fence. Do not repeat provider IO + # here or pass an unverified saved workspace as caller authority. + context = None if bound_input else self._session_context(session) + runtime_objective = str(context["objective"] or context["title"]) if context else "" except (ValueError, KeyError) as exc: # No Turn has been admitted when its owning context cannot resolve. self._send_error(str(exc), turn_replay_safe=True, delivery_state="not_delivered") return try: - if body.get("delivery") == "queue": + # Existing private bound Sessions share their canonical queue after + # fresh local-source admission. The body cannot assert an origin, + # provider principal or a broader local-owner conversation scope. + if body.get("delivery") == "queue" or bound_input: turn, created = self.server.runtime_controller.enqueue_turn( session_id=session_id, client_turn_id=client_turn_id, message=message, attachments=attachments, - work_dir=cast(Path, context["project"]), objective=runtime_objective, origin="web", + work_dir=cast(Path, context["project"]) if context else None, objective=runtime_objective, origin="web", ) else: + assert context is not None turn, created = self.server.runtime_controller.submit_turn( session_id=session_id, client_turn_id=client_turn_id, message=message, attachments=attachments, @@ -909,6 +918,10 @@ def _session_turn(self, session_id: str) -> None: except KeyError: self._send_error("chat session was not found", status=404, session_invalidated=True) return + except ValueError as exc: + self._send_error(str(exc), **({"delivery_state": "not_delivered", "turn_replay_safe": True} + if bound_input else {})) + return except Exception as exc: # noqa: BLE001 - compact local validation error. self._send_error(str(exc)) return diff --git a/loopx/control_plane/collaboration/conversation_binding.ts b/loopx/control_plane/collaboration/conversation_binding.ts index ae4784d83d..57119e2c79 100644 --- a/loopx/control_plane/collaboration/conversation_binding.ts +++ b/loopx/control_plane/collaboration/conversation_binding.ts @@ -287,6 +287,13 @@ export function resolveBoundConversation(params: JsonObject): JsonObject { observed(row, params.observation); const source = ref(params.source_ref, "source conversation identity"); const group = row.audience === "group"; + // The local operator may continue an existing private Session. This is a + // distinct host-observed origin, never a fabricated provider human event. + const local = params.origin === "web"; + if (params.origin !== undefined && !local) throw new EffectRuntimeRequestError("unsupported bound input origin"); + if (local && (group || params.session_context === undefined || params.private_human_message !== false)) { + throw new EffectRuntimeRequestError("local input requires an existing private bound Session"); + } let topic: JsonObject = {}; if (group) { const groupRef = ref(params.group_ref, "group reference"), topicRef = ref(params.topic_ref, "topic reference"); @@ -296,7 +303,7 @@ export function resolveBoundConversation(params: JsonObject): JsonObject { throw new EffectRuntimeRequestError("message audience is outside the binding grant"); } topic = {group_ref: groupRef, topic_ref: topicRef, filesystem_scope: "workspace_only"}; - } else if (params.sender_ref !== row.operator_ref || params.private_human_message !== true + } else if (params.sender_ref !== row.operator_ref || params.private_human_message !== !local || params.group_ref !== undefined || params.topic_ref !== undefined || params.group_human_message === true) { throw new EffectRuntimeRequestError("message audience is outside the binding grant"); } diff --git a/loopx/control_plane/collaboration/conversation_scope.ts b/loopx/control_plane/collaboration/conversation_scope.ts index 0655be2910..cccf8fae18 100644 --- a/loopx/control_plane/collaboration/conversation_scope.ts +++ b/loopx/control_plane/collaboration/conversation_scope.ts @@ -121,13 +121,14 @@ export function resolveConversationScope(input: Record): Conver && channel === `project.${context.project_ref}`) { return {kind: "project_workspace", goal_ids: [], private_conversation: true}; } - if (context.audience !== "local_owner" && (input.origin === undefined || input.origin === "lark" || input.origin === "external") + if (context.audience !== "local_owner" && (input.origin === undefined || input.origin === "lark" || input.origin === "external" + || input.origin === "web" && context.audience === "bound_owner") && channel === `project.external.${context.binding_id}.${context.source_ref}`) { return {kind: "project_workspace", goal_ids: [], private_conversation: false}; } } catch { /* Incomplete host identity grants no context. */ } } - if (goal === "loopx-manager" && (input.origin === undefined || input.origin === "lark" || input.origin === "external")) { + if (goal === "loopx-manager" && (input.origin === undefined || input.origin === "lark" || input.origin === "external" || input.origin === "web")) { try { const selected = stewardConversationIdentity({context: input.steward_context}); if (channel === selected.channel_id) { diff --git a/loopx/control_plane/collaboration/source_grant_observation.py b/loopx/control_plane/collaboration/source_grant_observation.py index 17f35abe9b..48ed930282 100644 --- a/loopx/control_plane/collaboration/source_grant_observation.py +++ b/loopx/control_plane/collaboration/source_grant_observation.py @@ -19,16 +19,20 @@ def external_source_policy(runtime_root: Path, session: dict[str, Any], turn: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]: - """One provider provenance check for context and explicit execution grants.""" + """Observe host provenance for the shared typed source-policy owner.""" ingress = _read(_root(runtime_root) / "ingress" / (_hash([session["session_id"], turn["client_turn_id"]]) + ".json")) - if (ingress["channel"] != session.get("channel_id") - or ingress["message_digest"] != _hash(turn.get("message")) - or turn.get("origin") not in {"lark", "external"}): - raise ValueError("source mismatch") - if turn.get("origin") == "external" and not conversation_scope( - session, origin="external").get("bound_steward"): - raise ValueError("source mismatch") + try: + effect_runtime_result("collaboration.source.ingress", { + "session": {key: session.get(key) for key in ("session_id", "goal_id", "channel_id", "steward_context")}, + "turn": {"client_turn_id": turn.get("client_turn_id"), "origin": turn.get("origin"), + "message_digest": _hash(turn.get("message"))}, + "ingress": {key: ingress[key] for key in ("session_id", "client_turn_id", "channel", "message_digest", + "sender_id", "source_id", "origin") if key in ingress}, + }) + except EffectRuntimeRejected as exc: + raise ValueError(str(exc)) from exc + policy = _read(_root(runtime_root) / "policy.json") if policy.get("schema_version") != POLICY_SCHEMA: raise ValueError("invalid policy") diff --git a/loopx/control_plane/collaboration/source_grants.ts b/loopx/control_plane/collaboration/source_grants.ts index ee00700c1a..da3e690837 100644 --- a/loopx/control_plane/collaboration/source_grants.ts +++ b/loopx/control_plane/collaboration/source_grants.ts @@ -1,7 +1,7 @@ import type { JsonObject } from "../effect_program.ts"; import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts"; import { requireBoolean, requireJsonObject, requireNonEmptyString, requireStringArray } from "../runtime_decode.ts"; -import { normalizeStewardGoalScope } from "./conversation_scope.ts"; +import { normalizeStewardGoalScope, resolveConversationScope } from "./conversation_scope.ts"; type Recipient = { goal_id: string; agent_id?: string }; type LocalDeliveryScope = "all_registered" | "selected"; @@ -101,6 +101,36 @@ export function resolveSourceRecipients(params: JsonObject): JsonObject { a.goal_id.localeCompare(b.goal_id) || a.agent_id!.localeCompare(b.agent_id!)) }; } +/** Verify the host's immutable ingress against its canonical Session/Turn. + * IO supplies digests, not a second provenance rule or model-authored identity. + * Local continuation retains the binding's narrow audience and source policy. + */ +export function validateSourceIngress(params: JsonObject): JsonObject { + const session = requireJsonObject(params.session, "source Session"); + const turn = requireJsonObject(params.turn, "source Turn"); + const ingress = requireJsonObject(params.ingress, "source ingress"); + const identity: [string, unknown][] = [["session_id", session.session_id], ["client_turn_id", turn.client_turn_id], + ["channel", session.channel_id], ["message_digest", turn.message_digest]]; + for (const [field, expected] of identity) { + if (requireNonEmptyString(ingress[field], `ingress ${field}`) !== + requireNonEmptyString(expected, `source ${field}`)) throw new EffectRuntimeRequestError("source mismatch"); + } + const origin = turn.origin; + const scope = resolveConversationScope({...session, origin}); + if (origin !== "lark" && origin !== "external" && origin !== "web" + || ingress.origin !== undefined && ingress.origin !== origin + || origin === "external" && !scope.bound_steward + || origin === "web" && (!scope.bound_steward || ingress.origin !== "web")) { + throw new EffectRuntimeRequestError("source mismatch"); + } + const sender = requireNonEmptyString(ingress.sender_id, "verified source sender"); + requireNonEmptyString(ingress.source_id, "source identity"); + if (origin === "web" && sender !== requireJsonObject(session.steward_context, "bound steward").operator_ref) { + throw new EffectRuntimeRequestError("source mismatch"); + } + return {verified: true}; +} + /** A separate, exact operator grant for existing governed work. Context delivery * alone never authorizes launch; neither models nor sources choose host profiles. */ diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index f4c880ea31..6319ad93db 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/loopx/control_plane/effect_runtime_handlers.ts @@ -621,6 +621,7 @@ export function createEffectRuntimeHandlers( ["collaboration.conversation.agent_target", lazyHandler(() => import("./collaboration/conversation_binding.ts"), ({resolveConversationAgentTarget}) => resolveConversationAgentTarget)], ["collaboration.peer.context_access", lazyHandler(() => import("./collaboration/peer_context.ts"), ({requirePeerContextAccess}) => requirePeerContextAccess)], ["collaboration.source.recipients", lazyHandler(() => import("./collaboration/source_grants.ts"), ({resolveSourceRecipients}) => resolveSourceRecipients)], + ["collaboration.source.ingress", lazyHandler(() => import("./collaboration/source_grants.ts"), ({validateSourceIngress}) => validateSourceIngress)], ["collaboration.source.execution_bindings", lazyHandler(() => import("./collaboration/source_grants.ts"), ({sourceExecutionBindings}) => sourceExecutionBindings)], ["collaboration.source.configure_recipient", lazyHandler(() => import("./collaboration/source_grants.ts"), ({configureSourceRecipient}) => configureSourceRecipient)], ["collaboration.source.configure_scope", lazyHandler(() => import("./collaboration/source_grants.ts"), ({configureSourceScope}) => configureSourceScope)], diff --git a/tests/control_plane/test_bound_owner_web_input.py b/tests/control_plane/test_bound_owner_web_input.py new file mode 100644 index 0000000000..0fc6d4a4b7 --- /dev/null +++ b/tests/control_plane/test_bound_owner_web_input.py @@ -0,0 +1,280 @@ +"""Local operator input shares an existing private binding, never a provider event. + +Real HTTP, Chat storage and File/SQLite authority; only the remote identity and +model subprocess are synthetic. This does not qualify installed UI or live Lark. +""" +import http.client +import json +from pathlib import Path +import runpy +import threading + +import pytest + +from canonical_authority_fixture import isolate_sqlite_runtime, promoted_create_fixture +from loopx.capabilities.native_chat.conversation_bindings import ChatConversationBindings +from loopx.capabilities.native_chat.project_context import ChatProjectContexts +from loopx.chat_runtime import ChatRuntimeController +from loopx.chat_server import ChatHTTPServer, ChatRequestHandler +from loopx.chat_store import ChatSessionStore +from loopx.control_plane.collaboration import conversation_scope +from loopx.control_plane.collaboration.source_grant_observation import external_source_policy +from loopx.control_plane.collaboration.source_grant_observation import source_context_target_authority + + +@pytest.fixture(params=["file", "sqlite"]) +def bound_chat(tmp_path, monkeypatch, request): + isolate_sqlite_runtime(tmp_path, monkeypatch) + registry, canonical, state = promoted_create_fixture(tmp_path, provider=request.param) + workspace = state.parents[3] + contexts = ChatProjectContexts([workspace], workspace_grant="workspace_read") + store = ChatSessionStore(tmp_path / "chat") + capture, fake = tmp_path / "requests.jsonl", tmp_path / "codex" + source = runpy.run_path(str(Path(__file__).parents[2] / "examples/loopx-chat-runtime-smoke.py"))["FAKE_CODEX"] + source = source.replace(' method = request.get("method")', + f' with open({str(capture)!r}, "a") as output:\n output.write(json.dumps(request) + "\\n")\n' + ' method = request.get("method")') + # The isolated group host must acknowledge its requested permission profile; + # keep the production profile check intact before testing HTTP source denial. + source = source.replace(' result = {"thread": {"id": "durable-thread"}}', + ' result = {"thread": {"id": "durable-thread"}, "activePermissionProfile": ' + '{"id": request["params"].get("permissions")}, "runtimeWorkspaceRoots": [request["params"]["cwd"]]}') + fake.write_text(source) + fake.chmod(0o700) + runtime = ChatRuntimeController(store=store, codex_bin=str(fake), project_contexts=contexts, registry_path=registry) + runtime.codex_home = tmp_path / "synthetic-codex-home" + runtime.codex_home.mkdir() + observed = {"transport_ref": "private-app", "provider_ref": "c" * 24, + "operator_ref": "d" * 24, "verified": True} + bindings = ChatConversationBindings(root=store.root, project_contexts=contexts, observe=lambda _: dict(observed)) + bindings.controller = runtime + contexts.conversation_bindings = bindings + runtime.manager_scope_resolver = bindings.steward_scope + server = ChatHTTPServer(("127.0.0.1", 0), ChatRequestHandler) + server.chat_store, server.runtime_controller = store, runtime + server.selected_goal_id, server.verbose = None, False + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + + def post(session, body): + connection = http.client.HTTPConnection(*server.server_address, timeout=15) + try: + connection.request("POST", f"/api/chat/sessions/{session['session_id']}/turns", + json.dumps(body), headers={"Content-Type": "application/json"}) + response = connection.getresponse() + return response.status, json.loads(response.read()) + finally: + connection.close() + + def open_bound(kind="project"): + binding = bindings.configure(transport_ref="private-app", project_ref=contexts.available()[0]["project_ref"], + executor_endpoint_id="codex", context_kind=kind, project_grant="workspace_read") + session, _ = runtime.open_session(goal_id=None, agent_id="codex", work_dir=workspace, objective="", + mode="resume_latest", conversation_binding_id=binding["binding_id"], + source_context={"source_ref": "a" * 24, "sender_ref": observed["operator_ref"], "private_human_message": True}) + return binding, session + + yield store, runtime, bindings, observed, post, open_bound, capture, canonical + server.shutdown() + server.server_close() + thread.join(timeout=2) + runtime.close() + + +def test_http_local_owner_continues_private_project_thread_and_queue(bound_chat): + store, runtime, _, _, post, open_bound, capture, _ = bound_chat + _, session = open_bound() + sid, original_thread = session["session_id"], session["upstream_thread_id"] + message = "Keep this original workspace and full message.\n" + ("detail " * 2500).rstrip() + status, accepted = post(session, {"message": message, "client_turn_id": "local-first"}) + assert status == 202, accepted + assert runtime.wait_for_turn(session_id=sid, turn_id=accepted["turn_id"], timeout_sec=15)["status"] == "completed" + status, replay = post(session, {"message": message, "client_turn_id": "local-first", "delivery": "queue"}) + assert status == 202 and replay["turn_id"] == accepted["turn_id"] and not replay["created"] + status, next_turn = post(session, {"message": message, "client_turn_id": "local-second"}) + assert status == 202 and next_turn["turn_id"] != accepted["turn_id"] + assert runtime.wait_for_turn(session_id=sid, turn_id=next_turn["turn_id"], timeout_sec=15)["status"] == "completed" + turn = store.load_turn(sid, accepted["turn_id"]) + assert turn["origin"] == "web" and turn["message"] == message + current = store.load_session(sid) + assert current["goal_id"] is None and current["project_context"] == session["project_context"] + assert current["upstream_thread_id"] == original_thread + assert conversation_scope(current, origin="web") == { + "kind": "project_workspace", "goal_ids": [], "private_conversation": False} + requests = [json.loads(line) for line in capture.read_text().splitlines()] + assert sum(row.get("method") == "thread/start" for row in requests) == 1 + assert sum(row.get("method") == "turn/start" for row in requests) == 2 + + +def test_http_observes_bound_source_once_before_the_publication_fence(bound_chat, monkeypatch): + _, runtime, bindings, observed, post, open_bound, _, _ = bound_chat + _, session = open_bound() + calls = [] + monkeypatch.setattr(bindings, "_observe_owner", lambda profile: calls.append(profile) or dict(observed)) + monkeypatch.setattr(runtime, "resume_session_queue", lambda **_: None) + status, accepted = post(session, {"message": "one source observation", "client_turn_id": "single-observation"}) + assert status == 202, accepted + assert calls == ["private-app"] + + +def test_http_local_input_cannot_inherit_a_group_topic(bound_chat): + store, runtime, bindings, observed, post, _, _, _ = bound_chat + project = runtime.project_contexts.available()[0] + group, topic = "1" * 24, "2" * 24 + binding = bindings.configure(transport_ref="private-app", project_ref=project["project_ref"], + executor_endpoint_id="codex", project_grant="workspace_read", audience="group", + group_refs=[group], available_group_refs=[group]) + session, _ = runtime.open_session(goal_id=None, agent_id="codex", work_dir=Path(project["workspace_path"]), + objective="", mode="resume_latest", conversation_binding_id=binding["binding_id"], + source_context={"source_ref": topic, "sender_ref": observed["operator_ref"], + "private_human_message": False, "group_ref": group, "topic_ref": topic, "group_human_message": True}) + status, denied = post(session, {"message": "not a verified group event", "client_turn_id": "group-web"}) + assert status == 400 and denied["delivery_state"] == "not_delivered" + assert store.queued_turns(session["session_id"]) == [] + + +def test_http_bound_steward_keeps_web_provenance_at_coordination_root(bound_chat, monkeypatch): + store, runtime, _, _, post, open_bound, _, canonical = bound_chat + _, session = open_bound("steward") + # Inspect the real canonical queue before dispatch; admission must persist + # provenance first, without calling a model or creating a parallel inbox. + monkeypatch.setattr(runtime, "resume_session_queue", lambda **_: None) + message = "Ask the already registered agent-a about this exact request." + status, accepted = post(session, {"message": message, "client_turn_id": "local-steward"}) + assert status == 202, accepted + turn = store.load_turn(session["session_id"], accepted["turn_id"]) + assert turn["origin"] == "web" and turn["status"] == "queued" + assert runtime.coordination_runtime_root == canonical != store.root.parent + ingress, policy = external_source_policy(canonical, session, turn) + assert ingress["source_message"] == message and ingress["origin"] == "web" + assert ingress["source_id"].startswith("web:") and ingress["session_id"] == session["session_id"] + assert ingress["client_turn_id"] == turn["client_turn_id"] + assert policy["sender_ids"] == [session["steward_context"]["operator_ref"]] + target = {"goal_id": "goal-a", "agent_id": "agent-a"} + grant = source_context_target_authority(canonical, session, turn, target) + assert grant["mode"] == "context_only" and grant["targets"] == [target] + assert "execution_bindings" not in policy + assert conversation_scope(session, origin="web") == { + "kind": "external_audience", "goal_ids": [], "private_conversation": False, "bound_steward": True} + with pytest.raises((OSError, ValueError)): + external_source_policy(store.root.parent, session, turn) + + +@pytest.mark.parametrize("mutation", ["operator_ref", "provider_ref", "verified", "workspace", "revoked", "closed", "body_origin"]) +def test_local_input_denies_lost_source_and_client_supplied_identity(bound_chat, mutation): + store, runtime, bindings, observed, post, open_bound, _, canonical = bound_chat + binding, session = open_bound() + body = {"message": "must stay unadmitted", "client_turn_id": "denied"} + if mutation in {"operator_ref", "provider_ref"}: + observed[mutation] = "f" * 24 + elif mutation == "verified": + observed[mutation] = False + elif mutation == "workspace": + Path(session["project_context"]["workspace_path"]).rename(canonical.parent / "removed-workspace") + elif mutation == "revoked": + bindings.disconnect(binding["binding_id"], expected_revision=bindings.read()["revision"]) + elif mutation == "closed": + store.update_session(session["session_id"], status="closed") + else: + body["origin"] = "lark" + status, denied = post(session, body) + assert status in {400, 404}, denied + assert denied["delivery_state"] == "not_delivered" + assert store.queued_turns(session["session_id"]) == [] + assert store.load_session(session["session_id"])["active_turn_id"] is None + + +def test_revocation_can_commit_while_provider_verification_waits(bound_chat, monkeypatch): + store, runtime, bindings, observed, post, open_bound, _, _ = bound_chat + binding, session = open_bound("steward") + monkeypatch.setattr(runtime, "resume_session_queue", lambda **_: None) + verifying, release = threading.Event(), threading.Event() + result = [] + + def observe(_): + snapshot = dict(observed) + verifying.set() + assert release.wait(5) + return snapshot + + monkeypatch.setattr(bindings, "_observe_owner", observe) + thread = threading.Thread(target=lambda: result.append(post(session, + {"message": "source revoked while observation waits", "client_turn_id": "race"}))) + thread.start() + try: + assert verifying.wait(5) + # This would deadlock/time out if admission held the publication lock + # during provider IO. The stale observation cannot restore the grant. + bindings.disconnect(binding["binding_id"], expected_revision=bindings.read()["revision"]) + finally: + release.set() + thread.join(timeout=10) + assert not thread.is_alive() and result[0][0] == 400 + assert store.queued_turns(session["session_id"]) == [] + + +def test_binding_publication_fence_covers_canonical_queue_commit(bound_chat, monkeypatch): + store, runtime, bindings, _, post, open_bound, _, _ = bound_chat + binding, session = open_bound() + monkeypatch.setattr(runtime, "resume_session_queue", lambda **_: None) + committing, release, revoked = threading.Event(), threading.Event(), threading.Event() + create = store.create_queued_turn + result, failures = [], [] + + def gated_commit(*args, **kwargs): + committing.set() + assert release.wait(5) + return create(*args, **kwargs) + + def revoke(): + try: + bindings.disconnect(binding["binding_id"], expected_revision=bindings.read()["revision"]) + except Exception as exc: + failures.append(exc) + finally: + revoked.set() + + monkeypatch.setattr(store, "create_queued_turn", gated_commit) + sender = threading.Thread(target=lambda: result.append(post(session, + {"message": "one accepted queue entry", "client_turn_id": "fenced"}))) + sender.start() + revoker = threading.Thread(target=revoke) + try: + assert committing.wait(5) + revoker.start() + assert not revoked.wait(0.15), "revocation crossed the queue publication fence" + finally: + release.set() + sender.join(timeout=10) + if revoker.ident is not None: + revoker.join(timeout=10) + assert revoked.is_set() and not failures + assert result[0][0] == 202, result + turn = store.load_turn(session["session_id"], result[0][1]["turn_id"]) + assert turn["status"] in {"interrupted", "cancelled"} + + +def test_provider_and_local_followups_keep_busy_fifo_and_exact_replay(bound_chat): + store, runtime, _, _, post, open_bound, _, _ = bound_chat + _, session = open_bound() + sid = session["session_id"] + active, _ = runtime.enqueue_turn(session_id=sid, client_turn_id="provider-first", + message="wait for interrupt", work_dir=Path(session["project_context"]["workspace_path"]), objective="", origin="lark") + import time + deadline = time.monotonic() + 5 + while not store.load_turn(sid, active["turn_id"]).get("upstream_turn_id") and time.monotonic() < deadline: + time.sleep(0.01) + assert store.load_session(sid)["active_turn_id"] == active["turn_id"] + queued = [] + for client in ("local-next", "local-after"): + status, accepted = post(session, {"message": "same text, different explicit requests", "client_turn_id": client}) + assert status == 202, accepted + queued.append(accepted["turn_id"]) + status, replay = post(session, {"message": "same text, different explicit requests", "client_turn_id": "local-next"}) + assert status == 202 and not replay["created"] and replay["turn_id"] == queued[0] + assert [turn["turn_id"] for turn in store.queued_turns(sid)] == queued + runtime.interrupt_turn(session_id=sid, turn_id=active["turn_id"]) + for tid in queued: + assert runtime.wait_for_turn(session_id=sid, turn_id=tid, timeout_sec=15)["status"] == "completed" + assert store.load_turn(sid, active["turn_id"])["origin"] == "lark" + assert all(store.load_turn(sid, tid)["origin"] == "web" for tid in queued) diff --git a/tests/control_plane_ts/conversation_binding.test.ts b/tests/control_plane_ts/conversation_binding.test.ts index 3f55951879..85a9b571a8 100644 --- a/tests/control_plane_ts/conversation_binding.test.ts +++ b/tests/control_plane_ts/conversation_binding.test.ts @@ -170,7 +170,7 @@ test("binding independently verifies the owner and does not create a Goal", () = assert.throws(() => planConversationBinding({...request, expected_revision: 1}), /revision/); }); -test("a selected steward has a verified empty portfolio and adopts only its own exact new creation", () => { +test("a selected steward retains its bounded portfolio for provider and local continuation", () => { const steward = {...row, context_kind: "steward", grant: "portfolio_read", goal_ids: []}; const next = planConversationBinding({...request, binding: steward}).state as typeof current; const use = {current: next, binding_id: row.binding_id, source_ref: "e".repeat(24), @@ -179,7 +179,7 @@ test("a selected steward has a verified empty portfolio and adopts only its own assert.equal(resolveConversationScope({channel_id: selected.channel_id, goal_id: "loopx-manager", steward_context: selected.context, origin: "lark"}).bound_steward, true); assert.equal(resolveConversationScope({channel_id: selected.channel_id, goal_id: "loopx-manager", - steward_context: selected.context, origin: "web"}).bound_steward, undefined); + steward_context: selected.context, origin: "web"}).bound_steward, true); const context = selected.context as Record; const proposal = {proposal_id: "proposal-" + "f".repeat(32), status: "applied", action_kind: "goal.create", context, receipt: {outcome: "goal_created", projection_verified: true, resource_ids: {goal_id: "fresh-goal"}}}; @@ -409,7 +409,7 @@ test("attached status names the registered recipient and stop cannot pretend hos }); -test("generic external provenance uses the same typed bound audience as legacy Lark", () => { +test("provider and authorized local input classify the same bounded private audience", () => { for (const contextKind of ["project", "steward"]) { const selectedRow = contextKind === "project" ? row : {...row, context_kind: "steward", grant: "portfolio_read", goal_ids: []}; const next = planConversationBinding({...request, binding: selectedRow}).state; @@ -420,7 +420,42 @@ test("generic external provenance uses the same typed bound audience as legacy L ...(contextKind === "project" ? {project_context: selected.context} : {steward_context: selected.context})}; assert.deepEqual(resolveConversationScope({...session, origin: "external"}), resolveConversationScope({...session, origin: "lark"})); - assert.notDeepEqual(resolveConversationScope({...session, origin: "web"}), + assert.deepEqual(resolveConversationScope({...session, origin: "web"}), resolveConversationScope({...session, origin: "external"})); } }); + + +test("local input requires the exact existing private context and fresh binding proof", () => { + for (const kind of ["project", "steward"]) { + const binding = kind === "project" ? row : {...row, context_kind: "steward", grant: "portfolio_read", goal_ids: []}; + const next = planConversationBinding({...request, binding}).state; + const providerUse = {current: next, binding_id: row.binding_id, source_ref: "e".repeat(24), + sender_ref: row.operator_ref, private_human_message: true, observation, available_projects: [project]}; + const original = resolveBoundConversation(providerUse); + const local = {...providerUse, origin: "web", private_human_message: false, session_context: original.context}; + assert.deepEqual(resolveBoundConversation(local), original); + const scope = resolveConversationScope({goal_id: kind === "project" ? null : "loopx-manager", + channel_id: original.channel_id, project_context: original.context, steward_context: original.context, origin: "web"}); + assert.equal(scope.private_conversation, false); + assert.equal(scope.kind, kind === "project" ? "project_workspace" : "external_audience"); + for (const bad of [{session_context: undefined}, {private_human_message: true}, {origin: "model"}, + {sender_ref: "f".repeat(24)}, {current}, {available_projects: []}, + {observation: {...observation, verified: false}}, + {session_context: {...original.context as object, source_ref: "f".repeat(24)}}]) { + assert.throws(() => resolveBoundConversation({...local, ...bad})); + } + } +}); + +test("local input cannot impersonate a group participant or inherit its workspace", () => { + const groupRef = "1".repeat(24), topic = "2".repeat(24); + const group = {...row, audience: "group", group_refs: [groupRef]}; + const next = planConversationBinding({...request, binding: group, available_group_refs: [groupRef]}).state; + const use = {current: next, binding_id: row.binding_id, source_ref: topic, group_ref: groupRef, topic_ref: topic, + sender_ref: "3".repeat(24), private_human_message: false, group_human_message: true, observation, available_projects: [project]}; + const selected = resolveBoundConversation(use); + assert.throws(() => resolveBoundConversation({...use, origin: "web", session_context: selected.context})); + assert.equal(resolveConversationScope({goal_id: null, channel_id: selected.channel_id, + project_context: selected.context, origin: "web"}).kind, "unavailable"); +}); diff --git a/tests/control_plane_ts/source_grants.test.ts b/tests/control_plane_ts/source_grants.test.ts index 9f19a50736..df102be292 100644 --- a/tests/control_plane_ts/source_grants.test.ts +++ b/tests/control_plane_ts/source_grants.test.ts @@ -1,6 +1,6 @@ import assert from "node:assert/strict"; import test from "node:test"; -import { configureSourceRecipient, configureSourceScope, resolveSourceRecipients, sourceExecutionBindings } from "../../loopx/control_plane/collaboration/source_grants.ts"; +import { configureSourceRecipient, configureSourceScope, resolveSourceRecipients, sourceExecutionBindings, validateSourceIngress } from "../../loopx/control_plane/collaboration/source_grants.ts"; const worker = { goal_id: "research", agent_id: "worker" }; const peer = { goal_id: "research", agent_id: "peer" }; @@ -149,3 +149,33 @@ test("evidence reads follow current registered Goals only after explicit all-reg assert.throws(() => resolveSourceEvidenceGoals({source: invalid})); } }); + + +test("source provenance is exact and local bound input never borrows provider lineage", () => { + const steward = {kind: "bound_steward", project_ref: "a".repeat(24), workspace_path: "/authorized/project", + audience: "bound_owner", grant: "portfolio_read", binding_id: "b".repeat(24), source_ref: "c".repeat(24), + provider_ref: "d".repeat(24), operator_ref: "e".repeat(24), goal_ids: []}; + const session = {session_id: "original", goal_id: "loopx-manager", steward_context: steward, + channel_id: `manager.external.native.${steward.binding_id}.${steward.source_ref}`}; + const turn = {client_turn_id: "local", origin: "web", message_digest: "digest"}; + const ingress = {session_id: "original", client_turn_id: "local", channel: session.channel_id, + message_digest: "digest", sender_id: steward.operator_ref, source_id: "web:original", origin: "web"}; + const params = {session, turn, ingress}; + assert.deepEqual(validateSourceIngress(params), {verified: true}); + for (const field of ["session_id", "client_turn_id", "channel", "message_digest", "sender_id", "origin"]) { + assert.throws(() => validateSourceIngress({...params, ingress: {...ingress, [field]: "other"}})); + } + for (const origin of ["lark", "external", "model"]) { + assert.throws(() => validateSourceIngress({...params, turn: {...turn, origin}})); + } + assert.throws(() => validateSourceIngress({...params, ingress: {...ingress, origin: undefined}})); + assert.throws(() => validateSourceIngress({...params, session: {...session, steward_context: undefined}})); + // Existing provider records omit origin and keep their historical shape. + const legacy = {...ingress, origin: undefined}; + for (const origin of ["lark", "external"]) { + assert.deepEqual(validateSourceIngress({...params, turn: {...turn, origin}, ingress: legacy}), {verified: true}); + } + const unbound = {...session, steward_context: undefined}; + assert.deepEqual(validateSourceIngress({...params, session: unbound, turn: {...turn, origin: "lark"}, ingress: legacy}), {verified: true}); + assert.throws(() => validateSourceIngress({...params, session: unbound, turn: {...turn, origin: "external"}, ingress: legacy})); +}); diff --git a/tests/test_chat_ordinary_project.py b/tests/test_chat_ordinary_project.py index 4877aff29e..2a63733c2f 100644 --- a/tests/test_chat_ordinary_project.py +++ b/tests/test_chat_ordinary_project.py @@ -283,8 +283,11 @@ def test_default_project_host_grant_is_write_and_read_only_launch_is_enforced(or _, opened = request("/api/chat/sessions", {"context_kind": "project", "project_ref": projects["projects"][0]["project_ref"]}) assert opened["goal_id"] is None assert runtime.adapters[opened["session_id"]].session.sandbox == "workspace-write" - row = json.loads(capture.read_text().splitlines()[-1]) - assert row["method"] == "thread/start" and row["params"]["sandbox"] == "workspace-write" + # Native thread naming may follow startup; verify the permission-bearing + # request itself rather than assuming it is the last transport operation. + starts = [row for line in capture.read_text().splitlines() + if (row := json.loads(line)).get("method") == "thread/start"] + assert len(starts) == 1 and starts[0]["params"]["sandbox"] == "workspace-write" proof = {"transport_ref": "notes-app", "provider_ref": "c" * 24, "operator_ref": "d" * 24, "verified": True} bindings = ChatConversationBindings(root=store.root, project_contexts=contexts, observe=lambda _: proof) contexts.conversation_bindings = bindings