vent_agent_2.0/tests/test_workspace_backend.py

"""文件后端在"缺 session 身份的 Runtime"下必须能自救。

背景（真实故障）：摘要压缩中间件在 `awrap_model_call` 里需要后端，它按框架自己的写法
构造一个 runtime：

```python
config = cast("RunnableConfig", getattr(runtime, "config", {}))
tool_runtime = ToolRuntime(..., config=config, ...)
```

而 `ModelRequest.runtime` 的类型是 `langgraph.runtime.Runtime`，**它没有 `config` 字段**，
于是 config 恒为 `{}`。若工厂只在"属性不存在"时回退运行上下文，这里就会直接抛
`ValueError: 文件后端缺少服务端会话身份`——表现为长会话（或读了长文档的子智能体）
一触发压缩就整轮失败，而不是压缩生效。
"""
from __future__ import annotations

from typing import Any

import pytest

from agents import workspace
from agents.workspace import backend_for_runtime, session_root


@pytest.fixture(autouse=True)
def isolate_root(tmp_path, monkeypatch):
    """把沙箱根挪到临时目录，避免测试往仓库 data/workspaces 写文件。"""
    monkeypatch.setattr(workspace, "_SANDBOX_BASE", tmp_path, raising=False)
    monkeypatch.setattr(workspace.config, "ROOT", tmp_path, raising=False)
    return tmp_path


class _LanggraphRuntime:
    """最小复刻 langgraph.runtime.Runtime：**故意没有 config 属性**。"""

    def __init__(self) -> None:
        self.context = None
        self.store = None
        self.stream_writer = lambda _: None


def _call_factory_with_empty_config(seen: dict[str, Any]) -> None:
    """在 langgraph 运行上下文里，用框架那种 config={} 的 ToolRuntime 调工厂。"""
    from langchain.tools import ToolRuntime

    runtime = ToolRuntime(
        state={}, context=None, stream_writer=lambda _: None, store=None,
        config={}, tool_call_id=None, tools=[],
    )
    backend = backend_for_runtime(runtime)
    seen["routes"] = sorted(backend.routes)
    seen["has_default"] = backend.default is not None


def test_backend_falls_back_to_run_context_when_runtime_config_is_empty():
    """config 为空字典时必须回退到运行上下文里的会话身份，而不是直接失败。"""
    from langgraph.graph import END, START, StateGraph

    seen: dict[str, Any] = {}

    def node(_state):
        _call_factory_with_empty_config(seen)
        return {}

    graph = StateGraph(dict)
    graph.add_node("n", node)
    graph.add_edge(START, "n")
    graph.add_edge("n", END)
    graph.compile().invoke(
        {}, {"configurable": {"workspace_user": "alice", "workspace_session": "s-1"}})

    assert "/skills/" in seen["routes"] and seen["has_default"]
    # 落到了正确的会话沙箱（而不是别处或临时目录）
    assert (session_root("alice", "s-1") / "workspace").is_dir()


def test_backend_still_fails_loudly_without_any_session_identity():
    """两个来源都没有身份时仍要失败——绝不能悄悄写到别的会话目录。"""
    from langgraph.graph import END, START, StateGraph

    def node(_state):
        backend_for_runtime(_LanggraphRuntime())
        return {}

    graph = StateGraph(dict)
    graph.add_node("n", node)
    graph.add_edge(START, "n")
    graph.add_edge("n", END)
    with pytest.raises(ValueError, match="缺少服务端会话身份"):
        graph.compile().invoke({}, {"configurable": {"unrelated": "x"}})


def test_backend_accepts_runtime_whose_config_carries_identity():
    """文件工具的路径（ToolRuntime 带完整 config）照旧工作。"""
    from langchain.tools import ToolRuntime

    runtime = ToolRuntime(
        state={}, context=None, stream_writer=lambda _: None, store=None, tools=[],
        config={"configurable": {"workspace_user": "bob", "workspace_session": "s-2"}},
        tool_call_id="call-1",
    )
    backend = backend_for_runtime(runtime)
    assert backend.default is not None
    assert (session_root("bob", "s-2") / "workspace").is_dir()


def test_backend_prefers_runtime_config_over_run_context():
    """两者都有身份时以 runtime.config 为准（显式传入的更权威）。"""
    from langchain.tools import ToolRuntime

    runtime = ToolRuntime(
        state={}, context=None, stream_writer=lambda _: None, store=None, tools=[],
        config={"configurable": {"workspace_user": "from-config", "workspace_session": "s"}},
        tool_call_id="call-1",
    )
    from langgraph.graph import END, START, StateGraph

    def node(_state):
        backend_for_runtime(runtime)
        return {}

    graph = StateGraph(dict)
    graph.add_node("n", node)
    graph.add_edge(START, "n")
    graph.add_edge("n", END)
    graph.compile().invoke(
        {}, {"configurable": {"workspace_user": "from-context", "workspace_session": "s"}})
    assert (session_root("from-config", "s") / "workspace").is_dir()