fix: prevent query-sensitive video collisions and unify ingestion identity

README.md

@@ -42,6 +42,17 @@ vra status                                          # counts + jobs
 curl -s localhost:8080/rss?limit=20                 # consume the output
 ```
 
+## Queue deadlines
+
+New job deadlines accept datetimes or ISO-8601 strings and are stored in UTC
+with six-digit microsecond precision. Naive values are interpreted as UTC;
+invalid values are rejected even when the payload matches an existing job.
+This prevents timezone offsets from changing when a job becomes due.
+
+Existing database rows are not rewritten by this change. Historical non-UTC
+`run_at` values require a separately reviewed, transactional normalization
+before replay; no automatic migration or database reset is performed.
+
 ## The stack (one posture — `vra engine` prints these exact commands)
 
 | role | model · quant | size |
@@ -142,3 +153,18 @@ Architecture and frozen contracts: `docs/v2/SPEC.md`. Per-stage evidence
 - An OpenAI-compatible engine serving `VRA_MODEL` — see `vra engine`
   for the exact per-platform commands; ASR server optional
   (`VRA_ASR=off` to skip)
+
+### Source identity compatibility
+
+New feed and manual/API submissions share a versioned `source:v2:` SHA-256
+identity of the trimmed acquisition URL or path. Query parameters, their order,
+encoding, fragments and path case are preserved; feed GUID is metadata, not a
+substitute for the media source. This deliberately favors a possible duplicate
+for differently spelled URLs over silently collapsing different videos.
+
+Existing legacy IDs, video UUIDs, jobs and saved artifacts are not migrated,
+merged or deleted. Old query-stripped records may already contain mismatched
+source metadata and summaries, so they are not automatically reused. Re-ingesting
+an old source can create a new record and duplicate RSS item; historical repair
+requires explicit review/reprocessing. No database schema-version bump or reset
+is part of this change. Explicit processing by an existing video UUID is retained.

tests/v2/test_api_app.py

@@ -303,10 +303,10 @@ def test_process_requires_source_or_video_id() -> None:
     assert resp.status_code == 400
 
 
-def test_canonical_for_source_is_stable_sha1_prefix() -> None:
+def test_canonical_for_source_is_stable_versioned_source() -> None:
     assert canonical_for_source("https://x/y") == canonical_for_source("https://x/y")
     assert canonical_for_source("https://x/y") != canonical_for_source("https://x/z")
-    assert canonical_for_source("https://x/y").startswith("url:")
+    assert canonical_for_source("https://x/y").startswith("source:v2:")
 
 
 def test_jobs_list_and_filters() -> None:
@@ -488,3 +488,11 @@ def test_asr_off_satisfies_ready_logic() -> None:
     asr = report["dependencies"]["asr"]
     assert asr["enabled"] is False
     assert report["runtime_view"]["state"] == "ready"
+
+
+def test_blank_source_is_bad_request_even_when_local_paths_allowed():
+    runtime = FakeRuntime(_config(allow_local_path=True))
+    with _client(runtime) as client:
+        response = client.post('/process', json={'source_url': '   '})
+    assert response.status_code == 400
+    assert response.json()['detail'] == 'source_url must not be blank'

tests/v2/test_integration.py

@@ -292,3 +292,129 @@ async def test_asr_off_still_completes_without_transcript(tmp_path):
         assert counts["videos_by_status"].get("done", 0) == 1
     finally:
         await runtime.queue.close()
+
+
+@pytest.mark.parametrize("kind", [JobKind.INGEST, JobKind.PROCESS])
+@pytest.mark.parametrize("last_attempt", [False, True])
+async def test_cancelled_job_persists_retry_or_terminal_state(tmp_path, kind, last_attempt):
+    import asyncio
+
+    from vra.types import VideoStatus
+
+    runtime = await _make_runtime(
+        tmp_path, FakeFetcher(entries=[]), FakeAcquirer(), FakeTranscriber(),
+        FakeExtractor(), FakeSummarizer(),
+    )
+    entered = asyncio.Event()
+    video, _ = await runtime.store.upsert_video_ref(
+        canonical_id="cancel:test", source_url="https://example.com/video",
+        feed_id=None, guid=None, title=None, published_at=None,
+    )
+    job = await runtime.queue.enqueue(kind, {"video_id": video.id})
+    if last_attempt:
+        await runtime.queue._db.execute("UPDATE jobs SET attempts = 2 WHERE id = ?", (job.id,))
+    claimed = await runtime.queue.claim((kind,))
+    assert claimed is not None
+
+    async def interrupted():
+        if kind is JobKind.PROCESS:
+            await runtime.store.set_video_status(video.id, VideoStatus.ACQUIRING)
+        entered.set()
+        await asyncio.Event().wait()
+
+    task = asyncio.create_task(_run_job(claimed, interrupted(), runtime.queue, runtime.store))
+    try:
+        await asyncio.wait_for(entered.wait(), 2)
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+        saved = await runtime.queue.get(job.id)
+        assert saved.state is (JobState.FAILED if last_attempt else JobState.QUEUED)
+        assert saved.attempts == (3 if last_attempt else 1)
+        assert saved.error == "job cancelled during execution"
+        if kind is JobKind.PROCESS:
+            assert (await runtime.store.get_video(video.id)).status is VideoStatus.FAILED
+        if not last_attempt:
+            assert saved.run_at is not None
+            # Advance the durable deadline without sleeping or contacting a service.
+            await runtime.queue._db.execute("UPDATE jobs SET run_at = NULL WHERE id = ?", (job.id,))
+            retry = await runtime.queue.claim((kind,))
+            assert retry.id == job.id and retry.attempts == 2
+            async def done():
+                return "done"
+            await _run_job(retry, done(), runtime.queue, runtime.store)
+            assert (await runtime.queue.get(job.id)).state is JobState.DONE
+    finally:
+        if not task.done():
+            task.cancel()
+            await asyncio.gather(task, return_exceptions=True)
+        await runtime.queue.close()
+
+
+async def test_runtime_close_requeues_worker_and_preserves_done_video(tmp_path):
+    import asyncio
+
+    from vra.types import VideoStatus
+
+    entered = asyncio.Event()
+
+    class CompletingProcessor:
+        async def process(self, video_id, *, force=False):
+            await runtime.store.set_video_status(video_id, VideoStatus.DONE)
+            entered.set()
+            await asyncio.Event().wait()
+
+    runtime = await _make_runtime(
+        tmp_path, FakeFetcher(entries=[]), FakeAcquirer(), FakeTranscriber(),
+        FakeExtractor(), FakeSummarizer(), cpu_workers=1,
+    )
+    runtime = dc.replace(runtime, processor=CompletingProcessor())
+    video, _ = await runtime.store.upsert_video_ref(
+        canonical_id="cancel:done", source_url="https://example.com/done",
+        feed_id=None, guid=None, title=None, published_at=None,
+    )
+    job = await runtime.queue.enqueue(JobKind.PROCESS, {"video_id": video.id})
+    try:
+        await runtime.start()
+        await asyncio.wait_for(entered.wait(), 2)
+    finally:
+        await runtime.close()
+    # A fresh connection proves recovery was committed before close returned.
+    db = await Database.connect(runtime.config.database_path)
+    try:
+        saved = await SqlJobQueue(db).get(job.id)
+        assert saved.state is JobState.QUEUED
+        assert (await SqlVideoStore(db).get_video(video.id)).status is VideoStatus.DONE
+    finally:
+        await db.close()
+
+
+@pytest.mark.parametrize("cleanup_fails", [False, True])
+async def test_built_runtime_closes_feed_and_transcription_clients(tmp_path, cleanup_fails):
+    from vra.runtime import build_runtime
+
+    runtime = await build_runtime(_cfg(tmp_path))
+    feed_http = runtime.ingester._fetcher._client
+    transcriber = runtime.processor._transcriber
+    asr_http = transcriber._client
+    if cleanup_fails:
+        class FailingSummarizer:
+            async def close(self):
+                await runtime.engine.close()
+                raise RuntimeError("cleanup failed")
+        runtime = dc.replace(runtime, summarizer=FailingSummarizer())
+    try:
+        if cleanup_fails:
+            with pytest.raises(RuntimeError, match="cleanup failed"):
+                await runtime.close()
+        else:
+            await runtime.close()
+        assert feed_http.is_closed
+        assert asr_http.is_closed
+        assert runtime.queue._db._closed
+    finally:
+        # The red-phase regression also cleans up its deliberately exposed leak.
+        await feed_http.aclose()
+        await transcriber.close()
+        await runtime.engine.close()
+        await runtime.queue.close()

tests/v2/test_source_identity.py

@@ -0,0 +1,30 @@
+from __future__ import annotations
+
+import pytest
+
+from vra.api.app import canonical_for_source
+from vra.identity import source_identity
+from vra.service.ingester import _canonical_id
+
+
+@pytest.mark.parametrize('left,right', [
+    ('https://youtube.com/watch?v=A', 'https://youtube.com/watch?v=B'),
+    ('https://x/?a=1&b=2', 'https://x/?b=2&a=1'),
+    ('https://x/?a=1&a=2', 'https://x/?a=2'),
+    ('https://x/?v=a%2Fb', 'https://x/?v=a/b'),
+    ('/Media/A.mp4', '/Media/a.mp4'),
+])
+def test_meaningful_source_distinctions_are_preserved(left, right):
+    assert source_identity(left) != source_identity(right)
+
+
+def test_feed_and_api_use_the_same_identity_independent_of_guid():
+    source = 'https://youtube.com/watch?v=A'
+    assert _canonical_id('https://unrelated-guid', source) == canonical_for_source(source)
+    assert _canonical_id('yt:video:A', source) == canonical_for_source(' ' + source + ' ')
+    assert source_identity(source).startswith('source:v2:')
+
+
+def test_blank_source_rejected():
+    with pytest.raises(ValueError):
+        source_identity('   ')

tests/v2/test_store-service_ingester.py

@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import hashlib
 from collections.abc import AsyncIterator, Callable
 from datetime import UTC, datetime
 from pathlib import Path
@@ -13,11 +12,12 @@
 
 import vra.service.ingester as ingester_module
 from vra.config import Config
+from vra.identity import source_identity
 from vra.service.ingester import FeedIngester, HttpFeedFetcher
 from vra.store.db import Database
 from vra.store.queue import SqlJobQueue
 from vra.store.repos import SqlVideoStore
-from vra.types import JobKind, JobState, VideoStatus
+from vra.types import JobKind, JobState, Summary, VideoStatus
 
 _RSS = """<?xml version="1.0" encoding="UTF-8"?>
 <rss version="2.0"><channel>
@@ -81,15 +81,15 @@ async def test_ingest_new_videos_enqueued(
     assert result.feed.last_checked is not None
     assert result.feed.error is None
 
-    # first entry: absolute http guid becomes the canonical id
-    first = await store.get_video_by_canonical("https://youtu.be/aaa")
+    # GUID is metadata; the acquisition source determines identity
+    first = await store.get_video_by_canonical(source_identity("https://youtu.be/aaa"))
     assert first is not None
     assert first.status == VideoStatus.PENDING
     assert first.guid == "https://youtu.be/aaa"
     assert first.published_at == datetime(2025, 6, 1, tzinfo=UTC)
 
-    # second entry: normalized url (host lowered, query dropped), sha1[:16]
-    expected = "url:" + hashlib.sha1(b"https://example.com/b").hexdigest()[:16]
+    # Query and case are preserved rather than guessed equivalent
+    expected = source_identity("https://Example.com/b?utm_source=x")
     second = await store.get_video_by_canonical(expected)
     assert second is not None
     assert second.source_url == "https://Example.com/b?utm_source=x"
@@ -119,9 +119,9 @@ async def test_ingest_done_skipped_and_pending_deduped(
 
     # mark both done: third poll skips entirely
     for video in [
-        await store.get_video_by_canonical("https://youtu.be/aaa"),
+        await store.get_video_by_canonical(source_identity("https://youtu.be/aaa")),
         await store.get_video_by_canonical(
-            "url:" + hashlib.sha1(b"https://example.com/b").hexdigest()[:16]
+            source_identity("https://Example.com/b?utm_source=x")
         ),
     ]:
         assert video is not None
@@ -138,9 +138,9 @@ async def test_ingest_failed_video_reenqueued(
     ingester = FeedIngester(FakeFetcher(_entries()), store, queue, Config())
     await ingester.ingest("https://example.com/rss")
 
-    first = await store.get_video_by_canonical("https://youtu.be/aaa")
+    first = await store.get_video_by_canonical(source_identity("https://youtu.be/aaa"))
     second = await store.get_video_by_canonical(
-        "url:" + hashlib.sha1(b"https://example.com/b").hexdigest()[:16]
+        source_identity("https://Example.com/b?utm_source=x")
     )
     assert first is not None and second is not None
 
@@ -222,7 +222,7 @@ async def test_ingest_skips_entries_without_source_url(
     result = await ingester.ingest("https://example.com/rss")
     assert (result.new_videos, result.enqueued) == (1, 1)
     assert await store.get_video_by_canonical(
-        "url:" + hashlib.sha1(b"https://x/1").hexdigest()[:16]
+        source_identity("https://x/1")
     ) is not None
 
 
@@ -412,3 +412,65 @@ def handle(request):
         with pytest.raises(ValueError, match='limit'):
             await HttpFeedFetcher(client, allow_private=True).fetch('http://localhost/start')
     assert len(seen) == 6
+
+
+async def test_query_distinctions_and_guid_reuse_do_not_overwrite(stack):
+    _, store, queue = stack
+    urls = ['https://youtube.com/watch?v=A', 'https://youtube.com/watch?v=B']
+    entries = [dict(source_url=url, title=url, guid='https://same-guid') for url in urls]
+    ingester = FeedIngester(FakeFetcher(entries), store, queue, Config())
+    result = await ingester.ingest('https://feed')
+    assert result.new_videos == 2
+    rows = [await store.get_video_by_canonical(source_identity(url)) for url in urls]
+    assert rows[0].id != rows[1].id
+    assert [row.source_url for row in rows] == urls
+    assert len(await queue.list(kind=JobKind.PROCESS)) == 2
+    assert (await ingester.ingest('https://feed')).new_videos == 0
+
+
+async def test_ambiguous_legacy_row_is_not_adopted_or_mutated(stack):
+    db, store, queue = stack
+    source = 'https://youtube.com/watch?v=B'
+    old, _ = await store.upsert_video_ref(
+        canonical_id='url:old-lossy-hash', source_url=source, feed_id=None,
+        guid='yt:video:B', title='old possibly mismatched artifact', published_at=None)
+    await store.set_video_status(old.id, VideoStatus.DONE)
+    await store.save_summary(old.id, Summary(summary='possibly source A, not B'))
+    old_summary = dict(await db.fetchone('SELECT * FROM summaries WHERE video_id = ?', (old.id,)))
+    old_job = await queue.enqueue(JobKind.PROCESS, {'video_id': old.id}, video_id=old.id)
+    result = await FeedIngester(FakeFetcher([dict(source_url=source, guid='yt:video:B')]),
+                                store, queue, Config()).ingest('https://feed')
+    assert result.new_videos == 1
+    current = await store.get_video_by_canonical(source_identity(source))
+    legacy = await store.get_video_by_canonical('url:old-lossy-hash')
+    assert current.id != legacy.id == old.id
+    assert legacy.status == VideoStatus.DONE
+    assert legacy.title == 'old possibly mismatched artifact'
+    assert dict(await db.fetchone('SELECT * FROM summaries WHERE video_id = ?', (old.id,))) == old_summary
+    assert any(job.id == old_job.id and job.video_id == old.id for job in await queue.list(kind=JobKind.PROCESS))
+
+
+@pytest.mark.parametrize('feed_first', [True, False])
+async def test_feed_and_api_share_record_and_default_job(stack, feed_first):
+    from types import SimpleNamespace
+
+    from vra.api.app import create_app
+    _, store, queue = stack
+    source = 'http://localhost/video?v=A'
+    config = Config(allow_private_net=True)
+    ingester = FeedIngester(FakeFetcher([dict(source_url=source, guid='yt:video:A')]),
+                            store, queue, config)
+    app = create_app(SimpleNamespace(config=config, store=store, queue=queue))
+    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app),
+                                 base_url='http://test') as client:
+        if feed_first:
+            await ingester.ingest('http://localhost/feed')
+        response = await client.post('/process', json={'source_url': source})
+        assert response.status_code == 200
+        if not feed_first:
+            await ingester.ingest('http://localhost/feed')
+    record = await store.get_video_by_canonical(source_identity(source))
+    assert response.json()['video_id'] == record.id
+    jobs = await queue.list(kind=JobKind.PROCESS)
+    assert len(jobs) == 1
+    assert jobs[0].video_id == record.id