fix(download): include file identity in queue/history dedup (#1058)

Distinct files of the same model version queued before a backend restart
were silently collapsed by deduplicate(), which grouped rows by
(model_id, model_version_id) only. Extract the file id from file_params
via json_extract and add it to the dedup key; rows without file identity
keep the old per-version behavior (NULL matches NULL).

tests/services/test_download_queue_service.py

@@ -7,6 +7,7 @@ compatibility with ``id``.
 
 import json
 import sqlite3
+import time
 from pathlib import Path
 
 import pytest
@@ -321,3 +322,176 @@ async def test_legacy_history_db_gains_file_params_column(tmp_path: Path) -> Non
     item = await svc.retry_from_history(download_id="dl-legacy-fp")
     assert item is not None
     assert json.loads(item["file_params"]) == {"id": 5}
+
+
+# ---------------------------------------------------------------------------
+# deduplicate() — file identity in the dedup key (#1058)
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_dedup_queue_keeps_distinct_files_of_same_version(tmp_path: Path) -> None:
+    """Two queued downloads of different files of one version both survive."""
+    svc = _make_service(tmp_path)
+    await svc.add_to_queue(
+        download_id="dl-a",
+        model_id=1,
+        model_version_id=100,
+        file_params={"id": 1001, "type": "Model"},
+    )
+    await svc.add_to_queue(
+        download_id="dl-b",
+        model_id=1,
+        model_version_id=100,
+        file_params={"id": 1002, "type": "Model"},
+    )
+
+    result = await svc.deduplicate()
+
+    assert result["removed_queue"] == 0
+    queue = await svc.get_queue()
+    assert sorted(item["download_id"] for item in queue) == ["dl-a", "dl-b"]
+
+
+@pytest.mark.asyncio
+async def test_dedup_queue_collapses_same_file_and_legacy_rows(tmp_path: Path) -> None:
+    """Same version + same/absent file id still collapses to the latest row."""
+    svc = _make_service(tmp_path)
+    # Same file id -> only the most recently enqueued row survives
+    await svc.add_to_queue(
+        download_id="dl-old",
+        model_id=1,
+        model_version_id=100,
+        file_params={"id": 1001},
+    )
+    await svc.add_to_queue(
+        download_id="dl-new",
+        model_id=1,
+        model_version_id=100,
+        file_params={"id": 1001},
+    )
+    # Rows without file identity keep the old per-version behavior
+    await svc.add_to_queue(
+        download_id="dl-legacy-old", model_id=2, model_version_id=200
+    )
+    await svc.add_to_queue(
+        download_id="dl-legacy-new", model_id=2, model_version_id=200
+    )
+
+    result = await svc.deduplicate()
+
+    assert result["removed_queue"] == 2
+    remaining = {item["download_id"] for item in await svc.get_queue()}
+    assert remaining == {"dl-new", "dl-legacy-new"}
+
+
+@pytest.mark.asyncio
+async def test_dedup_queue_does_not_mix_null_and_file_id(tmp_path: Path) -> None:
+    """A row with a file id never dedups against a row without one."""
+    svc = _make_service(tmp_path)
+    await svc.add_to_queue(
+        download_id="dl-file",
+        model_id=1,
+        model_version_id=100,
+        file_params={"id": 1001},
+    )
+    await svc.add_to_queue(
+        download_id="dl-nofile", model_id=1, model_version_id=100
+    )
+
+    result = await svc.deduplicate()
+
+    assert result["removed_queue"] == 0
+    assert len(await svc.get_queue()) == 2
+
+
+@pytest.mark.asyncio
+async def test_dedup_queue_unparseable_file_params_treated_as_none(tmp_path: Path) -> None:
+    """Corrupt file_params JSON falls back to the NULL file identity."""
+    svc = _make_service(tmp_path)
+    await svc.add_to_queue(
+        download_id="dl-plain", model_id=1, model_version_id=100
+    )
+    conn = sqlite3.connect(str(tmp_path / "queue.sqlite"))
+    conn.execute(
+        "INSERT INTO download_queue (download_id, model_id, model_version_id, "
+        "file_params, status, added_at) VALUES (?, ?, ?, ?, 'queued', ?)",
+        ("dl-corrupt", 1, 100, "{not json", time.time()),
+    )
+    conn.commit()
+    conn.close()
+
+    result = await svc.deduplicate()
+
+    assert result["removed_queue"] == 1
+    assert len(await svc.get_queue()) == 1
+
+
+@pytest.mark.asyncio
+async def test_dedup_history_keeps_distinct_files_of_same_version(tmp_path: Path) -> None:
+    """History rows of different files never collapse, even across statuses."""
+    svc = _make_service(tmp_path)
+    await svc.add_to_history(
+        download_id="dl-h1",
+        model_id=1,
+        model_version_id=100,
+        status="completed",
+        file_params={"id": 1001},
+    )
+    await svc.add_to_history(
+        download_id="dl-h2",
+        model_id=1,
+        model_version_id=100,
+        status="completed",
+        file_params={"id": 1002},
+    )
+    # A failed entry for file 1002 must not remove file 1001's completed row
+    await svc.add_to_history(
+        download_id="dl-h3",
+        model_id=1,
+        model_version_id=100,
+        status="failed",
+        file_params={"id": 1002},
+    )
+
+    result = await svc.deduplicate()
+
+    assert result["removed_history"] == 1  # only dl-h3 collapses (same file)
+    history = await svc.get_history()
+    remaining = {item["download_id"] for item in history["items"]}
+    assert remaining == {"dl-h1", "dl-h2"}
+
+
+@pytest.mark.asyncio
+async def test_dedup_history_collapses_same_file_and_legacy_rows(tmp_path: Path) -> None:
+    """Same file id / absent file id history rows still dedup as before."""
+    svc = _make_service(tmp_path)
+    # Same file id + same status -> keep the most recent row
+    await svc.add_to_history(
+        download_id="dl-x1",
+        model_id=1,
+        model_version_id=100,
+        status="completed",
+        file_params={"id": 1001},
+    )
+    await svc.add_to_history(
+        download_id="dl-x2",
+        model_id=1,
+        model_version_id=100,
+        status="completed",
+        file_params={"id": 1001},
+    )
+    # Legacy rows without file identity: cross-status dedup still applies
+    await svc.add_to_history(
+        download_id="dl-y1", model_id=2, model_version_id=200, status="failed"
+    )
+    await svc.add_to_history(
+        download_id="dl-y2", model_id=2, model_version_id=200, status="completed"
+    )
+
+    result = await svc.deduplicate()
+
+    assert result["removed_history"] == 2
+    history = await svc.get_history()
+    remaining = {item["download_id"] for item in history["items"]}
+    assert remaining == {"dl-x2", "dl-y2"}