"""§4b attachment recovery: multi-turn harvest, collision suffixing, content dedup,
per-file failure isolation, and the daemon's download-only request."""

from __future__ import annotations

import base64
import io
import json
import os
import queue
from pathlib import Path

import pytest

import chat
import ask_gpt
import convlog
import solwebd
from fake_browser import seat_for

CID = "66666666-6666-4666-8666-666666666666"


@pytest.fixture(autouse=True)
def registered_conversation():
    import registry
    registry.append("owner@example.com", CID, "existing")


def data_url(payload: bytes, ctype: str = "application/octet-stream") -> str:
    return f"data:{ctype};base64," + base64.b64encode(payload).decode()


def test_generated_download_waits_for_nonempty_stable_file(monkeypatch, tmp_path):
    download_dir = tmp_path / "browser-downloads"
    download_dir.mkdir()
    browser = MultiTurnBrowser([], [])
    browser.download_dir = download_dir

    def sync_script(source, _args=None):
        return True if source == chat.CLICK_DOWNLOAD else "absent"
    browser.sync_script = sync_script
    snapshots = [
        {},
        {download_dir / "bundle.zip": (1, 0)},
        {download_dir / "bundle.zip": (2, 4)},
        {download_dir / "bundle.zip": (2, 4)},
    ]
    now = iter((0.0, 0.5, 0.5, 1.0, 1.0, 3.1, 3.1))
    monkeypatch.setattr(chat, "_download_snapshot", lambda _browser: snapshots.pop(0))
    monkeypatch.setattr(chat.time, "monotonic", lambda: next(now))
    (download_dir / "bundle.zip").write_bytes(b"PK00")

    saved = chat._download_button(browser, 0, "bundle.zip", tmp_path)

    assert saved.read_bytes() == b"PK00"


class MultiTurnBrowser:
    """Answers HARVEST_ALL with one entry per assistant turn and FETCH with a
    scripted queue of blobs, in request order."""

    def __init__(self, per_turn, blobs):
        self.per_turn = per_turn
        self.blobs = list(blobs)

    def url(self):
        return f"https://chatgpt.com/c/{CID}"

    def script(self, source, args=None):
        assert source == chat.FETCH
        return self.blobs.pop(0) if self.blobs else {"error": "no more blobs scripted"}

    def sync_script(self, source, args=None):
        if source == chat.HARVEST_ALL:
            return self.per_turn
        raise AssertionError(f"unscripted call: {source[:50]!r}")


def test_collects_from_every_turn_not_just_the_last(tmp_path):
    browser = MultiTurnBrowser(
        per_turn=[
            {"images": ["https://x/a"], "files": [], "downloads": []},
            {"images": ["https://x/b"], "files": [], "downloads": []},
        ],
        blobs=[{"type": "image/png", "data": data_url(b"AAA")},
              {"type": "image/png", "data": data_url(b"BBB")}])
    saved, failures = chat.collect_attachments(browser, tmp_path)
    assert not failures
    assert {p.read_bytes() for p in saved} == {b"AAA", b"BBB"}


def test_name_collisions_across_turns_get_a_numeric_suffix(tmp_path):
    browser = MultiTurnBrowser(
        per_turn=[
            {"images": [], "files": [{"src": "https://x/1", "name": "report.pdf"}],
             "downloads": []},
            {"images": [], "files": [{"src": "https://x/2", "name": "report.pdf"}],
             "downloads": []},
        ],
        blobs=[{"type": "application/pdf", "data": data_url(b"one")},
              {"type": "application/pdf", "data": data_url(b"two")}])
    saved, failures = chat.collect_attachments(browser, tmp_path)
    assert not failures
    assert sorted(p.name for p in saved) == ["report-2.pdf", "report.pdf"]
    assert {p.read_bytes() for p in saved} == {b"one", b"two"}


def test_identical_content_is_skipped_not_duplicated(tmp_path):
    browser = MultiTurnBrowser(
        per_turn=[
            {"images": [], "files": [{"src": "https://x/1", "name": "same.txt"}],
             "downloads": []},
            {"images": [], "files": [{"src": "https://x/2", "name": "same-again.txt"}],
             "downloads": []},
        ],
        blobs=[{"type": "text/plain", "data": data_url(b"identical bytes")},
              {"type": "text/plain", "data": data_url(b"identical bytes")}])
    saved, failures = chat.collect_attachments(browser, tmp_path)
    assert not failures
    assert len(set(saved)) == 1
    assert len([p for p in tmp_path.iterdir() if p.is_file()]) == 1


def test_one_failed_fetch_does_not_abort_the_rest(tmp_path):
    browser = MultiTurnBrowser(
        per_turn=[{"images": [], "files": [
            {"src": "https://x/bad", "name": "bad.bin"},
            {"src": "https://x/good", "name": "good.bin"},
        ], "downloads": []}],
        blobs=[{"error": "http 404"},
              {"type": "application/octet-stream", "data": data_url(b"ok")}])
    saved, failures = chat.collect_attachments(browser, tmp_path)
    assert [p.name for p in saved] == ["good.bin"]
    assert failures == [{"file": "bad.bin", "error": "could not download file 1: "
                                                      "{'error': 'http 404'}"}]


def test_no_turns_at_all_is_an_error():
    browser = MultiTurnBrowser(per_turn=[], blobs=[])
    with pytest.raises(chat.ChatError):
        chat.collect_attachments(browser, None)


def test_daemon_download_uses_shared_lifecycle_and_emits_queue_phase(monkeypatch):
    monkeypatch.setattr(chat.time, "sleep", lambda *_: None)
    seat, marionette = seat_for([])
    marionette.turns = 1
    engine = solwebd.Engine(seat)
    calls = []
    events = []

    def fake_download(m, conversation_id, *, expected_account, out_dir, attempt, on_event, turn,
                      conversation_opened):
        calls.append((m, conversation_id, expected_account, out_dir, attempt, turn,
                      conversation_opened))
        return {"text": "", "thinking": "", "images": [], "files": ["/tmp/a.png"],
                "failures": [], "unavailable": [], "invalid": [], "planned": 1,
                "reused": 0, "conversation_id": conversation_id}

    monkeypatch.setattr(solwebd.chat, "download_attachments_on", fake_download)
    result = engine.ask({"conversation": CID, "download": True, "stamp": "s",
                         "attempt": "daemon-1"}, on_event=events.append)
    assert result["files"] == ["/tmp/a.png"]
    assert result["conversation_id"] == CID
    assert calls[0][0] is marionette and calls[0][1] == CID
    assert calls[0][2] == "owner@example.com"
    assert calls[0][4] == "daemon-1"
    assert calls[0][6] is True
    assert events[0]["type"] == "download_phase"
    assert events[0]["state"] == "queued"
    assert events[0]["attempt"] == "daemon-1"
    assert events[1]["state"] == "opening"


def test_daemon_unknown_id_fails_before_browser_open():
    unknown = "77777777-7777-4777-8777-777777777777"
    seat, marionette = seat_for([])
    engine = solwebd.Engine(seat)
    with pytest.raises(solwebd.ConversationNotFound, match="not in the local registry"):
        engine.ask({"conversation": unknown, "download": True, "stamp": "s"})
    assert marionette.navigations == []


def test_daemon_rejects_account_mismatch_before_navigation(monkeypatch):
    seat, marionette = seat_for([])
    marionette.account = "other@example.com"
    engine = solwebd.Engine(seat)
    with pytest.raises(solwebd.ConversationNotFound, match="current account other@example.com"):
        engine.ask({"conversation": CID, "download": True, "stamp": "s"})
    assert marionette.navigations
    assert set(marionette.navigations) == {"https://chatgpt.com/"}


def test_download_and_prompt_together_is_rejected():
    engine = solwebd.Engine(seat_for([])[0])
    with pytest.raises(ValueError):
        engine.ask({"conversation": CID, "download": True, "prompt": "hi", "stamp": "s"})


def test_download_without_conversation_is_rejected():
    engine = solwebd.Engine(seat_for([])[0])
    with pytest.raises(ValueError):
        engine.ask({"download": True, "stamp": "s"})


def test_daemon_preserves_structured_unsuccessful_download_result(monkeypatch):
    monkeypatch.setattr(chat.time, "sleep", lambda *_: None)
    seat, marionette = seat_for([])
    marionette.turns = 1
    engine = solwebd.Engine(seat)
    reply = {"text": "", "thinking": "", "images": [], "files": [],
             "failures": [{"file": "x", "error": "failed"}], "unavailable": [],
             "invalid": [], "planned": 1, "reused": 0, "conversation_id": CID}
    monkeypatch.setattr(solwebd.chat, "download_attachments_on",
                        lambda *args, **kwargs: reply)
    assert engine.ask({"conversation": CID, "download": True, "stamp": "s"}) == reply


def test_discovery_builds_complete_plan_before_fetch_and_classifies_entries(tmp_path):
    browser = MultiTurnBrowser(
        per_turn=[
            {"images": ["https://x/a"],
             "files": [{"src": "https://x/one", "name": "one.txt"}, {"name": "bad"},
                       {"src": 123, "name": "number.bin"}],
             "downloads": ["Download old.bin"]},
            {"images": [], "files": [{"src": "https://x/two", "name": "two.txt"}],
             "downloads": ["Download latest.csv"]},
        ],
        blobs=[])

    items, unavailable, invalid = chat.discover_attachments(browser)

    assert [(item.index, item.total, item.turn, item.kind, item.label) for item in items] == [
        (1, 4, 1, "image", "turn-1-image-1"),
        (2, 4, 1, "file", "one.txt"),
        (3, 4, 2, "file", "two.txt"),
        (4, 4, 2, "button", "Download latest.csv"),
    ]
    assert unavailable == [{
        "file": "Download old.bin", "kind": "button",
        "reason": "only latest-turn button downloads are recoverable",
    }]
    assert invalid == [
        {"file": "turn-1-file-2", "reason": "invalid metadata", "kind": "file"},
        {"file": "turn-1-file-3", "reason": "invalid metadata", "kind": "file"},
    ]
    assert browser.blobs == []


def test_shared_download_verifies_account_before_discovery(monkeypatch, tmp_path):
    browser = MultiTurnBrowser(
        per_turn=[{"images": [], "files": [], "downloads": []}], blobs=[])
    verified = []

    def reject(m, conversation_id, expected_account):
        verified.append((conversation_id, expected_account))
        raise chat.ChatError("account mismatch")

    monkeypatch.setattr(chat, "open_resume", reject)
    monkeypatch.setattr(
        chat, "discover_attachments",
        lambda m: (_ for _ in ()).throw(AssertionError("discovery ran before verification")))
    events = []

    with pytest.raises(chat.ChatError, match="account mismatch"):
        chat.download_attachments_on(
            browser, CID, expected_account="owner@example.com", out_dir=tmp_path,
            attempt="direct-1", on_event=events.append)

    assert verified == [(CID, "owner@example.com")]
    assert not any(event["type"] == "download_plan" for event in events)
    assert events[-1]["type"] == "error"


def test_dom_labels_are_one_printable_line_in_events_and_manifest(monkeypatch, tmp_path):
    browser = MultiTurnBrowser(
        per_turn=[{"images": [],
                   "files": [{"src": "https://x/file", "name": "report\n\x1b[31mspoof\x7f\t.csv"}],
                   "downloads": ["Download\r\n\x1b[2J evil.bin"]}],
        blobs=[])
    monkeypatch.setattr(chat, "open_resume", lambda m, conversation_id, expected_account: None)
    monkeypatch.setattr(
        chat, "_download_item",
        lambda *args, **kwargs: (_ for _ in ()).throw(chat.ChatError("fetch failed")))
    events = []

    result = chat.download_attachments_on(
        browser, CID, expected_account="owner@example.com", out_dir=tmp_path, attempt="direct-1", on_event=events.append)

    labels = [item["file"] for item in next(
        event for event in events if event["type"] == "download_plan")["items"]]
    labels += [event["file"] for event in events if event["type"] == "download"]
    labels += [failure["file"] for failure in result["failures"]]
    assert set(labels) == {"report [31mspoof .csv", "Download [2J evil.bin"}
    assert all("\n" not in label and "\r" not in label and "\x1b" not in label
               and all(char.isprintable() for char in label) for label in labels)

    stream = io.StringIO()
    monkeypatch.setattr(ask_gpt.sys, "stderr", stream)
    progress = ask_gpt.DownloadProgress(
        is_tty=False, clock=lambda: 0.0, interval=3600, out_dir=Path("ask-gpt"))
    progress.start()
    progress.on_event(next(event for event in events if event["type"] == "download_plan"))
    progress.stop()
    rendered = stream.getvalue()
    assert "\x1b" not in rendered
    assert "report [31mspoof .csv" in rendered
    assert "Download [2J evil.bin" in rendered


def test_control_only_dom_labels_use_deterministic_fallback():
    browser = MultiTurnBrowser(
        per_turn=[{"images": [], "files": [{"src": "https://x/file", "name": "\n\x1b\x7f"}],
                   "downloads": ["\r\t\x1b"]}], blobs=[])
    items, _, _ = chat.discover_attachments(browser)
    assert [item.label for item in items] == ["turn-1-file-1", "turn-1-button-1"]


def test_shared_download_emits_plan_before_fetch_and_structured_item_lifecycle(
        monkeypatch, tmp_path):
    browser = MultiTurnBrowser(
        per_turn=[{"images": [], "files": [
            {"src": "https://x/bad", "name": "bad.bin"},
            {"src": "https://x/good", "name": "good.bin"},
            {"src": "https://x/reused", "name": "copy.bin"},
        ], "downloads": []}],
        blobs=[{"error": "http 404"},
               {"type": "application/octet-stream", "data": data_url(b"ok")},
               {"type": "application/octet-stream", "data": data_url(b"ok")}])
    monkeypatch.setattr(chat, "open_resume", lambda m, conversation_id, expected_account: None)
    events = []

    result = chat.download_attachments_on(
        browser, CID, expected_account="owner@example.com", out_dir=tmp_path, attempt="direct-1", on_event=events.append)

    plan_at = next(i for i, event in enumerate(events) if event["type"] == "download_plan")
    started_at = next(i for i, event in enumerate(events) if event["type"] == "download")
    assert plan_at < started_at
    lifecycle = [event for event in events if event["type"] == "download"]
    assert [(event["index"], event["state"]) for event in lifecycle] == [
        (1, "started"), (1, "failed"), (2, "started"), (2, "saved"),
        (3, "started"), (3, "saved"),
    ]
    assert all(event["attempt"] == "direct-1" for event in events)
    assert result["planned"] == 3
    assert result["files"] == [str(tmp_path / "good.bin"), str(tmp_path / "good.bin")]
    assert result["reused"] == 1
    assert result["failures"] == [{"file": "bad.bin", "error": "could not download file 1: "
                                                            "{'error': 'http 404'}"}]


def test_sse_queue_never_drops_or_reorders_control_events_when_delta_storage_saturates():
    sink = solwebd.SSEQueue(maxsize=2)
    controls = [
        {"type": "meta", "conversation_id": CID},
        {"type": "download_plan", "items": [{"index": 1}]},
        {"type": "download", "state": "started", "index": 1},
        {"type": "download", "state": "saved", "index": 1},
        {"type": "done", "reply": {"files": ["a"]}},
    ]
    sink.put({"type": "answer", "delta": "a"})
    sink.put({"type": "answer", "delta": "b"})
    for control in controls:
        sink.put(control)
        sink.put({"type": "answer", "delta": control["type"]})

    received = []
    while True:
        try:
            received.append(sink.get(timeout=0))
        except queue.Empty:
            break

    assert [event["type"] for event in received if event["type"] not in {"answer", "replace"}] \
        == [event["type"] for event in controls]
    assert [event["seq"] for event in received] == sorted(event["seq"] for event in received)
    plan_pos = next(i for i, event in enumerate(received) if event["type"] == "download_plan")
    assert not any(event.get("answer", "").endswith("download_plan")
                   for event in received[:plan_pos])


def test_non_tty_progress_prints_manifest_item_states_and_summary(monkeypatch, tmp_path):
    stream = io.StringIO()
    monkeypatch.setattr(ask_gpt.sys, "stderr", stream)
    now = [0.0]
    progress = ask_gpt.DownloadProgress(
        is_tty=False, clock=lambda: now[0], interval=3600, out_dir=Path("ask-gpt"))
    progress.start()
    progress.on_event({"type": "download_phase", "state": "opening",
                       "conversation_id": CID, "attempt": "direct-1"})
    progress.on_event({"type": "download_plan", "attempt": "direct-1", "items": [
        {"index": 1, "total": 1, "kind": "file", "file": "result.zip"}],
        "unavailable": [{"file": "old.bin", "kind": "button", "reason": "older turn"}],
        "invalid": []})
    progress.on_event({"type": "download", "state": "started", "attempt": "direct-1",
                       "index": 1, "total": 1, "kind": "file", "file": "result.zip"})
    now[0] = 12.0
    progress.on_event({"type": "download", "state": "saved", "attempt": "direct-1",
                       "index": 1, "total": 1, "kind": "file", "file": "result.zip",
                       "path": str(tmp_path / "result.zip"), "reused_existing": False})
    progress.finish([str(tmp_path / "result.zip")], [],
                    [{"file": "old.bin", "reason": "older turn"}], [])

    output = stream.getvalue()
    assert "download: starting" in output
    assert f"download: opening conversation {CID}" in output
    assert "Will download 1 attachment; 1 unavailable" in output
    assert "1/1  [file]" in output and "result.zip" in output
    assert "download: file 1/1 · completed 0/1 · result.zip" in output
    assert "Recovered 1/1 attachment (1 file) to ask-gpt in 00:12; 1 unavailable." in output
    assert "\x1b[" not in output


def test_tty_progress_uses_in_place_bar_and_eta_after_completion(monkeypatch):
    stream = io.StringIO()
    monkeypatch.setattr(ask_gpt.sys, "stderr", stream)
    now = [0.0]
    progress = ask_gpt.DownloadProgress(
        is_tty=True, clock=lambda: now[0], interval=3600, out_dir=Path("ask-gpt"), width=90)
    progress.start()
    progress.on_event({"type": "download_plan", "attempt": "direct-1", "items": [
        {"index": 1, "total": 2, "kind": "file", "file": "first-long-name.zip"},
        {"index": 2, "total": 2, "kind": "image", "file": "second.png"}],
        "unavailable": [], "invalid": []})
    progress.on_event({"type": "download", "state": "started", "attempt": "direct-1",
                       "index": 1, "total": 2, "kind": "file", "file": "first-long-name.zip"})
    now[0] = 10.0
    progress.on_event({"type": "download", "state": "saved", "attempt": "direct-1",
                       "index": 1, "total": 2, "kind": "file", "file": "first-long-name.zip",
                       "path": "/tmp/first-long-name.zip", "reused_existing": False})
    now[0] = 11.0
    progress.on_event({"type": "download", "state": "started", "attempt": "direct-1",
                       "index": 2, "total": 2, "kind": "image", "file": "second.png"})
    progress.finish(["/tmp/first-long-name.zip"], [], [], [])

    output = stream.getvalue()
    assert "\r[" in output
    assert "file 2/2 · completed 1/2" in output
    assert "elapsed 00:11 · ETA ~00:10" in output


def _isolate_download_log(monkeypatch, tmp_path):
    monkeypatch.setattr(convlog, "STATE", tmp_path / "state")
    monkeypatch.setattr(convlog, "LOGS", tmp_path / "state" / "logs")
    monkeypatch.chdir(tmp_path)


def _successful_direct(events_seen):
    def run(conversation_id, *, expected_account, out_dir, mode, reseed, on_event, attempt):
        assert expected_account == "owner@example.com"
        events = [
            {"type": "download_phase", "state": "opening", "phase": "opening",
             "conversation_id": conversation_id, "attempt": attempt},
            {"type": "download_phase", "state": "scanning", "phase": "scanning",
             "conversation_id": conversation_id, "attempt": attempt},
            {"type": "download_plan", "phase": "scanning", "attempt": attempt,
             "items": [{"index": 1, "total": 1, "turn": 1, "kind": "file",
                        "file": "result.zip"}], "unavailable": [], "invalid": []},
            {"type": "download", "phase": "collecting", "state": "started",
             "attempt": attempt, "index": 1, "total": 1, "kind": "file",
             "file": "result.zip", "path": None},
            {"type": "download", "phase": "collecting", "state": "saved",
             "attempt": attempt, "index": 1, "total": 1, "kind": "file",
             "file": "result.zip", "path": str(out_dir / "result.zip"),
             "reused_existing": False},
        ]
        reply = {"text": "", "thinking": "", "images": [],
                 "files": [str(out_dir / "result.zip")], "failures": [],
                 "unavailable": [], "invalid": [], "planned": 1, "reused": 0,
                 "conversation_id": conversation_id}
        events.append({"type": "done", "phase": "done", "attempt": attempt,
                       "reply": reply})
        for event in events:
            events_seen.append(event)
            on_event(event)
        return [out_dir / "result.zip"], []
    return run


def test_default_download_command_renders_progress_without_live(
        monkeypatch, tmp_path, capsys):
    _isolate_download_log(monkeypatch, tmp_path)
    monkeypatch.setattr(ask_gpt, "daemon_endpoint", lambda: None)
    events = []
    monkeypatch.setattr(ask_gpt.chat, "download_attachments", _successful_direct(events))

    assert ask_gpt.main(["--download-attachments", CID, "--out", "downloads"]) == 0

    captured = capsys.readouterr()
    assert captured.out == f"saved: {tmp_path / 'downloads' / 'result.zip'}\n"
    assert "download: starting" in captured.err
    assert "Will download 1 attachment" in captured.err
    assert "Recovered 1/1 attachment" in captured.err
    assert events[0]["attempt"] == "direct-1"


def test_json_download_keeps_stdout_machine_pure(monkeypatch, tmp_path, capsys):
    _isolate_download_log(monkeypatch, tmp_path)
    monkeypatch.setattr(ask_gpt, "daemon_endpoint", lambda: None)
    monkeypatch.setattr(ask_gpt.chat, "download_attachments", _successful_direct([]))

    assert ask_gpt.main(["--download-attachments", CID, "--json"]) == 0

    captured = capsys.readouterr()
    result = json.loads(captured.out)
    assert result["planned"] == 1
    assert result["failures"] == []
    assert "download: starting" in captured.err


def test_daemon_disconnect_after_plan_never_starts_duplicate_direct_pass(
        monkeypatch, tmp_path, capsys):
    _isolate_download_log(monkeypatch, tmp_path)
    token = tmp_path / "token"
    token.write_text("x")
    monkeypatch.setattr(ask_gpt, "TOKEN_FILE", token)
    monkeypatch.setattr(ask_gpt, "daemon_endpoint", lambda: "http://daemon")

    def disconnect(*args, **kwargs):
        kwargs["on_event"]({"type": "download_plan", "phase": "scanning",
                            "attempt": "daemon-1", "items": [],
                            "unavailable": [], "invalid": []})
        raise ConnectionResetError("lost")

    monkeypatch.setattr(ask_gpt, "_ask_stream", disconnect)
    monkeypatch.setattr(ask_gpt.chat, "download_attachments",
                        lambda *args, **kwargs: pytest.fail("duplicate direct recovery started"))

    assert ask_gpt.main(["--download-attachments", CID]) == 1
    assert "after download plan" in capsys.readouterr().err


def test_direct_ctrl_c_reports_kept_files_and_exits_130(monkeypatch, tmp_path, capsys):
    _isolate_download_log(monkeypatch, tmp_path)
    monkeypatch.setattr(ask_gpt, "daemon_endpoint", lambda: None)

    def interrupt(conversation_id, *, expected_account, out_dir, mode, reseed, on_event, attempt):
        assert expected_account == "owner@example.com"
        on_event({"type": "download_plan", "phase": "scanning", "attempt": attempt,
                  "items": [{"index": 1, "total": 2, "kind": "file", "file": "a"},
                            {"index": 2, "total": 2, "kind": "file", "file": "b"}],
                  "unavailable": [], "invalid": []})
        on_event({"type": "download", "phase": "collecting", "attempt": attempt,
                  "state": "saved", "index": 1, "total": 2, "kind": "file",
                  "file": "a", "path": str(out_dir / "a"), "reused_existing": False})
        raise KeyboardInterrupt

    monkeypatch.setattr(ask_gpt.chat, "download_attachments", interrupt)
    assert ask_gpt.main(["--download-attachments", CID, "--out", "downloads"]) == 130
    err = capsys.readouterr().err
    assert "interrupted after 1/2" in err
    assert "completed files kept in downloads" in err
    assert "Traceback" not in err


def test_renderer_failure_does_not_abort_download(monkeypatch, tmp_path, capsys):
    _isolate_download_log(monkeypatch, tmp_path)
    monkeypatch.setattr(ask_gpt, "daemon_endpoint", lambda: None)
    monkeypatch.setattr(ask_gpt.chat, "download_attachments", _successful_direct([]))

    class BrokenProgress:
        plan_seen = False
        completed = 0

        def __init__(self, **kwargs):
            pass

        def start(self):
            pass

        def on_event(self, event):
            raise OSError("stderr closed")

        def finish(self, *args):
            pass

        def interrupt(self, **kwargs):
            pass

    monkeypatch.setattr(ask_gpt, "DownloadProgress", BrokenProgress)
    assert ask_gpt.main(["--download-attachments", CID]) == 0
    assert "saved:" in capsys.readouterr().out


def test_renderer_failure_cannot_enable_post_plan_fallback(monkeypatch, tmp_path):
    _isolate_download_log(monkeypatch, tmp_path)
    token = tmp_path / "token"
    token.write_text("x")
    monkeypatch.setattr(ask_gpt, "TOKEN_FILE", token)
    endpoints = iter(["http://daemon", None])
    monkeypatch.setattr(ask_gpt, "daemon_endpoint", lambda: next(endpoints))

    def disconnect(*args, **kwargs):
        kwargs["on_event"]({"type": "download_plan", "phase": "scanning",
                            "attempt": "daemon-1", "items": [],
                            "unavailable": [], "invalid": []})
        raise ConnectionResetError("lost")

    class BrokenProgress:
        plan_seen = False
        completed = 0

        def __init__(self, **kwargs):
            pass

        def start(self):
            pass

        def on_event(self, event):
            raise OSError("stderr closed")

        def stop(self):
            pass

        def interrupt(self, **kwargs):
            pass

    monkeypatch.setattr(ask_gpt, "DownloadProgress", BrokenProgress)
    monkeypatch.setattr(ask_gpt, "_ask_stream", disconnect)
    monkeypatch.setattr(ask_gpt.chat, "download_attachments",
                        lambda *args, **kwargs: pytest.fail("duplicate direct recovery started"))
    assert ask_gpt.main(["--download-attachments", CID]) == 1


def test_daemon_disconnect_before_plan_retries_direct_as_second_attempt(
        monkeypatch, tmp_path, capsys):
    _isolate_download_log(monkeypatch, tmp_path)
    token = tmp_path / "token"
    token.write_text("x")
    monkeypatch.setattr(ask_gpt, "TOKEN_FILE", token)
    endpoints = iter(["http://daemon", None])
    monkeypatch.setattr(ask_gpt, "daemon_endpoint", lambda: next(endpoints))
    monkeypatch.setattr(ask_gpt, "_ask_stream",
                        lambda *args, **kwargs: (_ for _ in ()).throw(ConnectionResetError("lost")))
    attempts = []
    direct = _successful_direct([])

    def capture(*args, **kwargs):
        attempts.append(kwargs["attempt"])
        return direct(*args, **kwargs)

    monkeypatch.setattr(ask_gpt.chat, "download_attachments", capture)
    assert ask_gpt.main(["--download-attachments", CID]) == 0
    assert attempts == ["direct-2"]
    assert "ended before plan" in capsys.readouterr().err


def test_daemon_ctrl_c_says_worker_may_continue(monkeypatch, tmp_path, capsys):
    _isolate_download_log(monkeypatch, tmp_path)
    token = tmp_path / "token"
    token.write_text("x")
    monkeypatch.setattr(ask_gpt, "TOKEN_FILE", token)
    monkeypatch.setattr(ask_gpt, "daemon_endpoint", lambda: "http://daemon")

    def interrupt(*args, **kwargs):
        kwargs["on_event"]({"type": "download_plan", "phase": "scanning",
                            "attempt": "daemon-1", "items": [
                                {"index": 1, "total": 1, "kind": "file", "file": "a"}],
                            "unavailable": [], "invalid": []})
        raise KeyboardInterrupt

    monkeypatch.setattr(ask_gpt, "_ask_stream", interrupt)
    monkeypatch.setattr(ask_gpt.chat, "download_attachments",
                        lambda *args, **kwargs: pytest.fail("direct recovery started"))
    assert ask_gpt.main(["--download-attachments", CID, "--out", "downloads"]) == 130
    err = capsys.readouterr().err
    assert "stopped watching after 0/1" in err
    assert "daemon may continue writing to downloads" in err


def test_filesystem_failure_is_fatal_and_stops_remaining_items(monkeypatch, tmp_path):
    browser = MultiTurnBrowser(
        per_turn=[{"images": ["https://x/a", "https://x/b"],
                   "files": [], "downloads": []}], blobs=[])
    monkeypatch.setattr(chat, "open_resume", lambda m, conversation_id, expected_account: None)
    attempted = []

    def fail(m, item, out_dir, used):
        attempted.append(item.index)
        raise PermissionError("read-only output")

    monkeypatch.setattr(chat, "_download_item", fail)
    events = []
    with pytest.raises(PermissionError):
        chat.download_attachments_on(
            browser, CID, expected_account="owner@example.com", out_dir=tmp_path, attempt="direct-1", on_event=events.append)
    assert attempted == [1]
    assert [(event["type"], event.get("state")) for event in events[-2:]] == [
        ("download", "failed"), ("error", None)]


def test_empty_manifest_emits_plan_and_done_without_fake_items(monkeypatch, tmp_path):
    browser = MultiTurnBrowser(
        per_turn=[{"images": [], "files": [], "downloads": []}], blobs=[])
    monkeypatch.setattr(chat, "open_resume", lambda m, conversation_id, expected_account: None)
    events = []
    result = chat.download_attachments_on(
        browser, CID, expected_account="owner@example.com", out_dir=tmp_path, attempt="direct-1", on_event=events.append)
    plan = next(event for event in events if event["type"] == "download_plan")
    assert plan["items"] == []
    assert result["planned"] == 0 and result["files"] == []
    assert events[-1]["type"] == "done"


def test_renderer_start_failure_is_disabled_and_never_escapes(monkeypatch):
    progress = ask_gpt.DownloadProgress(
        is_tty=False, clock=lambda: 0.0, interval=30.0, out_dir=Path("ask-gpt"))
    monkeypatch.setattr(progress, "_safe_print",
                        lambda text: (_ for _ in ()).throw(OSError("stderr closed")))
    progress.start()
    assert progress._disabled
    assert progress._ticker is None


def test_direct_filesystem_error_stops_progress_without_traceback(
        monkeypatch, tmp_path, capsys):
    _isolate_download_log(monkeypatch, tmp_path)
    monkeypatch.setattr(ask_gpt, "daemon_endpoint", lambda: None)

    def fail(conversation_id, *, expected_account, out_dir, mode, reseed, on_event, attempt):
        assert expected_account == "owner@example.com"
        on_event({"type": "download_plan", "phase": "scanning", "attempt": attempt,
                  "items": [{"index": 1, "total": 1, "kind": "file", "file": "a"}],
                  "unavailable": [], "invalid": []})
        raise PermissionError("read-only output")

    monkeypatch.setattr(ask_gpt.chat, "download_attachments", fail)
    assert ask_gpt.main(["--download-attachments", CID]) == 1
    err = capsys.readouterr().err
    assert "read-only output" in err
    assert "Traceback" not in err


def test_partial_blob_write_leaves_no_destination_or_temp_and_keeps_prior_output(
        monkeypatch, tmp_path):
    browser = MultiTurnBrowser(
        per_turn=[{"images": ["https://x/first", "https://x/second"],
                   "files": [], "downloads": []}],
        blobs=[{"type": "image/png", "data": data_url(b"first-complete")},
               {"type": "image/png", "data": data_url(b"second-partial")}])
    monkeypatch.setattr(chat, "open_resume", lambda m, conversation_id, expected_account: None)
    real_fdopen = chat.os.fdopen
    opened = 0

    class PartialWriter:
        def __init__(self, fd):
            self.fd = fd

        def __enter__(self):
            return self

        def write(self, data):
            os.write(self.fd, data[:3])
            return 3

        def flush(self):
            pass

        def __exit__(self, *_):
            os.close(self.fd)

    def fail_second(fd, *args, **kwargs):
        nonlocal opened
        opened += 1
        return PartialWriter(fd) if opened == 2 else real_fdopen(fd, *args, **kwargs)

    monkeypatch.setattr(chat.os, "fdopen", fail_second)
    with pytest.raises(OSError, match="short write"):
        chat.download_attachments_on(
            browser, CID, expected_account="owner@example.com", out_dir=tmp_path, attempt="direct-1", on_event=lambda event: None)

    assert (tmp_path / "generated-1.png").read_bytes() == b"first-complete"
    assert not (tmp_path / "generated-2.png").exists()
    assert not list(tmp_path.glob(".*.tmp"))


def test_atomic_publish_never_overwrites_destination_created_at_publish_time(
        monkeypatch, tmp_path):
    destination = tmp_path / "result.bin"
    real_link = chat.os.link

    def race(temp_path, publish_path):
        Path(publish_path).write_bytes(b"raced-writer")
        return real_link(temp_path, publish_path)

    monkeypatch.setattr(chat.os, "link", race)
    with pytest.raises(chat.ChatError, match="refusing to overwrite"):
        chat._write_blob_atomic(destination, b"our-content")

    assert destination.read_bytes() == b"raced-writer"
    assert not list(tmp_path.glob(".*.tmp"))