deer-flow/backend/tests/test_local_sandbox_content_path_translation.py

"""Issue #7461 — only executable content gets container paths translated.

``write_file`` rewrites container paths inside the bytes it stores so a script
the agent wrote can run under a host interpreter (PR #1935). Applying that to
documents put host paths into deliverables, and ``read_file`` masked the
rewrite for agent-written files, so the agent read back the virtual path and
reported content it had not stored. Documents, data, and configs are now
stored verbatim; code files keep the translation.
"""

from __future__ import annotations

import builtins
import os
import shutil
import threading
from pathlib import Path

import pytest

from deerflow.sandbox.local.local_sandbox import LocalSandbox, PathMapping


def _sandbox(tmp_path: Path) -> LocalSandbox:
    outputs = tmp_path / "outputs"
    workspace = tmp_path / "workspace"
    outputs.mkdir()
    workspace.mkdir()
    return LocalSandbox(
        id="test",
        path_mappings=[
            PathMapping(container_path="/mnt/user-data/outputs", local_path=str(outputs)),
            PathMapping(container_path="/mnt/user-data/workspace", local_path=str(workspace)),
        ],
    )


@pytest.mark.parametrize(
    "name,content",
    [
        ("README.md", "# Run notes\nThe results table is saved at /mnt/user-data/outputs/results.csv\n"),
        ("notes.txt", "see /mnt/user-data/outputs/results.csv and /mnt/user-data/workspace/plot.png\n"),
        ("data.csv", "label,path\nrows,/mnt/user-data/outputs/results.csv\n"),
        ("settings.json", '{"output": "/mnt/user-data/outputs/results.csv"}\n'),
        ("export.yaml", "output: /mnt/user-data/outputs/results.csv\n"),
    ],
    ids=["markdown", "text", "csv", "json", "yaml"],
)
def test_non_executable_content_is_stored_verbatim(tmp_path, name, content):
    """A dictation of exact bytes must survive the write and the read-back."""
    sandbox = _sandbox(tmp_path)
    virtual = f"/mnt/user-data/outputs/{name}"

    sandbox.write_file(virtual, content)

    stored = (tmp_path / "outputs" / name).read_bytes().decode("utf-8")
    assert stored == content
    # read_file returns the stored bytes, so the agent cannot claim a rewrite it did not see.
    assert sandbox.read_file(virtual) == content


def test_extensionless_document_without_shebang_is_stored_verbatim(tmp_path):
    """``README`` with no suffix is a document, not a program (the shared rule's negative case)."""
    sandbox = _sandbox(tmp_path)
    content = "Install notes: /mnt/user-data/workspace/setup.sh\n"

    sandbox.write_file("/mnt/user-data/outputs/README", content)

    assert (tmp_path / "outputs" / "README").read_text(encoding="utf-8") == content


def test_append_keeps_document_paths_virtual(tmp_path):
    sandbox = _sandbox(tmp_path)
    sandbox.write_file("/mnt/user-data/outputs/README.md", "# Run notes\n")
    sandbox.write_file("/mnt/user-data/outputs/README.md", "saved at /mnt/user-data/outputs/results.csv\n", append=True)

    assert (tmp_path / "outputs" / "README.md").read_text(encoding="utf-8") == ("# Run notes\nsaved at /mnt/user-data/outputs/results.csv\n")


def test_append_to_shebang_script_still_translates_paths(tmp_path):
    """An extensionless script's interpreter line stays at the top of the file."""
    sandbox = _sandbox(tmp_path)
    virtual = "/mnt/user-data/workspace/collect"
    sandbox.write_file(virtual, "#!/bin/bash\nset -eu\n")
    sandbox.write_file(virtual, "cat /mnt/user-data/outputs/results.csv\n", append=True)

    stored = (tmp_path / "workspace" / "collect").read_text(encoding="utf-8")
    assert stored.startswith("#!/bin/bash\n")
    assert "/mnt/user-data" not in stored


@pytest.mark.parametrize(
    "name,content",
    [
        ("plot.py", 'OUTPUT = "/mnt/user-data/outputs/results.csv"\n'),
        ("run.sh", 'export PATH="/mnt/user-data/workspace/bin:/mnt/user-data/outputs/bin"\n'),
        ("run", "#!/bin/bash\ncat /mnt/user-data/outputs/results.csv\n"),
    ],
    ids=["python", "shell", "shebang-without-suffix"],
)
def test_executable_content_still_translates_paths(tmp_path, name, content):
    """PR #1935: a script must see host paths, or ``bash``/``python`` cannot open them."""
    sandbox = _sandbox(tmp_path)
    virtual = f"/mnt/user-data/workspace/{name}"

    sandbox.write_file(virtual, content)

    stored = (tmp_path / "workspace" / name).read_text(encoding="utf-8")
    assert "/mnt/user-data" not in stored
    assert Path(os.path.realpath(tmp_path)).as_posix() in stored
    # The reverse translation keeps the agent's view consistent with what it wrote.
    assert sandbox.read_file(virtual) == content


@pytest.mark.parametrize("name", ["report.awk", "cleanup.sed", "plugin.lua", "plot.r", "run.tcl", "REPORT.AWK"])
def test_interpreter_script_suffixes_translate_container_paths(tmp_path, name):
    """``awk -f`` and friends open the script itself, so its container paths need host paths."""
    sandbox = _sandbox(tmp_path)
    content = 'BEGIN { while ((getline line < "/mnt/user-data/outputs/data.txt") > 0) print line }\n'
    virtual = f"/mnt/user-data/workspace/{name}"

    sandbox.write_file(virtual, content)

    stored = (tmp_path / "workspace" / name).read_text(encoding="utf-8")
    assert "/mnt/user-data" not in stored
    assert Path(os.path.realpath(tmp_path)).as_posix() in stored
    assert sandbox.read_file(virtual) == content


@pytest.mark.skipif(shutil.which("awk") is None, reason="awk is not installed")
def test_awk_script_written_through_the_sandbox_still_runs(tmp_path):
    """An AWK program that opens a container path must run through ``execute_command``."""
    sandbox = _sandbox(tmp_path)
    (tmp_path / "outputs" / "data.txt").write_text("alpha\nbeta\n", encoding="utf-8")
    sandbox.write_file(
        "/mnt/user-data/workspace/report.awk",
        'BEGIN { while ((getline line < "/mnt/user-data/outputs/data.txt") > 0) print line }\n',
    )

    output = sandbox.execute_command("awk -f /mnt/user-data/workspace/report.awk /dev/null", timeout=20)

    assert "alpha" in output
    assert "beta" in output


def test_append_does_not_read_the_destination_before_classifying(tmp_path, monkeypatch):
    """A ``.py`` suffix already settles it, so the append must not open the file for reading."""
    sandbox = _sandbox(tmp_path)
    virtual = "/mnt/user-data/outputs/events.py"
    sandbox.write_file(virtual, "print('ready')\n")
    destination = os.path.realpath(tmp_path / "outputs" / "events.py")
    real_open = builtins.open

    def guard_reads(file, mode="r", *args, **kwargs):
        reads_only = not set(mode) & set("wa+")
        if isinstance(file, str) and reads_only and os.path.realpath(file) == destination:
            raise AssertionError("append opened the destination for reading")
        return real_open(file, mode, *args, **kwargs)

    monkeypatch.setattr(builtins, "open", guard_reads)

    sandbox.write_file(virtual, "print('appended')\n", append=True)

    assert (tmp_path / "outputs" / "events.py").read_text(encoding="utf-8") == "print('ready')\nprint('appended')\n"


@pytest.mark.skipif(not hasattr(os, "mkfifo"), reason="named pipes are a POSIX primitive")
@pytest.mark.parametrize("name", ["events.py", "events"], ids=["suffix-settles-the-answer", "extensionless-shebang-sniff"])
def test_append_to_a_named_pipe_does_not_block_while_classifying(tmp_path, name):
    """Classification must not wait on the destination, whichever shape settles it.

    A ``.py`` suffix settles the answer with no I/O; an extensionless name
    reaches the shebang sniff, which has to open the destination without
    waiting and reject the non-regular handle before reading. Reading a FIFO
    waits for the other end and for data; the append used to complete as soon
    as a reader was connected.
    """
    sandbox = _sandbox(tmp_path)
    fifo = tmp_path / "outputs" / name
    os.mkfifo(fifo)
    virtual = f"/mnt/user-data/outputs/{name}"
    received: list[str] = []

    def read_pipe() -> None:
        with open(fifo, encoding="utf-8") as reader:
            received.append(reader.read())

    reader_thread = threading.Thread(target=read_pipe, daemon=True)
    reader_thread.start()

    errors: list[BaseException] = []

    def append_pipe() -> None:
        try:
            sandbox.write_file(virtual, "ready\n", append=True)
        except BaseException as exc:  # noqa: BLE001 - surfaced by the assertion below
            errors.append(exc)

    writer_thread = threading.Thread(target=append_pipe, daemon=True)
    writer_thread.start()
    writer_thread.join(5)

    assert not writer_thread.is_alive(), "append blocked while classifying the destination"
    assert errors == []
    reader_thread.join(5)
    assert received == ["ready\n"]