import hashlib
from datetime import UTC, datetime
from io import BytesIO

import pytest
from PIL import Image

from pigway_frame_flow.providers.photo_sources import PhotoSourceInput


@pytest.fixture
def library_source(service, settings, legacy_photo_engine):
    root = settings.data_dir.parent / "photos"
    root.mkdir()
    settings.photo_root = root
    Image.new("RGB", (900, 600), (25, 100, 220)).save(root / "photo.jpg")
    for source_id in ["family", "travel"]:
        service.photo_sources.save(
            source_id, PhotoSourceInput(name=source_id, kind="local", location=".")
        )
    return "family"


def create_plan(
    client, source="family", name="Watch", width=410, height=502, **changes
):
    response = client.post(
        "/api/v1/generation-plans",
        json={
            "source_id": source,
            "name": name,
            "width": width,
            "height": height,
            **changes,
        },
    )
    assert response.status_code == 200, response.text
    return response.json()


def bind(client, terminal, source, plan):
    response = client.put(
        "/api/v1/library-terminals/" + terminal + "/binding",
        json={"source_id": source, "plan_id": plan["id"]},
    )
    assert response.status_code == 200, response.text


def import_picture(client):
    response = client.post(
        "/api/v1/source-libraries/family/import", json={"item_id": "photo.jpg"}
    )
    assert response.status_code == 200, response.text
    return response.json()


def test_single_binding_and_cross_library_guards(client, service, library_source):
    first = create_plan(client)
    other = create_plan(client, "travel")
    bind(client, "watch", "family", first)
    assert (
        client.put(
            "/api/v1/library-terminals/watch/binding",
            json={"source_id": "travel", "plan_id": first["id"]},
        ).status_code
        == 409
    )
    bind(client, "watch", "travel", other)
    assert service.db.one("SELECT COUNT(*) AS n FROM terminal_bindings")["n"] == 1
    assert client.delete("/api/v1/generation-plans/" + other["id"]).status_code == 409
    assert client.delete("/api/v1/photo-sources/travel").status_code == 409
    assert (
        client.put(
            "/api/v1/library-terminals/living-room/binding",
            json={"source_id": "family", "plan_id": first["id"]},
        ).status_code
        == 422
    )
    assert (
        client.put(
            "/api/v1/generation-plans/" + other["id"],
            json={
                k: v
                for k, v in {**other, "width": 400}.items()
                if k not in ("id", "archived")
            },
        ).status_code
        == 422
    )
    assert (
        client.put(
            "/api/v1/generation-plans/" + first["id"],
            json={"name": "Move", "source_id": "travel"},
        ).status_code
        == 409
    )


def test_original_and_versions_share_picture_with_immutable_snapshots(
    client, service, library_source
):
    picture = import_picture(client)
    assert import_picture(client)["id"] == picture["id"]
    first = create_plan(client)
    create_plan(client, name="Frame", width=1200, height=1600)
    versions = client.post(
        "/api/v1/source-images/" + picture["id"] + "/generate", json={}
    ).json()
    assert len(versions) == 2
    assert all(v["source"] == "generated" for v in versions)
    assert all(picture["name"].removesuffix(".jpg") in v["name"] for v in versions)
    assert len({v["source_image_id"] for v in versions}) == 1
    repeated = client.post(
        "/api/v1/source-images/" + picture["id"] + "/generate", json={}
    ).json()
    assert {v["asset_id"] for v in versions} == {v["asset_id"] for v in repeated}
    payload = {k: v for k, v in first.items() if k not in ("id", "archived")}
    assert (
        client.put(
            "/api/v1/generation-plans/" + first["id"], json={**payload, "quality": 70}
        ).status_code
        == 200
    )
    updated = client.post(
        "/api/v1/source-images/" + picture["id"] + "/generate",
        json={"plan_id": first["id"]},
    ).json()[0]
    assert updated["asset_id"] not in {v["asset_id"] for v in versions}
    assert updated["plan"]["quality"] == 70
    assert (
        next(v for v in versions if v["plan_id"] == first["id"])["plan"]["quality"]
        == 85
    )
    other = create_plan(client, "travel")
    assert (
        client.post(
            "/api/v1/source-images/" + picture["id"] + "/generate",
            json={"plan_id": other["id"]},
        ).status_code
        == 409
    )
    assert client.get("/api/v1/assets").json() == []
    assert client.delete("/api/v1/assets/" + picture["id"]).status_code == 409
    assert (
        client.post(
            "/api/v1/devices/living-room/display",
            json={"content": {"type": "image", "asset_id": picture["id"]}},
        ).status_code
        == 409
    )
    assert (
        client.put(
            "/api/v1/source-images/" + picture["id"] + "/star", json={"starred": True}
        ).status_code
        == 200
    )
    assert (
        client.put(
            "/api/v1/source-images/" + picture["id"] + "/exclusion",
            json={"excluded": True},
        ).status_code
        == 409
    )
    assert client.delete("/api/v1/source-images/" + picture["id"]).status_code == 204
    assert service.db.one("SELECT COUNT(*) AS n FROM library_versions")["n"] == 0
    assert (service.settings.photo_root / "photo.jpg").exists()


def test_pipeline_selects_once_for_frame_and_watch_and_shared_plan(
    client, service, library_source, monkeypatch
):
    frame = create_plan(client, name="Frame", width=1200, height=1600, caption=True)
    watch = create_plan(client)
    service.registry.drivers["second-frame"] = service.registry.get("living-room")
    for terminal in ["living-room", "second-frame"]:
        bind(client, terminal, "family", frame)
    bind(client, "watch", "family", watch)

    def unavailable_frame(*args, **kwargs):
        raise AssertionError("Source generation must not read a frame device")

    original_device = service.registry.get("living-room").device
    monkeypatch.setattr(
        service.registry.get("living-room"), "device", unavailable_frame
    )
    read = service.photo_sources.read
    calls = []

    def source_read(*args, **kwargs):
        calls.append(1)
        data, name, _ = read(*args, **kwargs)
        return data, name, {"title": "Caption", "date": "2026-10-23"}

    monkeypatch.setattr(service.photo_sources, "read", source_read)
    result = client.post("/api/v1/source-libraries/family/run", json={}).json()
    assert result["errors"] == []
    assert len(calls) == 1
    assert len(result["outputs"]) == 3
    assert len({v["asset_id"] for v in result["outputs"]}) == 2
    monkeypatch.setattr(service.registry.get("living-room"), "device", original_device)
    for terminal in ["living-room", "watch"]:
        page = client.get("/api/v1/library-terminals/" + terminal + "/images").json()
        assert page["total"] == 1
        version = page["items"][0]
        with Image.open(BytesIO(client.get(version["url"]).content)) as image:
            if terminal == "watch":
                assert image.size == (410, 502)
                assert image.getpixel((200, 490))[2] > 200
            else:
                assert image.getpixel((10, 1590))[0] > 240
    assert client.post(
        "/api/v1/library-terminals/watch/images/"
        + result["outputs"][-1]["asset_id"]
        + "/send"
    ).status_code in (409, 422)
    assert service.db.one("SELECT COUNT(*) AS n FROM operations")["n"] == 0


def test_scheduled_pipeline_links_independent_delivery_and_rechecks_binding(
    client, app, service, library_source, monkeypatch
):
    frame = create_plan(client, name="Frame", width=1200, height=1600)
    watch = create_plan(client)
    bind(client, "living-room", "family", frame)
    bind(client, "watch", "family", watch)
    saved = client.post(
        "/api/v1/jobs",
        json={
            "name": "Shared source",
            "schedule": {"type": "daily", "time": "07:00"},
            "source": {"type": "photo_source", "source_id": "family"},
            "library_pipeline": True,
        },
    ).json()
    now = datetime.now(UTC)
    service.db.execute(
        "UPDATE jobs SET next_run=? WHERE id=?", (now.isoformat(), saved["id"])
    )
    app.state.scheduler.tick(now)
    root = service.db.one("SELECT id FROM operations WHERE action='library-generate'")
    assert root and service.process_one()
    op = client.get("/api/v1/operations/" + root["id"]).json()
    assert op["status"] == "COMPLETED"
    assert len(op["result"]["outputs"]) == 2
    child = service.db.one(
        "SELECT id FROM operations WHERE client_id='library-terminal'"
    )
    assert child
    assert (
        service.db.one("SELECT parent_operation_id FROM terminal_deliveries")[
            "parent_operation_id"
        ]
        == root["id"]
    )
    client.delete("/api/v1/library-terminals/living-room/binding")
    writes = []
    monkeypatch.setattr(
        service.registry.get("living-room"), "upload", lambda *args: writes.append(args)
    )
    assert service.process_one()
    assert service.operation(child["id"])["status"] == "FAILED"
    assert writes == []
    assert client.get("/api/v1/library-terminals/watch/images").json()["total"] == 1


def test_preview_is_unsaved_and_references_prevent_deletion(
    client, service, library_source
):
    picture = import_picture(client)
    plan = create_plan(client, name="Frame", width=1200, height=1600)
    payload = {k: v for k, v in plan.items() if k not in ("id", "archived")}
    assert (
        client.post(
            "/api/v1/source-images/" + picture["id"] + "/preview", json=payload
        ).status_code
        == 200
    )
    assert service.db.one("SELECT COUNT(*) AS n FROM library_versions")["n"] == 0
    bind(client, "living-room", "family", plan)
    version = client.post(
        "/api/v1/source-images/" + picture["id"] + "/generate",
        json={"plan_id": plan["id"]},
    ).json()[0]
    assert (
        client.post(
            "/api/v1/library-terminals/living-room/images/"
            + version["asset_id"]
            + "/send"
        ).status_code
        == 200
    )
    assert client.delete("/api/v1/source-images/" + picture["id"]).status_code == 409


def test_resumed_source_run_reuses_saved_picture_plans_and_delivery(
    client, service, library_source, monkeypatch
):
    frame = create_plan(client, name="Frame", width=1200, height=1600)
    bind(client, "living-room", "family", frame)
    root = service.submit(
        "library-generate",
        {"content": {"type": "photo_source", "source_id": "family"}, "options": {}},
        "living-room",
        "scheduler",
        "resume-test",
    )
    assert service.process_one()
    previous = service.operation(root)["result"]
    child = previous["outputs"][0]["operation_id"]
    service.db.execute("UPDATE operations SET status='COMPLETED' WHERE id=?", (child,))
    update = {k: v for k, v in frame.items() if k not in ("id", "archived")}
    assert (
        client.put(
            "/api/v1/generation-plans/" + frame["id"], json={**update, "quality": 60}
        ).status_code
        == 200
    )
    service.db.execute("UPDATE operations SET status='QUEUED' WHERE id=?", (root,))

    def unexpected_read(*args, **kwargs):
        raise AssertionError("Resumed run must not select a new photo")

    monkeypatch.setattr(service.photo_sources, "read", unexpected_read)
    service.settings.mock = False
    service.settings.write_enabled = False
    assert service.process_one()
    current = service.operation(root)["result"]
    assert current["source_image_id"] == previous["source_image_id"]
    assert current["outputs"][0]["operation_id"] == child
    assert service.db.one("SELECT COUNT(*) AS n FROM library_versions")["n"] == 1
    assert (
        service.db.one("SELECT COUNT(*) AS n FROM operations WHERE action='display'")[
            "n"
        ]
        == 1
    )


def test_one_plan_failure_preserves_other_output_and_source(
    client, service, library_source, monkeypatch
):
    from pigway_frame_flow.errors import ErrorCode, FlowError

    frame = create_plan(client, name="Frame", width=1200, height=1600)
    watch = create_plan(client)
    bind(client, "living-room", "family", frame)
    bind(client, "watch", "family", watch)
    generate = service.libraries.generate

    def fail_frame(picture_id, plan_id, *args, **kwargs):
        if plan_id == frame["id"]:
            raise FlowError(ErrorCode.INVALID_REQUEST, "Frame render failed")
        return generate(picture_id, plan_id, *args, **kwargs)

    monkeypatch.setattr(service.libraries, "generate", fail_frame)
    root = service.submit(
        "library-generate",
        {"content": {"type": "photo_source", "source_id": "family"}, "options": {}},
        "living-room",
        "scheduler",
        "partial-test",
    )
    assert service.process_one()
    op = client.get("/api/v1/operations/" + root).json()
    assert op["status"] == "FAILED"
    assert len(op["result"]["errors"]) == 1 and len(op["result"]["outputs"]) == 1
    assert op["result"]["outputs"][0]["terminal_id"] == "watch"
    picture_id = op["result"]["source_image_id"]
    monkeypatch.setattr(service.libraries, "generate", generate)
    assert (
        client.post(
            "/api/v1/source-images/" + picture_id + "/generate",
            json={"plan_id": frame["id"]},
        ).status_code
        == 200
    )
    assert service.db.one("SELECT COUNT(*) AS n FROM source_images")["n"] == 1


def test_successful_scheduled_frame_delivery_retains_job_provenance(
    client, app, service, library_source
):
    frame = create_plan(client, name="Frame", width=1200, height=1600)
    bind(client, "living-room", "family", frame)
    saved = client.post(
        "/api/v1/jobs",
        json={
            "name": "Library job",
            "schedule": {"type": "daily", "time": "07:00"},
            "source": {"type": "photo_source", "source_id": "family"},
            "library_pipeline": True,
        },
    ).json()
    now = datetime.now(UTC)
    service.db.execute(
        "UPDATE jobs SET next_run=? WHERE id=?", (now.isoformat(), saved["id"])
    )
    app.state.scheduler.tick(now)
    assert service.process_one() and service.process_one()
    assert (
        service.db.one("SELECT job_id FROM managed_cloud_uploads")["job_id"]
        == saved["id"]
    )
    service.sync()
    page = client.get("/api/v1/library-terminals/living-room/images").json()
    assert page["items"][0]["delivery_status"] == "COMPLETED"
    assert page["items"][0]["current"] is True
    from pigway_frame_flow.storage.database import encode, now_iso

    token = "library-reader-test"
    service.db.execute(
        "INSERT INTO clients VALUES(?,?,?,?,?,?,?)",
        (
            "reader",
            hashlib.sha256(token.encode()).hexdigest(),
            "Reader",
            encode(["images:read", "devices:read"]),
            now_iso(),
            None,
            1,
        ),
    )
    public = client.get(
        "/api/v1/library-terminals/living-room/images",
        headers={"Authorization": "Bearer " + token},
    )
    assert public.status_code == 200
    summary = public.json()["items"][0]["operation"]
    assert (
        "request_id" not in summary
        and "client_id" not in summary
        and "error" not in summary
    )
    assert (
        client.get(
            "/api/v1/operations/" + summary["operation_id"],
            headers={"Authorization": "Bearer " + token},
        ).status_code
        == 403
    )


def test_caption_plan_respects_full_picture_fit(client, service, library_source):
    picture = import_picture(client)
    service.db.execute(
        "UPDATE source_images SET caption=? WHERE id=?",
        ('{"title":"Caption"}', picture["id"]),
    )
    payload = {
        "source_id": "family",
        "name": "Caption",
        "width": 300,
        "height": 400,
        "caption": True,
        "fit": "contain",
    }
    response = client.post(
        "/api/v1/source-images/" + picture["id"] + "/preview", json=payload
    )
    assert response.status_code == 200
    with Image.open(BytesIO(response.content)) as image:
        assert image.getpixel((150, 0))[0] > 240
    response = client.post(
        "/api/v1/source-images/" + picture["id"] + "/preview",
        json={**payload, "fit": "cover"},
    )
    with Image.open(BytesIO(response.content)) as image:
        assert image.getpixel((150, 0))[0] < 60


def test_shared_version_keeps_separate_delivery_receipts_for_each_job(
    client, app, service, library_source
):
    frame = create_plan(client, name="Frame", width=1200, height=1600)
    bind(client, "living-room", "family", frame)
    job_ids = []
    for name in ["Morning", "Evening"]:
        job = client.post(
            "/api/v1/jobs",
            json={
                "name": name,
                "schedule": {"type": "daily", "time": "07:00"},
                "source": {"type": "photo_source", "source_id": "family"},
                "library_pipeline": True,
            },
        ).json()
        job_ids.append(job["id"])
    now = datetime.now(UTC)
    service.db.execute("UPDATE jobs SET next_run=?", (now.isoformat(),))
    app.state.scheduler.tick(now)
    assert service.process_one() and service.process_one()
    assert service.db.one("SELECT COUNT(*) AS n FROM library_versions")["n"] == 1
    assert service.db.one("SELECT COUNT(*) AS n FROM terminal_deliveries")["n"] == 2
    assert service.process_one()
    service.sync()
    assert service.process_one()
    service.sync()
    assert {
        r["job_id"] for r in service.db.all("SELECT job_id FROM managed_cloud_uploads")
    } == set(job_ids)
    page = client.get("/api/v1/library-terminals/living-room/images").json()
    assert page["total"] == 1 and page["items"][0]["delivery_status"] == "COMPLETED"


def test_watch_only_source_task_does_not_require_a_frame_driver(
    client, app, service, library_source
):
    plan = create_plan(client)
    bind(client, "watch", "family", plan)
    service.registry.drivers.clear()
    response = client.post(
        "/api/v1/jobs",
        json={
            "name": "Watch only",
            "device_id": "watch",
            "schedule": {"type": "daily", "time": "07:00"},
            "source": {"type": "photo_source", "source_id": "family"},
            "library_pipeline": True,
        },
    )
    assert response.status_code == 201
    now = datetime.now(UTC)
    service.db.execute("UPDATE jobs SET next_run=?", (now.isoformat(),))
    app.state.scheduler.tick(now)
    assert service.process_one()
    operation = service.db.one(
        "SELECT id,device_id,status FROM operations WHERE action='library-generate'"
    )
    assert (
        operation["device_id"] == "source:family" and operation["status"] == "COMPLETED"
    )
    assert (
        service.operation(operation["id"])["result"]["outputs"][0]["terminal_id"]
        == "watch"
    )


def test_interrupted_delivery_link_reuses_queue_idempotency(
    client, service, library_source
):
    plan = create_plan(client, name="Frame", width=1200, height=1600)
    bind(client, "living-room", "family", plan)
    root = service.submit(
        "library-generate",
        {"content": {"type": "photo_source", "source_id": "family"}, "options": {}},
        "source:family",
        "scheduler",
        "receipt-gap",
    )
    assert service.process_one()
    old = service.operation(root)["result"]
    child = old["outputs"][0]["operation_id"]
    service.db.execute("DELETE FROM terminal_deliveries WHERE operation_id=?", (child,))
    service.db.execute("UPDATE operations SET status='QUEUED' WHERE id=?", (root,))
    assert service.process_one()
    assert service.operation(root)["result"]["outputs"][0]["operation_id"] == child
    assert (
        service.db.one("SELECT COUNT(*) AS n FROM operations WHERE action='display'")[
            "n"
        ]
        == 1
    )
    assert (
        service.db.one("SELECT operation_id FROM terminal_deliveries")["operation_id"]
        == child
    )


def test_mode_switch_preserves_existing_schedule(client, service, library_source):
    plan = create_plan(client)
    bind(client, "watch", "family", plan)
    original = client.post(
        "/api/v1/jobs",
        json={
            "name": "Preserve timing",
            "schedule": {"type": "interval", "interval_minutes": 120},
            "source": {"type": "photo_source", "source_id": "family"},
        },
    ).json()
    switched = client.put(
        "/api/v1/jobs/" + original["id"] + "/library-mode", json={"enabled": True}
    )
    assert switched.status_code == 200
    current = switched.json()
    assert current["library_pipeline"] is True
    assert (
        current["next_run"] == original["next_run"]
        and current["schedule"] == original["schedule"]
        and current["enabled"] == original["enabled"]
    )
    assert (
        client.put(
            "/api/v1/jobs/" + original["id"] + "/library-mode", json={"enabled": False}
        ).status_code
        == 200
    )


def test_single_source_migration_preserves_timing_and_generates_without_device_writes(
    client, service, library_source, monkeypatch
):
    import runpy
    from pathlib import Path

    migrate = runpy.run_path(
        str(Path(__file__).parents[1] / "scripts/migrate-source-libraries.py")
    )["migrate"]
    job = client.post(
        "/api/v1/jobs",
        json={
            "name": "Existing source task",
            "schedule": {"type": "interval", "interval_minutes": 120},
            "source": {"type": "photo_source", "source_id": "family"},
        },
    ).json()

    def call(method, path, payload=None):
        response = client.request(method, path, json=payload)
        response.raise_for_status()
        return None if response.status_code == 204 else response.json()

    with pytest.raises(RuntimeError):
        migrate(call, apply=True)
    assert client.delete("/api/v1/photo-sources/travel").status_code == 204
    source_op = service.submit(
        "display",
        {
            "content": {"type": "photo_source", "source_id": "family"},
            "options": {"priority": 0},
        },
        "living-room",
        "scheduler",
        "current-photo",
    )
    assert service.process_one()
    service.sync()
    assert service.operation(source_op)["status"] == "COMPLETED"
    writes = []
    monkeypatch.setattr(
        service.registry.get("living-room"), "upload", lambda *args: writes.append(args)
    )
    result = migrate(call, apply=True)
    assert result["migrated"] and result["current_source_generated_locally"]
    assert writes == []
    migrated = client.get("/api/v1/jobs").json()[0]
    assert (
        migrated["library_pipeline"] is True and migrated["next_run"] == job["next_run"]
    )
    assert service.db.one("SELECT COUNT(*) AS n FROM terminal_bindings")["n"] == 2
    assert service.db.one("SELECT COUNT(*) AS n FROM source_images")["n"] == 1
    assert service.db.one("SELECT COUNT(*) AS n FROM library_versions")["n"] == 2
    assert migrate(call, apply=True)["already_configured"]


def test_terminal_current_picture_uses_synced_live_asset(
    client, service, library_source, monkeypatch
):
    from pigway_frame_flow.errors import ErrorCode, FlowError

    plan = create_plan(client, name="Frame", width=1200, height=1600)
    bind(client, "living-room", "family", plan)
    picture = import_picture(client)
    version = client.post(
        "/api/v1/source-images/" + picture["id"] + "/generate", json={}
    ).json()[0]
    current = {"synced": True, "source_known": True, "asset_id": version["asset_id"]}
    monkeypatch.setattr(service, "current_content", lambda _, **kwargs: current)
    endpoint = "/api/v1/library-terminals/living-room/images"
    assert client.get(endpoint).json()["items"][0]["current"] is True
    current["asset_id"] = "another-picture"
    assert client.get(endpoint).json()["items"][0]["current"] is False
    current.update(asset_id=version["asset_id"], synced=False)
    assert client.get(endpoint).json()["items"][0]["current"] is False
    current.update(synced=True, source_known=False)
    assert client.get(endpoint).json()["items"][0]["current"] is False

    def offline(_, **kwargs):
        raise FlowError(ErrorCode.CLOUD_UNAVAILABLE, "Offline", 502)

    monkeypatch.setattr(service, "current_content", offline)
    page = client.get(endpoint)
    assert page.status_code == 200 and page.json()["items"][0]["current"] is False

    driver = service.registry.get("living-room")
    offline_device = driver.device().model_copy(update={"status": "offline"})
    monkeypatch.setattr(driver, "device", lambda: offline_device)
    monkeypatch.setattr(
        service,
        "current_content",
        lambda _, **kwargs: pytest.fail(
            "Offline state must not be treated as a display confirmation"
        ),
    )
    assert client.get(endpoint).json()["items"][0]["current"] is False


def test_plan_rename_uses_current_labels_without_rerendering_legacy_versions(
    client, service, library_source
):
    import json

    from pigway_frame_flow.storage.database import encode

    plan = create_plan(client, name="Original", width=1200, height=1600)
    bind(client, "living-room", "family", plan)
    picture = import_picture(client)
    endpoint = "/api/v1/source-images/" + picture["id"] + "/generate"
    original = client.post(endpoint, json={"plan_id": plan["id"]}).json()[0]
    snapshot = original["plan"]
    original_bytes = service.assets.path(original["asset_id"]).read_bytes()
    # Existing deployed versions used the entire plan, including its old name.
    legacy_revision = hashlib.sha256(encode(snapshot).encode()).hexdigest()
    service.db.execute(
        "UPDATE library_versions SET revision=? WHERE asset_id=?",
        (legacy_revision, original["asset_id"]),
    )
    provenance = json.loads(
        service.db.one(
            "SELECT payload FROM asset_sources WHERE asset_id=?",
            (original["asset_id"],),
        )["payload"]
    )
    provenance["revision"] = legacy_revision
    service.db.execute(
        "UPDATE asset_sources SET payload=? WHERE asset_id=?",
        (encode(provenance), original["asset_id"]),
    )
    payload = {k: v for k, v in plan.items() if k not in ("id", "archived")}
    renamed = {**payload, "name": "Frame v2.1"}
    assert (
        client.put("/api/v1/generation-plans/" + plan["id"], json=renamed).status_code
        == 200
    )
    repeated = client.post(endpoint, json={"plan_id": plan["id"]}).json()[0]
    assert repeated["asset_id"] == original["asset_id"]
    assert repeated["revision"] == legacy_revision
    assert repeated["plan"] == snapshot
    assert repeated["plan_name"] == "Frame v2.1"
    assert repeated["name"] == "photo-Frame v2.1.jpg"
    assert repeated["stored_name"] == original["stored_name"]
    assert repeated["matches_current_plan"] is True
    assert service.assets.path(original["asset_id"]).read_bytes() == original_bytes
    assert service.db.one("SELECT COUNT(*) AS n FROM library_versions")["n"] == 1
    source_version = client.get("/api/v1/source-libraries/family/images").json()[
        "items"
    ][0]["versions"][0]
    terminal_version = client.get(
        "/api/v1/library-terminals/living-room/images"
    ).json()["items"][0]
    assert source_version["plan_name"] == terminal_version["plan_name"] == "Frame v2.1"
    assert source_version["name"] == terminal_version["name"]
    assert (
        client.put(
            "/api/v1/generation-plans/" + plan["id"], json={**renamed, "quality": 70}
        ).status_code
        == 200
    )
    old = client.get("/api/v1/library-terminals/living-room/images").json()["items"][0]
    assert old["matches_current_plan"] is False and old["plan"]["quality"] == 85
    updated = client.post(endpoint, json={"plan_id": plan["id"]}).json()[0]
    assert (
        updated["asset_id"] != original["asset_id"]
        and updated["matches_current_plan"] is True
    )
    assert updated["plan"]["quality"] == 70
    client.delete("/api/v1/library-terminals/living-room/binding")
    assert client.delete("/api/v1/generation-plans/" + plan["id"]).status_code == 204
    versions = client.get("/api/v1/source-libraries/family/images").json()["items"][0][
        "versions"
    ]
    assert len(versions) == 2 and all(v["plan_archived"] for v in versions)


def test_source_deletion_removes_unreferenced_archived_plans(
    client, service, library_source
):
    plan = create_plan(client)
    assert client.delete("/api/v1/generation-plans/" + plan["id"]).status_code == 204
    assert client.delete("/api/v1/photo-sources/family").status_code == 204
    assert (
        service.db.one("SELECT id FROM library_plans WHERE source_id='family'") is None
    )
    assert client.get("/api/v1/generation-plans?source_id=family").status_code == 404


def test_manual_upload_survives_reload_and_generates_bound_versions(client, service):
    data = BytesIO()
    Image.new("RGB", (700, 500), "blue").save(data, "PNG")
    path = "/api/v1/source-libraries/manual-uploads/upload"
    result = client.post(
        path, files={"file": ("original.png", data.getvalue(), "image/png")}
    )
    assert result.status_code == 200, result.text
    picture = result.json()
    with Image.open(service.assets.path(picture["id"])) as image:
        assert image.size == (700, 500)
    assert (
        client.post(path, files={"file": ("same.png", data.getvalue())}).json()["id"]
        == picture["id"]
    )
    listed = client.get("/api/v1/source-libraries/manual-uploads/images").json()
    assert listed["total"] == 1 and listed["items"][0]["id"] == picture["id"]
    plan = create_plan(client, "manual-uploads")
    bind(client, "watch", "manual-uploads", plan)
    generated = client.post(
        "/api/v1/source-images/" + picture["id"] + "/generate",
        json={"plan_id": plan["id"]},
    )
    assert generated.status_code == 200, generated.text
    assert generated.json()[0]["plan"]["width"] == 410
    run = client.post(
        "/api/v1/source-libraries/manual-uploads/run", json={"deliver": False}
    )
    assert run.status_code == 200, run.text
    assert (
        client.get("/api/v1/source-libraries/manual-uploads/images").json()["total"]
        == 1
    )
    assert client.get("/api/v1/library-terminals/watch/images").json()["total"] == 1


def test_invalid_upload_does_not_create_source(client, service):
    response = client.post(
        "/api/v1/source-libraries/manual-uploads/upload",
        files={"file": ("bad.jpg", b"not an image")},
    )
    assert response.status_code == 400
    assert service.photo_sources.list() == []


def test_ingest_sql_failure_rolls_back_asset_and_file(
    client, service, library_source, monkeypatch
):
    original = service.db.execute

    def fail(query, args=()):
        if "INSERT INTO source_images" in query:
            raise RuntimeError("injected SQL failure")
        return original(query, args)

    monkeypatch.setattr(service.db, "execute", fail)
    with pytest.raises(RuntimeError):
        service.libraries.import_picture("family", "photo.jpg")
    assert service.db.all("SELECT * FROM assets") == []
    assert service.db.all("SELECT * FROM source_images") == []
    assert list((service.settings.data_dir / "source-images").iterdir()) == []
    assert service.db.all("SELECT * FROM asset_file_journal") == []


def test_generation_sql_failure_keeps_source_without_orphan(
    client, service, library_source, monkeypatch
):
    picture = import_picture(client)
    plan = create_plan(client)
    original = service.db.execute

    def fail(query, args=()):
        if "INSERT INTO library_versions" in query:
            raise RuntimeError("injected SQL failure")
        return original(query, args)

    monkeypatch.setattr(service.db, "execute", fail)
    with pytest.raises(RuntimeError):
        service.libraries.generate(picture["id"], plan["id"])
    assert len(service.db.all("SELECT * FROM assets")) == 1
    assert list((service.settings.data_dir / "renditions").iterdir()) == []
    assert service.assets.path(picture["id"]).exists()


def test_delete_sql_failure_preserves_files_and_relations(
    client, service, library_source, monkeypatch
):
    picture = import_picture(client)
    plan = create_plan(client)
    version = service.libraries.generate(picture["id"], plan["id"])
    original = service.db.execute

    def fail(query, args=()):
        if query == "DELETE FROM source_images WHERE id=?":
            raise RuntimeError("injected SQL failure")
        return original(query, args)

    monkeypatch.setattr(service.db, "execute", fail)
    with pytest.raises(RuntimeError):
        service.libraries.delete_picture(picture["id"])
    assert service.assets.path(version["asset_id"]).exists()
    assert (
        service.libraries.picture(picture["id"])["versions"][0]["asset_id"]
        == version["asset_id"]
    )
    assert service.db.all("SELECT * FROM asset_file_journal") == []


def test_delete_file_failure_retries_on_restart(
    client, service, library_source, monkeypatch
):
    from pathlib import Path

    from pigway_frame_flow.renderers.images import Assets

    picture = import_picture(client)
    path = service.assets.path(picture["id"])
    unlink = Path.unlink

    def fail(file, *args, **kwargs):
        if file == path:
            raise OSError("injected unlink failure")
        return unlink(file, *args, **kwargs)

    with monkeypatch.context() as patch:
        patch.setattr(Path, "unlink", fail)
        service.libraries.delete_picture(picture["id"])
        assert service.db.all("SELECT * FROM source_images") == []
        assert (
            service.db.one("SELECT id FROM asset_file_journal")["id"] == picture["id"]
        )
    Assets(service.db, service.settings)
    assert not path.exists()
    assert service.db.all("SELECT * FROM asset_file_journal") == []


def test_asset_disk_write_failure_leaves_no_partial_file(service, monkeypatch):
    import os

    fsync = os.fsync
    calls = 0

    def fail(fd):
        nonlocal calls
        calls += 1
        if calls == 1:
            raise OSError("injected disk write failure")
        return fsync(fd)

    monkeypatch.setattr(os, "fsync", fail)
    with pytest.raises(OSError):
        service.assets.put(b"data", "failed.jpg")
    assert service.db.all("SELECT * FROM assets") == []
    assert service.db.all("SELECT * FROM asset_file_journal") == []
    assert list((service.settings.data_dir / "images").iterdir()) == []


def test_asset_startup_removes_interrupted_creation(service):
    from pigway_frame_flow.renderers.images import Assets

    asset_id = "c" * 32
    root = service.settings.data_dir / "images"
    (root / (asset_id + ".jpg")).write_bytes(b"incomplete")
    (root / (asset_id + ".part")).write_bytes(b"partial")
    service.db.execute(
        "INSERT INTO asset_file_journal VALUES(?,?)", (asset_id, "images")
    )
    Assets(service.db, service.settings)
    assert not list(root.iterdir())
    assert service.db.all("SELECT * FROM asset_file_journal") == []