vent_agent_2.0/extensions/vent_airflow/workspace_paths.py

"""会话工作区路径解析：让工具能读写当前会话沙箱里的文件。

工具收到的路径是模型看到的**虚拟路径**（如 `/workspace/uploads/xxx.pdf`）。要打开
真实文件，需要把虚拟路径映射回 `data/workspaces/<user_hash>/<session_hash>/`。
会话身份由运行层写入图配置的 `configurable`，这里用 langgraph 的 get_config 取回。

为什么在这里重写哈希映射而不是直接 import `agents.workspace.session_root`：分层纪律
要求工具侧不依赖 agents 层。风险是两处映射漂移，因此 `tests/test_airflow_tools.py`
里有一条断言两者结果一致的测试——漂移会变成失败的测试，而不是线上找不到文件。
"""
from __future__ import annotations

import hashlib
from pathlib import Path

import config

from algorithms.airflow_review.contracts import now_iso


class WorkspacePathError(ValueError):
    """无法确定会话沙箱或路径越界。"""


def workspace_root() -> Path | None:
    """取当前会话的沙箱根；不在运行上下文里（CLI/测试）返回 None。"""
    try:
        from langgraph.config import get_config

        configurable = (get_config() or {}).get("configurable", {}) or {}
    except Exception:
        return None
    user, session = configurable.get("workspace_user"), configurable.get("workspace_session")
    if not user or not session:
        return None
    return session_root(str(user), str(session))


def session_root(user_name: str, session_id: str) -> Path:
    """与 agents.workspace.session_root 保持一致的映射（见模块说明）。"""
    user_key = hashlib.sha256(user_name.encode()).hexdigest()
    session_key = hashlib.sha256(session_id.encode()).hexdigest()
    return config.ROOT / "data" / "workspaces" / user_key / session_key


def resolve_input(path: str, *, allow_repo_relative: bool = True) -> Path:
    """把模型给的路径解析成可读的真实文件路径。

    支持三种写法：会话虚拟路径（/workspace/...）、项目内相对路径（便于脚本与测试）、
    绝对路径（要求文件存在）。越出沙箱的路径一律拒绝。
    """
    if not path or not str(path).strip():
        raise WorkspacePathError("文件路径为空")
    raw = str(path).strip().replace("\\", "/")
    root = workspace_root()
    if raw.startswith("/workspace/") or raw.startswith("/conversation_history/"):
        if root is None:
            raise WorkspacePathError(
                "该路径是会话工作区路径，但当前不在会话运行上下文中，无法定位沙箱；"
                "请在对话中调用，或改用真实文件路径")
        target = (root / raw.lstrip("/")).resolve()
        if root.resolve() not in target.parents and target != root.resolve():
            raise WorkspacePathError(f"路径越出会话沙箱：{path}")
        return target
    candidate = Path(raw)
    if candidate.is_absolute():
        if not candidate.is_file():
            raise WorkspacePathError(f"文件不存在：{candidate}")
        return candidate
    if allow_repo_relative:
        repo_relative = (config.ROOT / raw).resolve()
        if repo_relative.is_file():
            return repo_relative
    raise WorkspacePathError(f"无法解析文件路径：{path}")


def output_path(name: str, *, suffix: str = ".json") -> tuple[Path, str]:
    """为本次工具调用分配一个输出文件；返回 (真实路径, 虚拟路径)。

    没有会话上下文时落到项目 data/airflow_outputs/，保证脚本与测试也能跑。
    """
    safe = "".join(ch if ch.isalnum() or ch in "-_" else "_" for ch in name)[:60]
    stamp = now_iso().replace(":", "").replace("-", "").replace("+", "")[:15]
    filename = f"{stamp}-{safe}{suffix}"
    root = workspace_root()
    if root is None:
        directory = config.resolve_path(config.AIRFLOW_OUTPUT_DIR
                                       or f"data/{config.AIRFLOW_OUTPUT_DIRNAME}")
        directory.mkdir(parents=True, exist_ok=True)
        return directory / filename, str(directory / filename)
    directory = root / "workspace" / "outputs" / config.AIRFLOW_OUTPUT_DIRNAME
    directory.mkdir(parents=True, exist_ok=True)
    relative = f"/workspace/outputs/{config.AIRFLOW_OUTPUT_DIRNAME}/{filename}"
    return directory / filename, relative


def write_artifact(payload: str | bytes, name: str, *, suffix: str = ".json") -> str:
    """写中间产物并返回模型可读的虚拟路径（无会话上下文时返回真实路径）。"""
    path, virtual = output_path(name, suffix=suffix)
    if isinstance(payload, bytes):
        path.write_bytes(payload)
    else:
        path.write_text(payload, encoding="utf-8")
    return virtual