fix: 岗位会商 handoff 将上游产物镜像到当前席位沙箱

席位线程隔离导致下游 read_file 找不到上游路径;镜像到 workspace/upstream 并改写 handoff 引用,附带安全路径校验与 SQL 回退测试。

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
董宏凯andCursor committed 2026-08-27 18:35:17 +08:00
1 parent 440b490b0d
commit 32e509b2b3
3 files changed
+238 -7

No files matched your search

+13 -2
View File
@@ -528,8 +528,19 @@ node state stay consistent. A rejected/stale node returns to `done` only after
a fresh Markdown delivery is recorded. a fresh Markdown delivery is recorded.
For direct-agent handoff, the router resolves artifact paths only inside the For direct-agent handoff, the router resolves artifact paths only inside the
completed upstream node's own thread, then includes bounded UTF-8 excerpts for completed upstream node's own thread, then includes bounded UTF-8 excerpts for
text outputs; binary artifacts stay as metadata-only references. text outputs; binary artifacts stay as metadata-only references. Because each
`POST /sessions/{id}/archive` turns a session into a read-only historical seat runs on an isolated LangGraph thread, a downstream `read_file` on the
upstream artifact path would fail with "File not found"; the handoff therefore
also **mirrors** every upstream `outputs/` deliverable into the current seat's
own `workspace/upstream/` directory (`_mirror_upstream_artifacts`, bytes from
the upstream sandbox file with the SQL recovery copy as fallback, idempotent
per turn via size comparison, capped at 8 files / 2 MB each / 8 MB total) and
rewrites each handoff artifact ref's `path` to the mirror (`upstream_path`
keeps the original; `mirrored_to_sandbox: true`), with a matching note in the
hidden instructions telling the agent it may `read_file` the full text. The
mirror deliberately stays outside `outputs/` so it never appears in the seat's
own delivery manifest.
`POST /sessions/{id}/archive` turns the session into a read-only historical
record: task/intent updates, node binding, hidden-context construction, and record: task/intent updates, node binding, hidden-context construction, and
turn completion are rejected, while read endpoints remain available. turn completion are rejected, while read endpoints remain available.
Built-in summary and action-plan completion also preserves a visible report Built-in summary and action-plan completion also preserves a visible report
@@ -46,6 +46,17 @@ NODE_STATUSES = {"locked", "ready", "running", "done", "stale", "rejected", "err
_HANDOFF_TEXT_BUDGET = 12_000 _HANDOFF_TEXT_BUDGET = 12_000
_MAX_INLINE_ARTIFACTS_PER_NODE = 4 _MAX_INLINE_ARTIFACTS_PER_NODE = 4
_MAX_INLINE_ARTIFACT_CHARS = 1_800 _MAX_INLINE_ARTIFACT_CHARS = 1_800
# Upstream deliverables are additionally mirrored into the current seat's own
# workspace (outputs/<rel> → workspace/upstream/<rel>) before each turn. Every
# seat runs on an isolated LangGraph thread, so without this copy a downstream
# ``read_file`` on the upstream artifact path fails with "File not found" and
# the seat is limited to the short excerpt inlined into the handoff prompt.
# The mirror deliberately stays outside outputs/ so it never appears in this
# seat's own delivery manifest.
_UPSTREAM_MIRROR_PREFIX = "mnt/user-data/workspace/upstream"
_MAX_MIRRORED_ARTIFACTS = 8
_MAX_MIRRORED_ARTIFACT_BYTES = 2_000_000
_MIRRORED_TOTAL_BYTES = 8_000_000
# The handoff prompt remains intentionally small, but the recovery copy stored # The handoff prompt remains intentionally small, but the recovery copy stored
# in SQL must retain a complete normal Markdown/JSON deliverable. The ceiling # in SQL must retain a complete normal Markdown/JSON deliverable. The ceiling
# protects the database from an accidental multi-gigabyte log while covering # protects the database from an accidental multi-gigabyte log while covering
@@ -685,6 +696,11 @@ def _build_node_context_text(
continue continue
artifact_name = artifact.get("name") or artifact.get("path") or "未命名产物" artifact_name = artifact.get("name") or artifact.get("path") or "未命名产物"
lines.append(f" - {artifact_name}") lines.append(f" - {artifact_name}")
if artifact.get("mirrored_to_sandbox") and artifact.get("path"):
lines.append(
f" 完整文件已同步到本席位沙箱 /{artifact['path']},可直接用 read_file 读取全文"
"(仅供参考;本席位自己的交付物仍必须写入 /mnt/user-data/outputs/)。"
)
content = str(artifact.get("content") or "").strip() content = str(artifact.get("content") or "").strip()
if content: if content:
lines.append(f" 摘要/正文片段:{content}") lines.append(f" 摘要/正文片段:{content}")
@@ -824,6 +840,95 @@ async def _artifact_refs_with_content(
return refs, consumed return refs, consumed
def _outputs_relative_path(value: Any) -> str:
"""Return the ``outputs/``-relative posix path of a manifest artifact.
Empty string when the path is not a plain file directly under
``mnt/user-data/outputs/`` (wrong prefix, traversal segment, ...).
"""
path = str(value or "").replace("\\", "/").lstrip("/")
prefix = "mnt/user-data/outputs/"
if not path.startswith(prefix):
return ""
relative = path[len(prefix) :]
if not relative:
return ""
segments = relative.split("/")
if any(segment in ("", ".", "..") for segment in segments):
return ""
return relative
def _mirror_virtual_path(relative: str) -> str:
return f"{_UPSTREAM_MIRROR_PREFIX}/{relative}"
async def _mirror_upstream_artifacts(
current: dict[str, Any],
upstream_candidates: list[dict[str, Any]],
) -> dict[str, str]:
"""Copy upstream deliverables into the current seat's workspace sandbox.
Bytes are preferred from the upstream thread's own sandbox file; the SQL
recovery copy persisted in the manifest is the fallback when that disk is
gone. Only server-recorded upstream threads and manifest paths that pass
:func:`_outputs_relative_path` are read, and the target resolves through
the same thread virtual-path resolver, so a client-supplied manifest can
never point the copy outside a thread's user-data directory. Idempotent
per turn: an identically sized mirror is left untouched. Best effort —
failures simply keep the excerpt-only handoff behaviour.
"""
target_thread_id = str(current.get("thread_id") or "")
if not target_thread_id or not upstream_candidates:
return {}
mirrored: dict[str, str] = {}
total_bytes = 0
for node in upstream_candidates:
source_thread_id = str(node.get("thread_id") or "")
for artifact in node.get("artifact_manifest") or []:
if len(mirrored) >= _MAX_MIRRORED_ARTIFACTS or total_bytes >= _MIRRORED_TOTAL_BYTES:
return mirrored
if not isinstance(artifact, dict):
continue
original = str(artifact.get("path") or "")
relative = _outputs_relative_path(original)
if not relative:
continue
try:
payload: bytes | None = None
if source_thread_id:
source_path = await aresolve_thread_virtual_path(source_thread_id, original)
if await asyncio.to_thread(source_path.is_file):
stat = await asyncio.to_thread(source_path.stat)
if stat.st_size > _MAX_MIRRORED_ARTIFACT_BYTES:
continue
payload = await asyncio.to_thread(source_path.read_bytes)
if payload is None:
content = artifact.get("content")
if not isinstance(content, str) or not content:
continue
payload = content.encode("utf-8")
if len(payload) > _MAX_MIRRORED_ARTIFACT_BYTES:
continue
mirror_virtual = _mirror_virtual_path(relative)
target_path = await aresolve_thread_virtual_path(target_thread_id, mirror_virtual)
if not (
await asyncio.to_thread(target_path.is_file)
and (await asyncio.to_thread(target_path.stat)).st_size == len(payload)
):
await asyncio.to_thread(target_path.parent.mkdir, parents=True, exist_ok=True)
await asyncio.to_thread(target_path.write_bytes, payload)
mirrored[original] = mirror_virtual
total_bytes += len(payload)
except (HTTPException, OSError, ValueError):
# The upstream file may have been removed, or a provider may be
# unavailable. Skip it; the inline excerpt still carries the
# essential content into the handoff.
continue
return mirrored
async def _handoff_context(request: Request, session_id: str, node_key: str) -> dict[str, Any]: async def _handoff_context(request: Request, session_id: str, node_key: str) -> dict[str, Any]:
"""Construct bounded, deterministic hidden context for one direct-agent turn.""" """Construct bounded, deterministic hidden context for one direct-agent turn."""
store = _get_store(request) store = _get_store(request)
@@ -877,6 +982,20 @@ async def _handoff_context(request: Request, session_id: str, node_key: str) ->
) )
if remaining <= 0: if remaining <= 0:
break break
# Point each artifact ref at the workspace mirror when one exists: the
# agent then reads a path that is actually resolvable inside its own
# thread sandbox instead of the upstream thread's outputs directory.
mirrored_paths = await _mirror_upstream_artifacts(current, upstream_candidates)
for node in upstream_nodes:
for artifact in node.get("artifacts") or []:
mirror = mirrored_paths.get(str(artifact.get("path") or ""))
if not mirror:
continue
artifact["upstream_path"] = artifact["path"]
artifact["path"] = mirror
artifact["mirrored_to_sandbox"] = True
return { return {
"kind": "position_roundtable_handoff", "kind": "position_roundtable_handoff",
"instructions": _build_node_context_text( "instructions": _build_node_context_text(
@@ -6,6 +6,7 @@ import asyncio
import importlib.util import importlib.util
import json import json
import zipfile import zipfile
from pathlib import Path
from types import SimpleNamespace from types import SimpleNamespace
from typing import Any from typing import Any
@@ -1244,8 +1245,11 @@ def test_activation_requires_a_structured_intent_snapshot() -> None:
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_handoff_context_inlines_only_upstream_thread_text_artifacts(tmp_path, monkeypatch): async def test_handoff_context_inlines_only_upstream_thread_text_artifacts(tmp_path, monkeypatch):
artifact = tmp_path / "handoff.md" upstream_dir = tmp_path / "up" / "outputs"
upstream_dir.mkdir(parents=True)
artifact = upstream_dir / "handoff.md"
artifact.write_text("上游产物正文:可供下一岗位直接参考。", encoding="utf-8") artifact.write_text("上游产物正文:可供下一岗位直接参考。", encoding="utf-8")
mirror_root = tmp_path / "current" / "workspace" / "upstream"
calls: list[tuple[str, str]] = [] calls: list[tuple[str, str]] = []
class StubStore: class StubStore:
@@ -1307,9 +1311,12 @@ async def test_handoff_context_inlines_only_upstream_thread_text_artifacts(tmp_p
async def resolve(thread_id: str, path: str): async def resolve(thread_id: str, path: str):
calls.append((thread_id, path)) calls.append((thread_id, path))
assert thread_id == "upstream-thread" if thread_id == "upstream-thread":
assert path == "mnt/user-data/outputs/handoff.md" assert path == "mnt/user-data/outputs/handoff.md"
return artifact return artifact
assert thread_id == "current-thread"
assert path == "mnt/user-data/workspace/upstream/handoff.md"
return mirror_root / "handoff.md"
monkeypatch.setattr(position_roundtable_router, "aresolve_thread_virtual_path", resolve) monkeypatch.setattr(position_roundtable_router, "aresolve_thread_virtual_path", resolve)
request = SimpleNamespace( request = SimpleNamespace(
@@ -1325,8 +1332,102 @@ async def test_handoff_context_inlines_only_upstream_thread_text_artifacts(tmp_p
assert len(context["upstream_nodes"]) == 1 assert len(context["upstream_nodes"]) == 1
assert upstream["latest_answer"] == "上游问答结果" assert upstream["latest_answer"] == "上游问答结果"
assert upstream["artifacts"][0]["content"] == "上游产物正文:可供下一岗位直接参考。" assert upstream["artifacts"][0]["content"] == "上游产物正文:可供下一岗位直接参考。"
assert upstream["artifacts"][0]["upstream_path"] == "mnt/user-data/outputs/handoff.md"
assert upstream["artifacts"][0]["path"] == "mnt/user-data/workspace/upstream/handoff.md"
assert upstream["artifacts"][0]["mirrored_to_sandbox"] is True
assert "content" not in upstream["artifacts"][1] assert "content" not in upstream["artifacts"][1]
assert calls == [("upstream-thread", "mnt/user-data/outputs/handoff.md")] assert "mirrored_to_sandbox" not in upstream["artifacts"][1]
assert upstream["artifacts"][1]["path"] == "mnt/user-data/uploads/outside.md"
assert (mirror_root / "handoff.md").read_text(encoding="utf-8") == "上游产物正文:可供下一岗位直接参考。"
assert "/mnt/user-data/workspace/upstream/handoff.md" in context["instructions"]
assert set(calls) == {
("upstream-thread", "mnt/user-data/outputs/handoff.md"),
("current-thread", "mnt/user-data/workspace/upstream/handoff.md"),
}
@pytest.mark.asyncio
async def test_handoff_mirror_falls_back_to_persisted_content_and_skips_unsafe_paths(
tmp_path, monkeypatch
):
mirror_root = tmp_path / "mirror"
resolved: list[tuple[str, str]] = []
class StubStore:
async def get_session(self, session_id, user_id):
assert (session_id, user_id) == ("session-a", "user-a")
return {"task_snapshot": {"title": "测试任务"}, "intent_snapshot": {}}
async def list_nodes(self, session_id, user_id):
return [
{
"node_key": "stage-1-seat-1-agent-a",
"stage_index": 0,
"seat_index": 0,
"agent_id": "agent-a",
"position_id": "planning",
"thread_id": "upstream-thread",
"status": "done",
"revision": 1,
"latest_answer": "上游问答结果",
"artifact_manifest": [
{
"name": "gone.md",
"path": "mnt/user-data/outputs/gone.md",
"mime_type": "text/markdown",
"content": "数据库恢复正文",
},
{
"name": "escape.md",
"path": "mnt/user-data/outputs/../escape.md",
"mime_type": "text/markdown",
"content": "不应被镜像",
},
],
},
{
"node_key": "stage-2-seat-1-agent-b",
"stage_index": 1,
"seat_index": 0,
"agent_id": "agent-b",
"position_id": "solution",
"thread_id": "current-thread",
"status": "ready",
"revision": 0,
"latest_answer": None,
"artifact_manifest": [],
},
]
missing_source = tmp_path / "missing-source"
async def resolve(thread_id: str, path: str):
resolved.append((thread_id, path))
if thread_id == "upstream-thread":
return missing_source / "not-materialized.md"
assert thread_id == "current-thread"
return mirror_root / Path(path).name
monkeypatch.setattr(position_roundtable_router, "aresolve_thread_virtual_path", resolve)
request = SimpleNamespace(
app=SimpleNamespace(state=SimpleNamespace(position_roundtable_store=StubStore())),
state=SimpleNamespace(user=SimpleNamespace(id="user-a")),
)
context = await position_roundtable_router._handoff_context(
request, "session-a", "stage-2-seat-1-agent-b"
)
artifacts = context["upstream_nodes"][0]["artifacts"]
assert artifacts[0]["path"] == "mnt/user-data/workspace/upstream/gone.md"
assert artifacts[0]["mirrored_to_sandbox"] is True
assert (mirror_root / "gone.md").read_text(encoding="utf-8") == "数据库恢复正文"
# 持久化正文同时内嵌,沙箱磁盘丢失也不影响下一岗位读到内容。
assert artifacts[0]["content"] == "数据库恢复正文"
assert artifacts[1]["path"] == "mnt/user-data/outputs/../escape.md"
assert "mirrored_to_sandbox" not in artifacts[1]
assert not (mirror_root / "escape.md").exists()
assert all(".." not in path for _, path in resolved)
@pytest.mark.asyncio @pytest.mark.asyncio