diff --git a/offline-backend-20260512/backend/CLAUDE.md b/offline-backend-20260512/backend/CLAUDE.md index 453bb26d..488331e5 100644 --- a/offline-backend-20260512/backend/CLAUDE.md +++ b/offline-backend-20260512/backend/CLAUDE.md @@ -528,8 +528,19 @@ node state stay consistent. A rejected/stale node returns to `done` only after a fresh Markdown delivery is recorded. 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 -text outputs; binary artifacts stay as metadata-only references. -`POST /sessions/{id}/archive` turns a session into a read-only historical +text outputs; binary artifacts stay as metadata-only references. Because each +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 turn completion are rejected, while read endpoints remain available. Built-in summary and action-plan completion also preserves a visible report diff --git a/offline-backend-20260512/backend/app/gateway/routers/position_roundtable.py b/offline-backend-20260512/backend/app/gateway/routers/position_roundtable.py index 005fade1..4cccf12d 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/position_roundtable.py +++ b/offline-backend-20260512/backend/app/gateway/routers/position_roundtable.py @@ -46,6 +46,17 @@ NODE_STATUSES = {"locked", "ready", "running", "done", "stale", "rejected", "err _HANDOFF_TEXT_BUDGET = 12_000 _MAX_INLINE_ARTIFACTS_PER_NODE = 4 _MAX_INLINE_ARTIFACT_CHARS = 1_800 +# Upstream deliverables are additionally mirrored into the current seat's own +# workspace (outputs/ → workspace/upstream/) 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 # in SQL must retain a complete normal Markdown/JSON deliverable. The ceiling # protects the database from an accidental multi-gigabyte log while covering @@ -685,6 +696,11 @@ def _build_node_context_text( continue artifact_name = artifact.get("name") or artifact.get("path") or "未命名产物" 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() if content: lines.append(f" 摘要/正文片段：{content}") @@ -824,6 +840,95 @@ async def _artifact_refs_with_content( 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]: """Construct bounded, deterministic hidden context for one direct-agent turn.""" store = _get_store(request) @@ -877,6 +982,20 @@ async def _handoff_context(request: Request, session_id: str, node_key: str) -> ) if remaining <= 0: 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 { "kind": "position_roundtable_handoff", "instructions": _build_node_context_text( diff --git a/offline-backend-20260512/backend/tests/test_position_roundtable_sessions.py b/offline-backend-20260512/backend/tests/test_position_roundtable_sessions.py index 84082b4e..df6225e1 100644 --- a/offline-backend-20260512/backend/tests/test_position_roundtable_sessions.py +++ b/offline-backend-20260512/backend/tests/test_position_roundtable_sessions.py @@ -6,6 +6,7 @@ import asyncio import importlib.util import json import zipfile +from pathlib import Path from types import SimpleNamespace from typing import Any @@ -1244,8 +1245,11 @@ def test_activation_requires_a_structured_intent_snapshot() -> None: @pytest.mark.asyncio 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") + mirror_root = tmp_path / "current" / "workspace" / "upstream" calls: list[tuple[str, str]] = [] 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): calls.append((thread_id, path)) - assert thread_id == "upstream-thread" - assert path == "mnt/user-data/outputs/handoff.md" - return artifact + if thread_id == "upstream-thread": + assert path == "mnt/user-data/outputs/handoff.md" + 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) 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 upstream["latest_answer"] == "上游问答结果" 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 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