confluence_web/tests/integration/test_real_image.py

"""Backend + real pi runtime image with scripted model and Confluence peers.

Proves the actual bridge peers connect: real SDK turns, tool arguments and
results, collection ordering, exact artifact bytes, cleanup, cancellation,
timeouts, busy gating, and container isolation canaries. No network, no
credentials: Confluence is a scripted transport and the model is scripted.
"""

from __future__ import annotations

import asyncio
import json
import time
from pathlib import Path

import httpx
import pytest
from httpx import ASGITransport

from backend.app import create_app, SESSION_COOKIE_NAME
from backend.artifacts import ArtifactStore
from backend.containers import DockerContainerManager
from backend.errors import AppError, BusyError, ModelOutputLimitError, QueryTimeoutError, UpstreamFailedError
from backend.runner import QueryRunner
from backend.settings import Settings

from tests.integration.conftest import (
    APPROVED_ORIGIN,
    CANARY_PAT,
    CHECKLIST_BYTES,
    DENIED_PAGE_ID,
    FINAL_ANSWER,
    PAGE_ID,
    PAGE_URL,
    RUNTIME_IMAGE,
    ScriptedModelAdapter,
    example_script,
    make_confluence_factory,
    requires_runtime_image,
    text_turn,
    tool_message_text,
    tool_turn,
)

pytestmark = [pytest.mark.integration, requires_runtime_image()]

EXPECTED_TOOLS = {"confluence_search", "confluence_view", "confluence_list_spaces", "bash", "read", "write", "edit"}


def make_settings(tmp_path: Path, **overrides) -> Settings:
    base = dict(
        runtime_image=RUNTIME_IMAGE,
        approved_confluence_origins=[APPROVED_ORIGIN],
        query_timeout_seconds=120.0,
        cleanup_timeout_seconds=10.0,
        container_label_value=f"integration-{tmp_path.name[-8:]}",
        artifact_storage_dir=tmp_path / "artifacts",
    )
    base.update(overrides)
    return Settings(**base)


def make_runner(tmp_path: Path, script, **overrides):
    settings = make_settings(tmp_path, **overrides)
    store = ArtifactStore(settings.artifact_storage_dir)
    mgr = DockerContainerManager(settings)
    adapter = ScriptedModelAdapter(script)
    calls = []
    runner = QueryRunner(
        settings=settings,
        container_manager=mgr,
        artifact_store=store,
        model_adapter=adapter,
        confluence_client_factory=make_confluence_factory(calls),
    )
    return runner, settings, store, mgr, adapter, calls


async def assert_no_containers(mgr: DockerContainerManager, settings: Settings) -> None:
    code, out, _ = await mgr._exec_docker(
        ["ps", "-a", "-q", "--filter", f"label={settings.container_label_key}={settings.container_label_value}"]
    )
    assert code == 0
    assert out.strip() == "", "no application container may remain after the run"


async def run_query(runner: QueryRunner, prompt: str = "How do I deploy service X? Create a checklist file too.", **kwargs):
    return await runner.run(
        prompt=prompt,
        confluence_url=APPROVED_ORIGIN,
        confluence_pat=CANARY_PAT,
        session_id=kwargs.pop("session_id", "cw_sess_integration"),
        **kwargs,
    )


# --- shared example ---------------------------------------------------------------

@pytest.mark.asyncio
async def test_shared_example_end_to_end(tmp_path: Path):
    runner, settings, store, mgr, adapter, calls = make_runner(tmp_path, example_script())
    started = time.monotonic()
    result = await run_query(runner)
    elapsed = time.monotonic() - started

    # Container removed before the result is returned.
    await assert_no_containers(mgr, settings)

    # Authoritative history: exactly the two Confluence calls, in start order.
    hist = result["tool_history"]
    assert [h["tool"] for h in hist] == ["confluence_search", "confluence_view"]
    assert all(h["status"] == "success" and h["error"] is None and h["cache_hit"] is False for h in hist)
    assert hist[0]["parameters"]["query"] == "deploy service X"
    assert hist[0]["parameters"]["limit"] == 10 and hist[0]["parameters"]["offset"] == 0
    assert hist[0]["result"]["pages"][0]["page_id"] == PAGE_ID
    assert hist[0]["result"]["pages"][0]["url"] == PAGE_URL
    assert hist[0]["result"]["pagination"] == {"offset": 0, "limit": 10, "has_more": False}
    assert hist[1]["parameters"] == {"page_id": PAGE_ID}
    assert hist[1]["result"]["markdown"].strip() == "Deploy service X using the release checklist."
    assert hist[1]["result"]["truncated"] is False
    assert hist[0]["tool_call_id"] != hist[1]["tool_call_id"]
    assert all(h["tool_call_id"].startswith("a_") for h in hist)
    assert hist[0]["started_at"] <= hist[1]["started_at"]

    # Search matches never count; the read does.
    assert [p["page_id"] for p in result["pages_accessed"]] == [PAGE_ID]
    assert result["pages_accessed"][0]["url"] == PAGE_URL
    assert result["pages_accessed"][0]["accessed_at"]

    # Final answer and artifact.
    assert result["markdown"] == FINAL_ANSWER
    assert PAGE_URL in result["markdown"]
    assert len(result["artifacts"]) == 1
    art = result["artifacts"][0]
    assert art["name"] == "checklist.md" and art["size_bytes"] == 32
    dl = store.get_artifact_for_download(art["id"], session_id="cw_sess_integration")
    assert dl is not None
    path, name, size = dl
    assert (path.read_bytes(), name, size) == (CHECKLIST_BYTES, "checklist.md", 32)
    store.release_reader(art["id"])
    assert store.get_artifact_for_download(art["id"], session_id="cw_other") is None
    assert result["duration_seconds"] >= 0 and elapsed < 90
    assert result["warnings"] == []

    # Actual SDK turns reached the scripted model with the neutral contract shape.
    reqs = adapter.received_requests
    assert len(reqs) == 4
    assert all(r["system_instruction"] for r in reqs)
    assert all(m["role"] != "system" for r in reqs for m in r["messages"])
    assert {t["name"] for t in reqs[0]["tools"]} == EXPECTED_TOOLS
    assert all(isinstance(t["input_schema"], dict) for t in reqs[0]["tools"])
    assert reqs[0]["messages"][0]["role"] == "user"
    assert "deploy service X" in reqs[0]["messages"][0]["content"][0]["text"]

    # Turn 2 carries the assistant tool call and the search result as a tool message.
    roles = [m["role"] for m in reqs[1]["messages"]]
    assert roles == ["user", "assistant", "tool"]
    assistant = reqs[1]["messages"][1]
    call = next(c for c in assistant["content"] if c["type"] == "tool_call")
    assert call["name"] == "confluence_search" and call["arguments"] == {"query": "deploy service X"}
    assert isinstance(call["arguments"], dict)
    search_text = tool_message_text(reqs[1], call["id"])
    assert "Deployment Guide" in search_text and PAGE_URL in search_text
    assert reqs[1]["messages"][2]["is_error"] is False

    # Turn 4 carries the local write result; local tools stay out of history.
    write_call = next(c for m in reqs[3]["messages"] if m["role"] == "assistant" for c in m["content"] if c["type"] == "tool_call" and c["name"] == "write")
    assert write_call["arguments"]["path"] == "/work/artifacts/checklist.md"
    assert reqs[3]["messages"][-1]["role"] == "tool" and reqs[3]["messages"][-1]["is_error"] is False
    assert "write" not in [h["tool"] for h in hist]

    # Backend called Confluence exactly twice, with backend-constructed CQL.
    assert [c["path"] for c in calls] == ["/rest/api/content/search", f"/rest/api/content/{PAGE_ID}"]
    assert 'text ~ "deploy service X"' in calls[0]["params"]["cql"]

    # The PAT never reaches the model or the result.
    assert CANARY_PAT not in json.dumps(reqs) and CANARY_PAT not in json.dumps(result)


@pytest.mark.asyncio
async def test_no_artifacts_and_empty_search(tmp_path: Path):
    script = [
        tool_turn([("call_1", "confluence_search", {"query": "nothing here", "limit": 5})]),
        text_turn("No matching pages were found."),
    ]
    runner, settings, store, mgr, adapter, calls = make_runner(tmp_path, script)
    result = await run_query(runner, prompt="Find nothing here")
    await assert_no_containers(mgr, settings)
    assert result["artifacts"] == [] and result["pages_accessed"] == []
    assert len(result["tool_history"]) == 1
    entry = result["tool_history"][0]
    assert entry["status"] == "success"
    assert entry["result"] == {"pages": [], "pagination": {"offset": 0, "limit": 5, "has_more": False}}
    assert "[]" in tool_message_text(adapter.received_requests[1], "call_1")
    assert result["markdown"] == "No matching pages were found."


@pytest.mark.asyncio
async def test_cached_repeat_view_and_denied_page(tmp_path: Path):
    script = [
        tool_turn([("call_1", "confluence_view", {"page_id": PAGE_ID})]),
        tool_turn([("call_2", "confluence_view", {"page_id": PAGE_ID})]),
        tool_turn([("call_3", "confluence_view", {"page_id": DENIED_PAGE_ID})]),
        text_turn(f"Only [Deployment Guide]({PAGE_URL}) was readable."),
    ]
    runner, settings, store, mgr, adapter, calls = make_runner(tmp_path, script)
    result = await run_query(runner)
    await assert_no_containers(mgr, settings)

    hist = result["tool_history"]
    assert [h["tool"] for h in hist] == ["confluence_view"] * 3
    assert [h["cache_hit"] for h in hist] == [False, True, False]
    assert hist[1]["result"] == hist[0]["result"]
    assert len({h["tool_call_id"] for h in hist}) == 3
    assert hist[2]["status"] == "error" and hist[2]["result"] is None
    err = hist[2]["error"]
    assert set(err) == {"code", "message"} and err["code"] and len(err["message"].encode()) <= 1024
    assert "SECRET-UPSTREAM-BODY" not in json.dumps(result)
    assert "SECRET-UPSTREAM-BODY" not in json.dumps(adapter.received_requests)

    # Cache hit did not call upstream again; the denied read did.
    assert [c["path"] for c in calls] == [f"/rest/api/content/{PAGE_ID}", f"/rest/api/content/{DENIED_PAGE_ID}"]
    # One unique page, the denied one never counted.
    assert [p["page_id"] for p in result["pages_accessed"]] == [PAGE_ID]
    # The agent saw the failure as a recoverable tool error and continued.
    denied_msg = next(m for m in adapter.received_requests[3]["messages"] if m["role"] == "tool" and m["tool_call_id"] == "call_3")
    assert denied_msg["is_error"] is True
    assert result["markdown"].startswith("Only")


@pytest.mark.asyncio
async def test_empty_artifact_and_html_artifact(tmp_path: Path):
    script = [
        tool_turn([("call_1", "bash", {"command": "touch /work/artifacts/empty.txt; printf '<script>alert(1)</script>' > /work/artifacts/page.html"})]),
        text_turn("Created two files."),
    ]
    runner, settings, store, mgr, adapter, calls = make_runner(tmp_path, script)
    result = await run_query(runner)
    await assert_no_containers(mgr, settings)
    by_name = {a["name"]: a for a in result["artifacts"]}
    assert set(by_name) == {"empty.txt", "page.html"}
    assert by_name["empty.txt"]["size_bytes"] == 0
    dl = store.get_artifact_for_download(by_name["empty.txt"]["id"], session_id="cw_sess_integration")
    assert dl is not None and dl[0].read_bytes() == b"" and dl[2] == 0
    store.release_reader(by_name["empty.txt"]["id"])
    assert result["pages_accessed"] == [] and result["tool_history"] == []


# --- failure paths ---------------------------------------------------------------------

@pytest.mark.asyncio
async def test_model_upstream_failure_discards_exports(tmp_path: Path):
    script = [
        tool_turn([("call_1", "bash", {"command": "printf 'partial' > /work/artifacts/partial.txt"})]),
        UpstreamFailedError("scripted provider outage"),
    ]
    runner, settings, store, mgr, adapter, calls = make_runner(tmp_path, script)
    with pytest.raises(AppError) as excinfo:
        await run_query(runner)
    assert excinfo.value.code in {"upstream_failed", "execution_failed"}
    await assert_no_containers(mgr, settings)
    assert store.global_reserved_bytes == 0
    assert not any(p.is_file() for p in settings.artifact_storage_dir.rglob("*"))


@pytest.mark.asyncio
async def test_model_length_stop_is_output_limit(tmp_path: Path):
    script = [text_turn("Truncated answer that never finished", stop_reason="length")]
    runner, settings, store, mgr, adapter, calls = make_runner(tmp_path, script)
    with pytest.raises(AppError) as excinfo:
        await run_query(runner)
    assert isinstance(excinfo.value, ModelOutputLimitError), excinfo.value.code
    assert excinfo.value.status_code == 502
    await assert_no_containers(mgr, settings)


@pytest.mark.asyncio
async def test_empty_terminal_answer_fails(tmp_path: Path):
    runner, settings, store, mgr, adapter, calls = make_runner(tmp_path, [text_turn("")])
    with pytest.raises(AppError) as excinfo:
        await run_query(runner)
    assert excinfo.value.code == "execution_failed"
    await assert_no_containers(mgr, settings)


@pytest.mark.asyncio
async def test_timeout_during_model_work(tmp_path: Path):
    async def stall():
        await asyncio.sleep(120)
    runner, settings, store, mgr, adapter, calls = make_runner(tmp_path, [stall], query_timeout_seconds=8.0)
    started = time.monotonic()
    with pytest.raises(QueryTimeoutError):
        await run_query(runner)
    assert time.monotonic() - started < 40
    await assert_no_containers(mgr, settings)
    assert store.global_reserved_bytes == 0


@pytest.mark.asyncio
async def test_disconnect_cancels_and_busy_gate_holds(tmp_path: Path):
    async def stall():
        await asyncio.sleep(120)
    runner, settings, store, mgr, adapter, calls = make_runner(
        tmp_path, [stall] + example_script(), query_timeout_seconds=90.0
    )
    disconnected = False
    first = asyncio.create_task(run_query(runner, is_disconnected=lambda: disconnected))
    await asyncio.wait_for(adapter.started.wait(), 60)

    # Busy while the first run is active.
    with pytest.raises(BusyError):
        await run_query(runner, session_id="cw_second")

    disconnected = True
    with pytest.raises(BaseException) as excinfo:
        await first
    assert isinstance(excinfo.value, (asyncio.CancelledError, AppError)), type(excinfo.value)
    await assert_no_containers(mgr, settings)
    assert store.global_reserved_bytes == 0

    # After cleanup the gate admits a new run, which completes normally.
    result = await run_query(runner, session_id="cw_third")
    assert result["artifacts"][0]["size_bytes"] == 32
    await assert_no_containers(mgr, settings)


# --- isolation canaries under the actual image and launch flags ------------------------

PROBE = r"""
set +e
echo "UID=$(id -u) GID=$(id -g)"
echo "WORK=$(ls -A /work | tr '\n' ' ')"
echo "HOME_LS=$(ls -A "$HOME" | tr '\n' ' ')"
echo "TMP_LS=$(ls -A /tmp | tr '\n' ' ')"
echo "NET=$(ls /sys/class/net | tr '\n' ' ')"
echo "CAPEFF=$(grep CapEff /proc/self/status)"
echo "NNP=$(grep NoNewPrivs /proc/self/status)"
echo "ENV_BEGIN"; env; echo "ENV_END"
echo "PID1_ENVIRON=$(cat /proc/1/environ 2>&1 | tr '\0' ' ' | head -c 300)"
echo "PROCS=$(ps -eo pid,comm --no-headers | tr '\n' ';')"
echo "OPT_WRITABLE=$(touch /opt/agent/x 2>&1 || echo readonly)"
echo "ROOT_WRITABLE=$(touch /x 2>&1 || echo readonly)"
echo "DOCKER_SOCK=$(ls /var/run/docker.sock /run/docker.sock 2>&1)"
echo "HOST_FILES=$(ls /etc/shadow ~/.pi /root 2>&1 | tr '\n' ' ')"
echo "MOUNTS=$(awk '{print $2":"$3}' /proc/mounts | tr '\n' ' ')"
"""


@pytest.mark.asyncio
async def test_fresh_workspace_and_canaries_across_runs(tmp_path: Path):
    seed = [
        tool_turn([("call_1", "bash", {"command": "echo scratch > /work/scratch.txt; echo m > $HOME/marker; echo t > /tmp/tmarker; echo ok"})]),
        text_turn("seeded"),
    ]
    runner, settings, store, mgr, adapter, calls = make_runner(tmp_path, seed + [
        tool_turn([("call_2", "bash", {"command": PROBE})]),
        text_turn("probed"),
    ])
    first = await run_query(runner)
    assert first["markdown"] == "seeded"
    second = await run_query(runner)
    assert second["markdown"] == "probed"
    await assert_no_containers(mgr, settings)

    probe_out = tool_message_text(adapter.received_requests[3], "call_2")
    assert "UID=10001 GID=10001" in probe_out
    assert "scratch.txt" not in probe_out and "marker" not in probe_out
    assert "WORK=artifacts" in probe_out
    assert "NET=lo" in probe_out
    assert "CapEff:\t0000000000000000" in probe_out
    assert "NoNewPrivs:\t1" in probe_out
    env_block = probe_out.split("ENV_BEGIN")[1].split("ENV_END")[0]
    env_names = {line.split("=", 1)[0] for line in env_block.strip().splitlines() if "=" in line}
    assert env_names <= {"HOME", "LANG", "PATH", "PWD", "SHLVL", "_", "OLDPWD", "NODE_OPTIONS", "AGENT_WORK_DIR"}, env_names
    assert CANARY_PAT not in probe_out
    assert "MODEL_API_KEY" not in probe_out and "CONFLUENCE" not in env_block
    assert "PID1_ENVIRON=cat: /proc/1/environ: Permission denied" in probe_out or "PID1_ENVIRON=" in probe_out
    assert "Read-only file system" in probe_out.split("OPT_WRITABLE=")[1].split("ROOT_WRITABLE=")[0]
    assert "Read-only file system" in probe_out.split("ROOT_WRITABLE=")[1].split("DOCKER_SOCK=")[0]
    assert "No such file" in probe_out.split("DOCKER_SOCK=")[1].split("\n")[0]
    host_files = probe_out.split("HOST_FILES=")[1].split("MOUNTS=")[0]
    assert "cannot access '/home/agent/.pi'" in host_files and "'/root': Permission denied" in host_files
    assert "/work:tmpfs" in probe_out and "/tmp:tmpfs" in probe_out and "/home/agent:tmpfs" in probe_out
    assert CANARY_PAT not in json.dumps(adapter.received_requests)


# --- HTTP surface with the real image --------------------------------------------------

@pytest.mark.asyncio
async def test_http_query_and_download_ownership(tmp_path: Path):
    settings = make_settings(tmp_path)
    store = ArtifactStore(settings.artifact_storage_dir)
    mgr = DockerContainerManager(settings)
    adapter = ScriptedModelAdapter(example_script())
    app = create_app(
        settings=settings,
        container_manager=mgr,
        artifact_store=store,
        model_adapter=adapter,
        confluence_client_factory=make_confluence_factory(),
    )
    transport = ASGITransport(app=app)
    origin = "http://testserver"
    async with httpx.AsyncClient(transport=transport, base_url=origin) as a, httpx.AsyncClient(transport=transport, base_url=origin) as b:
        boot = await a.get("/")
        assert boot.status_code == 200 and SESSION_COOKIE_NAME in boot.cookies
        verify = await a.post("/api/v1/auth/verify", json={"url": APPROVED_ORIGIN, "pat": CANARY_PAT}, headers={"Origin": origin})
        assert verify.status_code == 200 and verify.json() == {"valid": True}
        assert verify.headers["cache-control"] == "no-store"

        resp = await a.post(
            "/api/v1/query",
            json={"prompt": "How do I deploy service X?", "credentials": {"url": APPROVED_ORIGIN, "pat": CANARY_PAT}},
            headers={"Origin": origin},
            timeout=120,
        )
        assert resp.status_code == 200, resp.text
        assert resp.headers["cache-control"] == "no-store"
        body = resp.json()
        assert set(body) == {"session_id", "markdown", "pages_accessed", "tool_history", "artifacts", "warnings", "duration_seconds"}
        await assert_no_containers(mgr, settings)
        art = body["artifacts"][0]
        assert art["id"] != body["session_id"]

        dl = await a.get(f"/api/v1/artifacts/{art['id']}")
        assert dl.status_code == 200 and dl.content == CHECKLIST_BYTES
        assert dl.headers["content-type"].startswith("application/octet-stream")
        assert dl.headers["content-disposition"].startswith("attachment")
        assert dl.headers["x-content-type-options"] == "nosniff"
        assert dl.headers["cache-control"] == "no-store"

        # Session B, unknown ID and expired ID are indistinguishable 404s.
        await b.get("/")
        wrong = await b.get(f"/api/v1/artifacts/{art['id']}")
        unknown = await a.get("/api/v1/artifacts/does-not-exist")
        store._committed[art["id"]].expires_at_ts = 0  # force expiry
        expired = await a.get(f"/api/v1/artifacts/{art['id']}")
        for r in (wrong, unknown, expired):
            assert r.status_code == 404
            assert r.json() == {"error": {"code": "artifact_not_found", "message": unknown.json()["error"]["message"]}}
            assert r.headers["cache-control"] == "no-store"
        assert wrong.content == unknown.content == expired.content


# --- generated boundaries through the whole stack ---------------------------------------

@pytest.mark.asyncio
async def test_large_prompt_and_answer_round_trip(tmp_path: Path):
    """Exactly 16 MiB decoded prompt with heavy JSON escaping, and a multi-MiB answer."""
    unit = 'line "quoted" \\ back\ttab é中\U0001f600 \x01\n'  # quotes, backslash, control chars, multibyte
    prompt = unit * (16 * 1024 * 1024 // len(unit.encode("utf-8")))
    prompt += "x" * (16 * 1024 * 1024 - len(prompt.encode("utf-8")))
    assert len(prompt.encode("utf-8")) == 16 * 1024 * 1024
    section = "## Section\n\n" + ("Deploy service X carefully. " * 40) + "\n\n```\ncode \"block\" \\ \n```\n\n"
    answer = "# Big answer\n\n" + section * (4 * 1024 * 1024 // len(section.encode("utf-8"))) + f"\nSee [Deployment Guide]({PAGE_URL}).\n"
    script = [
        tool_turn([("call_1", "bash", {"command": "wc -c /work/artifacts 2>/dev/null; echo ok"})]),
        text_turn(answer, in_tok=1000, out_tok=1000),
    ]
    runner, settings, store, mgr, adapter, calls = make_runner(tmp_path, script)
    started = time.monotonic()
    result = await run_query(runner, prompt=prompt)
    elapsed = time.monotonic() - started
    await assert_no_containers(mgr, settings)
    assert result["markdown"] == answer
    first_user = adapter.received_requests[0]["messages"][0]["content"][0]["text"]
    assert first_user == prompt, "prompt must reach the model byte-for-byte through the bridge"
    assert elapsed < 150, elapsed

    # One byte over the decoded limit is rejected before any container starts.
    with pytest.raises(AppError) as excinfo:
        await run_query(runner, prompt=prompt + "y")
    assert excinfo.value.code == "invalid_input"
    await assert_no_containers(mgr, settings)