tests/test_pipeline_e2e.py

"""End-to-end pipeline tests (app/pipeline.py) with a mocked LLM.

The forge scripts, validator, generator and esbuild bundling all run for
real; only the vision-LLM call is substituted.
"""

from __future__ import annotations

import asyncio
import dataclasses
import json
import shutil
import subprocess
import sys
from pathlib import Path

import pytest

from app import pipeline as pipeline_module
from app.gallery import GalleryError, GalleryStore
from app.pipeline import JobRegistry, run_job
from tests.conftest import MockLLM

REPO_ROOT = Path(__file__).resolve().parents[1]
FIXTURES = Path(__file__).resolve().parent / "fixtures"

ESBUILD = REPO_ROOT / "node_modules" / ".bin" / "esbuild"
needs_bundler = pytest.mark.skipif(
    not ESBUILD.exists(),
    reason="esbuild not installed (run npm ci)")


def shallow_spec_text(tmp_path: Path) -> str:
    """A structurally valid but strict-quality-failing starter spec."""
    assessment = tmp_path / "a.json"
    spec = tmp_path / "s.json"
    forge = REPO_ROOT / "forge"
    subprocess.run(
        [sys.executable, str(forge / "stage2_spec" / "new_pre_spec_assessment.py"),
         "Widget", "--complexity", "simple", "--out", str(assessment)],
        check=True, capture_output=True)
    subprocess.run(
        [sys.executable, str(forge / "stage2_spec" / "new_sculpt_spec.py"),
         "Widget", "--assessment", str(assessment), "--out", str(spec)],
        check=True, capture_output=True)
    return spec.read_text(encoding="utf-8")


async def run(tmp_path, make_settings, replies, raw_upload, **env):
    from app.pipeline import JobRegistry as JR

    registry = JR(tmp_path / "runs")
    job = registry.create()
    settings = make_settings(runs_dir=str(tmp_path / "runs"), **env)
    await run_job(job, raw_upload=raw_upload, object_hint=None,
                  settings=settings, llm=MockLLM(replies))
    return job


@pytest.mark.asyncio
@needs_bundler
async def test_happy_path_full_artifacts(tmp_path, make_settings,
                                         ref_png_bytes, canned_spec_text):
    job = await run(tmp_path, make_settings, [canned_spec_text], ref_png_bytes)
    assert job.status == "done", job.error

    result = job.result
    assert result["targetName"] == "Mug"
    assert result["generatedPass"] == "hosted-preview"
    assert result["generationMode"] == "hosted-unreviewed-preview"
    assert result["reviewStatus"] == "unreviewed"
    assert result["completedPasses"] == []
    assert result["components"] >= 2
    assert result["honesty"]  # honesty notes always present
    assert result["shareRequested"] is True
    assert result["shared"] is True
    assert result["galleryItem"]["id"] == job.id
    assert (tmp_path / "gallery" / job.id / "item.json").is_file()
    assert not list(job.dir.glob(".gallery-publish-*"))
    assert not list((tmp_path / "gallery").glob(".publishing-*"))

    factory = (job.dir / "factory.ts").read_text(encoding="utf-8")
    assert "export function createMugModel" in factory
    assert "sculptRuntime" in factory

    bundle = job.dir / "model.bundle.js"
    assert bundle.exists() and bundle.stat().st_size > 100_000  # three included

    standalone = (job.dir / "standalone.html").read_text(encoding="utf-8")
    assert "await import(url)" in standalone
    assert "mod.mountViewer" in standalone
    assert "img2threejs" in standalone  # attribution header

    spec = json.loads((job.dir / "spec.json").read_text(encoding="utf-8"))
    assert spec["reviewHistory"] == []
    assert spec["sculptPipeline"]["passGateMode"] == "locked-sequential"
    compile_spec = json.loads(
        (job.dir / "compile-spec.json").read_text(encoding="utf-8"))
    assert compile_spec["reviewHistory"] == []
    assert compile_spec["sculptPipeline"]["passGateMode"] == "hosted-unreviewed-preview"


@pytest.mark.asyncio
@needs_bundler
async def test_gallery_outage_preserves_completed_result(
    tmp_path, make_settings, ref_png_bytes, canned_spec_text
):
    class UnavailableGallery(GalleryStore):
        def publish(self, **kwargs):
            raise GalleryError("storage temporarily unavailable")

    registry = JobRegistry(tmp_path / "runs")
    job = registry.create()
    settings = make_settings(runs_dir=str(tmp_path / "runs"))
    await run_job(
        job,
        raw_upload=ref_png_bytes,
        object_hint=None,
        settings=settings,
        llm=MockLLM([canned_spec_text]),
        gallery_store=UnavailableGallery(tmp_path / "gallery"),
    )

    assert job.status == "done", job.error
    assert job.result["shareRequested"] is True
    assert job.result["shared"] is False
    assert job.result["galleryItem"] is None
    assert "model is ready" in job.result["publicationWarning"]
    assert (job.dir / "model.bundle.js").is_file()
    publishing = [
        event for event in job.events if event["stage"] == "publishing"
    ]
    assert publishing[-1]["status"] == "done"
    assert publishing[-1]["data"]["warning"] is True

    stages = [e["stage"] for e in job.events]
    assert stages[0] == "intake"
    assert "spec-authoring" in stages and "generation" in stages
    assert "publishing" in stages
    assert job.events[-1]["stage"] == "done" and job.events[-1]["status"] == "done"


@pytest.mark.asyncio
async def test_opaque_stage_wait_emits_elapsed_feedback(tmp_path, monkeypatch):
    registry = JobRegistry(tmp_path / "runs")
    job = registry.create()
    monkeypatch.setattr(pipeline_module, "STAGE_PROGRESS_INTERVAL_S", 0.01)

    result = await pipeline_module._await_with_feedback(
        job,
        stage="generation",
        awaitable=asyncio.sleep(0.035, result="ready"),
        message="Still generating ({elapsed}s in this stage, {total}s total).",
    )

    assert result == "ready"
    progress = [
        event for event in job.events
        if event["stage"] == "generation" and event["status"] == "progress"
    ]
    assert len(progress) >= 2
    assert all("elapsedSeconds" in event["data"] for event in progress)
    assert all("%" not in event["message"] for event in progress)


@pytest.mark.asyncio
@needs_bundler
async def test_gallery_publication_timeout_preserves_result_and_kills_worker(
    tmp_path,
    make_settings,
    ref_png_bytes,
    canned_spec_text,
    monkeypatch,
):
    def hanging_worker(*_args):
        return [sys.executable, "-c", "import time; time.sleep(60)"]

    monkeypatch.setattr(
        pipeline_module, "_gallery_worker_command", hanging_worker
    )
    settings = dataclasses.replace(
        make_settings(runs_dir=str(tmp_path / "runs")),
        gallery_publish_timeout_s=0.05,
    )
    registry = JobRegistry(tmp_path / "runs")
    job = registry.create()

    await run_job(
        job,
        raw_upload=ref_png_bytes,
        object_hint=None,
        settings=settings,
        llm=MockLLM([canned_spec_text]),
    )

    assert job.status == "done", job.error
    assert job.result["shared"] is False
    assert "publicationWarning" in job.result
    assert not (tmp_path / "gallery" / job.id).exists()
    assert not list((tmp_path / "gallery").glob(".publishing-*"))
    assert not list(job.dir.glob(".gallery-publish-*"))


@pytest.mark.asyncio
async def test_esbuild_is_killed_and_reaped_on_outer_cancellation(
    tmp_path, make_settings, monkeypatch
):
    executable = tmp_path / "fake-esbuild"
    executable.write_text("# test executable\n", encoding="utf-8")
    executable.chmod(0o755)
    monkeypatch.setattr(pipeline_module, "REPO_ROOT", tmp_path)
    settings = dataclasses.replace(
        make_settings(runs_dir=str(tmp_path / "runs")),
        esbuild_entry="fake-esbuild",
    )
    job = JobRegistry(tmp_path / "runs").create()
    entry = job.dir / "entry.js"
    entry.write_text("export {};\n", encoding="utf-8")
    started = asyncio.Event()

    class FakeProcess:
        returncode = None
        killed = False
        waited = False

        async def communicate(self):
            started.set()
            await asyncio.Future()

        def kill(self):
            self.killed = True
            self.returncode = -9

        async def wait(self):
            self.waited = True
            return self.returncode

    process = FakeProcess()

    async def fake_subprocess(*_args, **_kwargs):
        return process

    monkeypatch.setattr(
        pipeline_module.asyncio,
        "create_subprocess_exec",
        fake_subprocess,
    )
    task = asyncio.create_task(
        pipeline_module._run_esbuild(
            job, entry, job.dir / "model.bundle.js", settings
        )
    )
    await started.wait()
    task.cancel()
    with pytest.raises(asyncio.CancelledError):
        await task

    assert process.killed is True
    assert process.waited is True


@pytest.mark.asyncio
@needs_bundler
async def test_validator_feedback_repair_round(tmp_path, make_settings,
                                               ref_png_bytes, canned_spec_text):
    shallow = shallow_spec_text(tmp_path)
    job = await run(tmp_path, make_settings, [shallow, canned_spec_text], ref_png_bytes)
    assert job.status == "done", job.error
    mock_calls = 2  # first spec rejected, repair round accepted
    assert len(job.events) > 0
    # The repair prompt must carry the validator's actual error strings.
    # (mock recorded every message list)


@pytest.mark.asyncio
async def test_persistent_validation_failure_is_honest(tmp_path, make_settings,
                                                       ref_png_bytes):
    shallow = shallow_spec_text(tmp_path)
    job = await run(tmp_path, make_settings, [shallow], ref_png_bytes,
                    SPEC_REPAIR_ROUNDS=2)
    assert job.status == "error"
    assert job.error["code"] == "spec_validation_failed"
    assert job.error["detail"]["errors"]  # validator output surfaced
    assert not (job.dir / "factory.ts").exists()  # nothing fabricated
    assert not (job.dir / "model.bundle.js").exists()


@pytest.mark.asyncio
async def test_llm_garbage_json_is_honest(tmp_path, make_settings, ref_png_bytes):
    job = await run(tmp_path, make_settings, ["definitely not json"], ref_png_bytes)
    assert job.status == "error"
    assert job.error["code"] == "spec_validation_failed"
    assert "parseable JSON" in job.error["detail"]["errors"][0]
    assert not (job.dir / "factory.ts").exists()


@pytest.mark.asyncio
@needs_bundler
async def test_bad_json_enters_repair_round(tmp_path, make_settings,
                                            ref_png_bytes, canned_spec_text):
    job = await run(
        tmp_path, make_settings,
        ["definitely not json", canned_spec_text], ref_png_bytes,
        SPEC_REPAIR_ROUNDS=1,
    )
    assert job.status == "done", job.error
    assert any(
        event["stage"] == "spec-authoring" and event["status"] == "progress"
        for event in job.events
    )


@pytest.mark.asyncio
@needs_bundler
async def test_hosted_primitive_error_enters_repair_round(
    tmp_path, make_settings, ref_png_bytes, canned_spec, canned_spec_text,
):
    unsupported = json.loads(json.dumps(canned_spec))
    unsupported["componentTree"][0]["primitive"] = "metaball"
    job = await run(
        tmp_path, make_settings,
        [json.dumps(unsupported), canned_spec_text], ref_png_bytes,
        SPEC_REPAIR_ROUNDS=1,
    )
    assert job.status == "done", job.error
    progress = [
        event for event in job.events
        if event["stage"] == "spec-authoring" and event["status"] == "progress"
    ]
    assert progress
    assert "unsupported primitive" in json.dumps(progress[0])


@pytest.mark.asyncio
async def test_unsuitable_image_verdict_is_honest(tmp_path, make_settings,
                                                  ref_png_bytes, canned_spec):
    rejected = dict(canned_spec)
    rejected["suitability"] = "reject"
    job = await run(tmp_path, make_settings, [json.dumps(rejected)], ref_png_bytes)
    assert job.status == "error"
    assert job.error["code"] == "unsuitable_image"
    assert not (job.dir / "factory.ts").exists()


@pytest.mark.asyncio
async def test_bad_image_bytes_rejected_at_intake(tmp_path, make_settings):
    job = await run(tmp_path, make_settings, ["unused"], b"<html>nope</html>")
    assert job.status == "error"
    assert job.error["stage"] == "intake"
    assert job.error["code"] in {"not_an_image", "svg_rejected"}
    gallery = tmp_path / "gallery"
    assert not (gallery / job.id).exists()


@pytest.mark.asyncio
@needs_bundler
async def test_fenced_llm_reply_is_repaired(tmp_path, make_settings,
                                            ref_png_bytes, canned_spec_text):
    fenced = f"Here is the spec:\n```json\n{canned_spec_text}\n```"
    job = await run(tmp_path, make_settings, [fenced], ref_png_bytes)
    assert job.status == "done", job.error


@pytest.mark.asyncio
@needs_bundler
async def test_concurrent_jobs_isolated(tmp_path, make_settings,
                                        ref_png_bytes, canned_spec_text):
    import asyncio

    registry = JobRegistry(tmp_path / "runs")
    settings = make_settings(runs_dir=str(tmp_path / "runs"))

    async def one():
        job = registry.create()
        await run_job(job, raw_upload=ref_png_bytes, object_hint=None,
                      settings=settings, llm=MockLLM([canned_spec_text]))
        return job

    jobs = await asyncio.gather(one(), one(), one())
    assert {j.status for j in jobs} == {"done"}
    assert len({j.id for j in jobs}) == 3
    for job in jobs:
        assert (job.dir / "factory.ts").exists()