diff --git a/README.md b/README.md
index e36fe9d..8c67365 100644
--- a/README.md
+++ b/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.
diff --git a/docs/v2/performance/queue-wakeup/README.md b/docs/v2/performance/queue-wakeup/README.md
new file mode 100644
index 0000000..7876f7d
--- /dev/null
+++ b/docs/v2/performance/queue-wakeup/README.md
@@ -0,0 +1,21 @@
+# Local queue wakeup experiment (2026-10-30)
+
+Baseline: `f938dd09510ecbd16fd62ca8428adb2401785b69`. Candidate: the sibling commit containing this report.
+
+Scope: real temporary SQLite, fake no-op processing, no network/model/device calls. Run `PROBE_LABEL=<label> PYTHONPATH=<checkout> python docs/v2/performance/queue-wakeup/probe.py` with the project's locked Python dependencies. Each worker-count trial contains 20 sparse jobs, a 32-job burst and 10 enqueues of the same future job. Sources and raw observations are included. The baseline is an unmodified worktree, not a simulated polling implementation.
+
+| Workers | Baseline sparse median/p95 ms | Candidate sparse median/p95 ms | Baseline/candidate total empty claims |
+| --- | --- | --- | --- |
+| 2 (default) | 973.58 / 977.20 | 1.41 / 4.37 | 44 / 74 |
+| 8 | 795.93 / 1754.34 | 2.98 / 5.95 | 120 / 392 |
+| 32 | 973.07 / 974.51 | 2.51 / 5.22 | 704 / 1664 |
+
+All trials processed exactly 52 distinct jobs and left zero worker tasks pending after shutdown. p95 is nearest-rank over only 20 sparse observations; this is not production p95. Host contention was uncontrolled (including concurrent unrelated validation), so retain the outliers, do not infer their cause, and do not compare total CPU across different wall durations. This experiment establishes local notification responsiveness, not model throughput, inference latency, sustained memory, idle CPU or end-to-end download speed.
+
+## Design and tradeoff
+
+Workers capture the current notification epoch before the durable claim. Enqueue rotates and signals that epoch only after commit and only for a stored queued/due job. Cancellation cannot clear another worker's signal. SQLite still decides ownership; notifications never carry jobs. The one-second fallback remains for other queue instances/processes and future retry deadlines. Schema, contracts, retry limits and processing concurrency are unchanged.
+
+Broadcast schedules every idle worker of the matching kind. Burst traffic can therefore cause extra losing claims, especially at high worker counts; the table exposes this cost rather than claiming a universal CPU win. The intermediate broadcast version generated 20/80/320 useless empty claims for the ten future-job enqueues at W=2/8/32. Matching the stored deadline to claim's lexical ordering reduced that isolated delta to zero in all three final trials. Existing legacy timestamp-format limitations are not repaired by this change.
+
+A more complex FIFO notifier was not adopted: cancellation handoff and another consumer winning a claim would require additional correctness machinery. Measure real workload pressure before making that tradeoff. Current tests cover claim/wait races, dedup, transaction rollback, future deadlines, stale recovery, both runtime worker kinds, cancellation isolation and epoch collection. A gate-based mutation test failed for both worker kinds when notification waiting was replaced by the old sleep; its success does not depend on a subsecond performance threshold.
diff --git a/docs/v2/performance/queue-wakeup/baseline.json b/docs/v2/performance/queue-wakeup/baseline.json
new file mode 100644
index 0000000..67d0607
--- /dev/null
+++ b/docs/v2/performance/queue-wakeup/baseline.json
@@ -0,0 +1 @@
+{"label": "baseline", "matrix": [{"workers": 2, "sparse": {"claims": 62, "empty": 42, "p50_ms": 973.5771255000145, "p95_ms": 977.201490997686}, "total_claims": 96, "total_empty": 44, "future_duplicate_empty_delta": 0, "completed": 52, "cpu_s": 0.2427470500000002, "wall_s": 21.48554625900215, "workers_pending_after_shutdown": 0}, {"workers": 8, "sparse": {"claims": 132, "empty": 112, "p50_ms": 795.928465500765, "p95_ms": 1754.3417900014902}, "total_claims": 172, "total_empty": 120, "future_duplicate_empty_delta": 1, "completed": 52, "cpu_s": 5.076605553, "wall_s": 20.55640929600122, "workers_pending_after_shutdown": 0}, {"workers": 32, "sparse": {"claims": 692, "empty": 672, "p50_ms": 973.0669439995836, "p95_ms": 974.506904000009}, "total_claims": 756, "total_empty": 704, "future_duplicate_empty_delta": 0, "completed": 52, "cpu_s": 0.3201224079999996, "wall_s": 21.41443181700015, "workers_pending_after_shutdown": 0}]}
diff --git a/docs/v2/performance/queue-wakeup/candidate.json b/docs/v2/performance/queue-wakeup/candidate.json
new file mode 100644
index 0000000..095eead
--- /dev/null
+++ b/docs/v2/performance/queue-wakeup/candidate.json
@@ -0,0 +1 @@
+{"label": "due-filtered-candidate", "matrix": [{"workers": 2, "sparse": {"claims": 62, "empty": 42, "p50_ms": 1.4125375000730855, "p95_ms": 4.3662630014296155}, "total_claims": 126, "total_empty": 74, "future_duplicate_empty_delta": 0, "completed": 52, "cpu_s": 0.09358409400000003, "wall_s": 1.1074707369989483, "workers_pending_after_shutdown": 0}, {"workers": 8, "sparse": {"claims": 188, "empty": 168, "p50_ms": 2.9835054992872756, "p95_ms": 5.952845000138041}, "total_claims": 444, "total_empty": 392, "future_duplicate_empty_delta": 0, "completed": 52, "cpu_s": 0.29827248599999995, "wall_s": 1.3256090899994888, "workers_pending_after_shutdown": 0}, {"workers": 32, "sparse": {"claims": 692, "empty": 672, "p50_ms": 2.5117889999819454, "p95_ms": 5.2226640000299085}, "total_claims": 1716, "total_empty": 1664, "future_duplicate_empty_delta": 0, "completed": 52, "cpu_s": 0.6735719319999999, "wall_s": 1.6200977379994583, "workers_pending_after_shutdown": 0}]}
diff --git a/docs/v2/performance/queue-wakeup/intermediate-broadcast.json b/docs/v2/performance/queue-wakeup/intermediate-broadcast.json
new file mode 100644
index 0000000..2a86603
--- /dev/null
+++ b/docs/v2/performance/queue-wakeup/intermediate-broadcast.json
@@ -0,0 +1 @@
+{"label": "candidate", "matrix": [{"workers": 2, "sparse": {"claims": 62, "empty": 42, "p50_ms": 1.9677584987221053, "p95_ms": 10.104683999088593}, "total_claims": 146, "total_empty": 94, "future_duplicate_empty_delta": 20, "completed": 52, "cpu_s": 0.13998825400000003, "wall_s": 1.1897931290004635, "workers_pending_after_shutdown": 0}, {"workers": 8, "sparse": {"claims": 188, "empty": 168, "p50_ms": 1.7775479991541943, "p95_ms": 3.8986910003586672}, "total_claims": 524, "total_empty": 472, "future_duplicate_empty_delta": 80, "completed": 52, "cpu_s": 0.23946446499999996, "wall_s": 1.2175366330011457, "workers_pending_after_shutdown": 0}, {"workers": 32, "sparse": {"claims": 692, "empty": 672, "p50_ms": 2.3371019997284748, "p95_ms": 4.613050001353258}, "total_claims": 2036, "total_empty": 1984, "future_duplicate_empty_delta": 320, "completed": 52, "cpu_s": 0.7991928630000001, "wall_s": 1.721258014000341, "workers_pending_after_shutdown": 0}]}
diff --git a/docs/v2/performance/queue-wakeup/probe.py b/docs/v2/performance/queue-wakeup/probe.py
new file mode 100644
index 0000000..062f0d6
--- /dev/null
+++ b/docs/v2/performance/queue-wakeup/probe.py
@@ -0,0 +1,121 @@
+import asyncio
+import json
+import os
+import statistics
+import tempfile
+import time
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+from types import SimpleNamespace
+
+from vra.runtime import _process_worker
+from vra.store.db import Database
+from vra.store.queue import SqlJobQueue
+from vra.store.repos import SqlVideoStore
+from vra.types import JobKind
+
+
+async def run(n):
+    with tempfile.TemporaryDirectory() as tmp:
+        db = await Database.connect(str(Path(tmp) / "q.sqlite"))
+        await db.migrate()
+        q = SqlJobQueue(db)
+        claims = 0
+        empty = 0
+        observed = asyncio.Queue()
+        starts = {}
+        samples = []
+        completed = []
+        original = q.claim
+
+        async def claim(kinds):
+            nonlocal claims, empty
+            job = await original(kinds)
+            claims += 1
+            if job is None:
+                empty += 1
+            else:
+                key = job.payload["video_id"]
+                samples.append((time.perf_counter() - starts[key]) * 1000)
+                observed.put_nowait(key)
+            return job
+
+        q.claim = claim
+
+        async def process(key, **kw):
+            completed.append(key)
+
+        stop = asyncio.Event()
+        tasks = [
+            asyncio.create_task(
+                _process_worker(
+                    q, SqlVideoStore(db), SimpleNamespace(process=process), stop
+                )
+            )
+            for _ in range(n)
+        ]
+        cpu = time.process_time()
+        wall = time.perf_counter()
+        try:
+            await asyncio.sleep(0.1)
+            for i in range(20):
+                key = f"sparse-{i}"
+                starts[key] = time.perf_counter()
+                await q.enqueue(JobKind.PROCESS, {"video_id": key})
+                assert await asyncio.wait_for(observed.get(), 3) == key
+                await asyncio.sleep(0.03)
+            sparse = {
+                "claims": claims,
+                "empty": empty,
+                "p50_ms": statistics.median(samples),
+                "p95_ms": sorted(samples)[18],
+            }
+            for i in range(32):
+                key = f"burst-{i}"
+                starts[key] = time.perf_counter()
+                await q.enqueue(JobKind.PROCESS, {"video_id": key})
+            burst = [await asyncio.wait_for(observed.get(), 3) for _ in range(32)]
+            assert len(set(burst)) == 32
+            await asyncio.sleep(0.1)
+            before = empty
+            for _ in range(10):
+                await q.enqueue(
+                    JobKind.PROCESS,
+                    {"video_id": "future"},
+                    run_at=datetime.now(UTC) + timedelta(minutes=1),
+                )
+                await asyncio.sleep(0.01)
+            await asyncio.sleep(0.1)
+            assert len(completed) == 52 and len(set(completed)) == 52
+            result = {
+                "workers": n,
+                "sparse": sparse,
+                "total_claims": claims,
+                "total_empty": empty,
+                "future_duplicate_empty_delta": empty - before,
+                "completed": len(completed),
+                "cpu_s": time.process_time() - cpu,
+                "wall_s": time.perf_counter() - wall,
+            }
+        finally:
+            stop.set()
+            for t in tasks:
+                t.cancel()
+            await asyncio.gather(*tasks, return_exceptions=True)
+            result["workers_pending_after_shutdown"] = sum(not t.done() for t in tasks)
+            await db.close()
+        return result
+
+
+async def main():
+    print(
+        json.dumps(
+            {
+                "label": os.environ["PROBE_LABEL"],
+                "matrix": [await run(n) for n in [2, 8, 32]],
+            }
+        )
+    )
+
+
+asyncio.run(main())
diff --git a/tests/v2/test_api_app.py b/tests/v2/test_api_app.py
index 2b2296b..a567f5f 100644
--- a/tests/v2/test_api_app.py
+++ b/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'
diff --git a/tests/v2/test_integration.py b/tests/v2/test_integration.py
index a8d450d..b7f3f67 100644
--- a/tests/v2/test_integration.py
+++ b/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()
diff --git a/tests/v2/test_queue_wakeup.py b/tests/v2/test_queue_wakeup.py
new file mode 100644
index 0000000..377ea0a
--- /dev/null
+++ b/tests/v2/test_queue_wakeup.py
@@ -0,0 +1,279 @@
+"""Offline wakeup/cancellation regressions; SQLite, not the notification, owns jobs."""
+
+import asyncio
+import contextlib
+import gc
+import weakref
+from datetime import UTC, datetime, timedelta
+from types import SimpleNamespace
+
+import pytest
+
+from vra.runtime import _ingest_worker, _process_worker
+from vra.store.db import Database
+from vra.store.queue import SqlJobQueue
+from vra.store.repos import SqlVideoStore
+from vra.types import JobKind
+
+
+@pytest.fixture
+async def queue(tmp_path):
+    db = await Database.connect(str(tmp_path / "queue.sqlite"))
+    await db.migrate()
+    try:
+        yield SqlJobQueue(db)
+    finally:
+        await db.close()
+
+
+async def cancel(task):
+    task.cancel()
+    with contextlib.suppress(asyncio.CancelledError):
+        await task
+
+
+async def test_enqueue_notifies_after_commit_and_only_its_kind(queue, monkeypatch):
+    ingest = queue.work_wakeup(JobKind.INGEST)
+    process = asyncio.create_task(queue.work_wakeup(JobKind.PROCESS)())
+    states = []
+    original = queue._notify_work
+
+    def notify(kind):
+        states.append(queue._db._conn.in_transaction)
+        original(kind)
+
+    monkeypatch.setattr(queue, "_notify_work", notify)
+    try:
+        job = await queue.enqueue(JobKind.INGEST, {"url": "fixture://one"})
+        assert await asyncio.wait_for(ingest(), 1)
+        assert states == [False]
+        await asyncio.sleep(0)
+        assert not process.done()
+        assert (await queue.claim((JobKind.INGEST,))).id == job.id
+    finally:
+        await cancel(process)
+
+
+async def test_enqueue_between_empty_claim_and_wait_is_not_lost(queue):
+    wakeup = queue.work_wakeup(JobKind.PROCESS)
+    assert await queue.claim((JobKind.PROCESS,)) is None
+    job = await queue.enqueue(JobKind.PROCESS, {"video_id": "one"})
+    assert await asyncio.wait_for(wakeup(), 1)
+    assert (await queue.claim((JobKind.PROCESS,))).id == job.id
+
+
+async def test_multiple_waiters_do_not_consume_each_others_generation(queue):
+    first = queue.work_wakeup(JobKind.PROCESS)
+    second = queue.work_wakeup(JobKind.PROCESS)
+    job = await queue.enqueue(JobKind.PROCESS, {"video_id": "one"})
+    assert await asyncio.wait_for(asyncio.gather(first(), second()), 1) == [True, True]
+    claims = await asyncio.gather(
+        queue.claim((JobKind.PROCESS,)), queue.claim((JobKind.PROCESS,))
+    )
+    assert [claim.id for claim in claims if claim] == [job.id]
+    next_generation = asyncio.create_task(queue.work_wakeup(JobKind.PROCESS)())
+    try:
+        await asyncio.sleep(0)
+        assert not next_generation.done()
+        await queue.enqueue(JobKind.PROCESS, {"video_id": "two"})
+        assert await asyncio.wait_for(next_generation, 1)
+    finally:
+        await cancel(next_generation)
+
+
+async def test_queued_duplicate_wakes_but_running_duplicate_does_not(queue):
+    shadow = SqlJobQueue(queue._db)
+    job = await shadow.enqueue(JobKind.PROCESS, {"video_id": "one"})
+    wakeup = queue.work_wakeup(JobKind.PROCESS)
+    duplicate = await queue.enqueue(JobKind.PROCESS, {"video_id": "one"})
+    assert duplicate.id == job.id
+    assert await asyncio.wait_for(wakeup(), 1)
+    await queue.claim((JobKind.PROCESS,))
+    waiting = asyncio.create_task(queue.work_wakeup(JobKind.PROCESS)())
+    try:
+        assert (await queue.enqueue(JobKind.PROCESS, {"video_id": "one"})).id == job.id
+        await asyncio.sleep(0)
+        assert not waiting.done()
+    finally:
+        await cancel(waiting)
+
+
+async def test_failed_transaction_does_not_signal_and_future_job_stays_not_due(queue):
+    waiting = asyncio.create_task(queue.work_wakeup(JobKind.PROCESS)())
+    try:
+        with pytest.raises(Exception, match="FOREIGN KEY"):
+            await queue.enqueue(JobKind.PROCESS, {"video_id": "missing"}, video_id="missing")
+        await asyncio.sleep(0)
+        assert not waiting.done()
+        await queue.enqueue(
+            JobKind.PROCESS, {"video_id": "future"},
+            run_at=datetime.now(UTC) + timedelta(minutes=1),
+        )
+        await asyncio.sleep(0)
+        assert not waiting.done()
+        await queue.enqueue(JobKind.PROCESS, {"video_id": "future"})
+        await asyncio.sleep(0)
+        assert not waiting.done()  # dedup preserves the existing future deadline
+        assert await queue.claim((JobKind.PROCESS,)) is None
+    finally:
+        await cancel(waiting)
+
+
+async def test_stale_recovery_notifies_after_making_jobs_due(queue):
+    job = await queue.enqueue(JobKind.PROCESS, {"video_id": "stale"})
+    await queue.claim((JobKind.PROCESS,))
+    await queue._db.execute(
+        "UPDATE jobs SET updated_at = ? WHERE id = ?", ("2000-01-01T00:00:00.000000+00:00", job.id)
+    )
+    wakeup = queue.work_wakeup(JobKind.PROCESS)
+    assert await queue.requeue_stale(60) == 1
+    assert await asyncio.wait_for(wakeup(), 1)
+    assert (await queue.claim((JobKind.PROCESS,))).id == job.id
+
+
+async def test_rotation_does_not_retain_notification_history(queue):
+    references = []
+    for index in range(100):
+        token = queue.work_wakeup(JobKind.PROCESS)
+        references.append(weakref.ref(token.__self__))
+        await queue.enqueue(JobKind.PROCESS, {"video_id": str(index)})
+    del token
+    gc.collect()
+    assert all(reference() is None for reference in references)
+
+
+@pytest.mark.parametrize("kind", [JobKind.INGEST, JobKind.PROCESS])
+@pytest.mark.parametrize("other_instance", [False, True])
+async def test_real_worker_observes_local_wakeup_or_external_poll_fallback(
+    queue, monkeypatch, kind, other_instance
+):
+    empty = asyncio.Event()
+    started = asyncio.Event()
+    original_claim = queue.claim
+
+    async def claim(kinds):
+        job = await original_claim(kinds)
+        if job is None:
+            empty.set()
+        return job
+
+    async def consume(*_args, **_kwargs):
+        started.set()
+        return None
+
+    monkeypatch.setattr(queue, "claim", claim)
+    service = SimpleNamespace(ingest=consume, process=consume)
+    stop = asyncio.Event()
+    worker = _ingest_worker if kind is JobKind.INGEST else _process_worker
+    task = asyncio.create_task(worker(queue, SqlVideoStore(queue._db), service, stop))
+    try:
+        await asyncio.wait_for(empty.wait(), 3)
+        target = SqlJobQueue(queue._db) if other_instance else queue
+        await target.enqueue(kind, {"url": "fixture://one", "video_id": "one"})
+        await asyncio.wait_for(started.wait(), 3)
+    finally:
+        stop.set()
+        await cancel(task)
+    assert task.done()
+
+
+@pytest.mark.parametrize("kind", [JobKind.INGEST, JobKind.PROCESS])
+async def test_local_worker_awaits_notification_before_claiming_again(queue, monkeypatch, kind):
+    """A polling-only worker never enters this gate, regardless of its speed."""
+    # Keep the fallback outside the watchdog; the gate, not elapsed speed, proves wakeup.
+    monkeypatch.setattr("vra.runtime._POLL_INTERVAL_S", 10.0)
+    waiting = asyncio.Event()
+    notified = asyncio.Event()
+    release = asyncio.Event()
+    started = asyncio.Event()
+    original = queue.work_wakeup
+
+    def snapshot(job_kind):
+        token = original(job_kind)
+
+        async def wait():
+            waiting.set()
+            result = await token()
+            notified.set()
+            await release.wait()
+            return result
+
+        return wait
+
+    async def consume(*_args, **_kwargs):
+        started.set()
+
+    monkeypatch.setattr(queue, "work_wakeup", snapshot)
+    service = SimpleNamespace(ingest=consume, process=consume)
+    stop = asyncio.Event()
+    worker = _ingest_worker if kind is JobKind.INGEST else _process_worker
+    task = asyncio.create_task(worker(queue, SqlVideoStore(queue._db), service, stop))
+    try:
+        await asyncio.wait_for(waiting.wait(), 3)
+        await queue.enqueue(kind, {"url": "fixture://one", "video_id": "one"})
+        await asyncio.wait_for(notified.wait(), 3)
+        assert not started.is_set()
+        release.set()
+        await asyncio.wait_for(started.wait(), 3)
+    finally:
+        stop.set()
+        await cancel(task)
+
+
+async def test_cancelled_waiter_does_not_consume_survivor_notification(queue):
+    for index in range(20):
+        token = queue.work_wakeup(JobKind.PROCESS)
+        event = token.__self__
+        cancelled = asyncio.create_task(token())
+        survivor = asyncio.create_task(token())
+        try:
+            await asyncio.sleep(0)
+            await cancel(cancelled)
+            assert len(event._waiters) == 1
+            job = await queue.enqueue(JobKind.PROCESS, {"video_id": f"cancel-{index}"})
+            assert await asyncio.wait_for(survivor, 1)
+            assert not event._waiters
+            assert (await queue.claim((JobKind.PROCESS,))).id == job.id
+            assert await queue.claim((JobKind.PROCESS,)) is None
+        finally:
+            await cancel(cancelled)
+            await cancel(survivor)
+
+
+async def test_future_local_job_runs_via_unchanged_fallback(queue, monkeypatch):
+    started = asyncio.Event()
+    waiting = asyncio.Event()
+    original = queue.work_wakeup
+    notified = []
+
+    def snapshot(kind):
+        token = original(kind)
+
+        async def wait():
+            waiting.set()
+            result = await token()
+            notified.append(True)
+            return result
+
+        return wait
+
+    monkeypatch.setattr(queue, "work_wakeup", snapshot)
+    deadline = None
+
+    async def consume(*_args, **_kwargs):
+        assert deadline is not None and datetime.now(UTC) >= deadline
+        started.set()
+
+    stop = asyncio.Event()
+    task = asyncio.create_task(_process_worker(
+        queue, SqlVideoStore(queue._db), SimpleNamespace(process=consume), stop
+    ))
+    try:
+        await asyncio.wait_for(waiting.wait(), 3)
+        deadline = datetime.now(UTC) + timedelta(milliseconds=100)
+        await queue.enqueue(JobKind.PROCESS, {"video_id": "later"}, run_at=deadline)
+        await asyncio.wait_for(started.wait(), 3)
+        assert notified == []  # the runtime's unchanged timeout, not a local signal, drove the claim
+    finally:
+        stop.set()
+        await cancel(task)
diff --git a/tests/v2/test_source_identity.py b/tests/v2/test_source_identity.py
new file mode 100644
index 0000000..a84bd33
--- /dev/null
+++ b/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('   ')
diff --git a/tests/v2/test_store-service_ingester.py b/tests/v2/test_store-service_ingester.py
index ffb594f..6e830e1 100644
--- a/tests/v2/test_store-service_ingester.py
+++ b/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
diff --git a/tests/v2/test_store-service_queue.py b/tests/v2/test_store-service_queue.py
index c24eca6..989f541 100644
--- a/tests/v2/test_store-service_queue.py
+++ b/tests/v2/test_store-service_queue.py
@@ -238,3 +238,59 @@ async def test_queue_close_delegates_to_database(tmp_path: Path) -> None:
     queue = SqlJobQueue(db)
     await queue.close()
     assert db.path == str(tmp_path / "vra.db")  # attributes intact after close
+
+
+@pytest.mark.parametrize("as_string", [False, True])
+async def test_offset_deadlines_use_utc_instant(queue: SqlJobQueue, monkeypatch, as_string: bool) -> None:
+    from datetime import timezone
+
+    from vra.store import queue as module
+
+    now = datetime(2026, 10, 29, 12, tzinfo=UTC)
+    monkeypatch.setattr(module, "_now_iso", lambda: module._iso(now))
+    past = datetime(2026, 10, 29, 19, tzinfo=timezone(timedelta(hours=8)))
+    future = datetime(2026, 10, 29, 5, tzinfo=timezone(timedelta(hours=-8)))
+    later = await queue.enqueue(JobKind.PROCESS, {"offset": "future"},
+                                run_at=future.isoformat() if as_string else future)
+    due = await queue.enqueue(JobKind.PROCESS, {"offset": "past"},
+                              run_at=past.isoformat() if as_string else past)
+    assert due.run_at == datetime(2026, 10, 29, 11, tzinfo=UTC)
+    assert later.run_at == datetime(2026, 10, 29, 13, tzinfo=UTC)
+    assert (await queue.claim((JobKind.PROCESS,))).id == due.id
+    assert await queue.claim((JobKind.PROCESS,)) is None
+
+
+async def test_deadline_microsecond_boundary(queue: SqlJobQueue, monkeypatch) -> None:
+    from vra.store import queue as module
+
+    now = datetime(2026, 10, 29, 12, tzinfo=UTC)
+    monkeypatch.setattr(module, "_now_iso", lambda: module._iso(now))
+    await queue.enqueue(JobKind.PROCESS, {"future": 1}, run_at=now + timedelta(microseconds=1))
+    due = await queue.enqueue(JobKind.PROCESS, {"due": 1}, run_at=now.replace(tzinfo=None))
+    assert (await queue.claim((JobKind.PROCESS,))).id == due.id
+    assert await queue.claim((JobKind.PROCESS,)) is None
+
+
+@pytest.mark.parametrize("value", ["not a date", "2026-13-01", 123, [], {}])
+async def test_invalid_deadline_never_inserts(queue: SqlJobQueue, value) -> None:
+    with pytest.raises((TypeError, ValueError)):
+        await queue.enqueue(JobKind.PROCESS, {"invalid": str(value)}, run_at=value)
+    assert await queue.list() == []
+
+
+def test_deadline_canonical_serialization() -> None:
+    from vra.store.queue import _iso
+
+    assert _iso(None) is None
+    expected = "2026-10-29T12:00:00.000000+00:00"
+    for value in ["2026-10-29T12:00:00Z", "2026-10-29T20:00:00+08:00",
+                  "2026-10-29T12:00:00", datetime(2026, 10, 29, 12)]:
+        assert _iso(value) == expected
+
+
+async def test_invalid_duplicate_deadline_is_rejected(queue: SqlJobQueue) -> None:
+    original = await queue.enqueue(JobKind.PROCESS, {"same": 1})
+    with pytest.raises(ValueError):
+        await queue.enqueue(JobKind.PROCESS, {"same": 1}, run_at="invalid")
+    jobs = await queue.list()
+    assert len(jobs) == 1 and jobs[0].id == original.id
diff --git a/vra/api/app.py b/vra/api/app.py
index b990811..a68351b 100644
--- a/vra/api/app.py
+++ b/vra/api/app.py
@@ -25,7 +25,6 @@
 from __future__ import annotations
 
 import asyncio
-import hashlib
 import importlib.resources
 import ipaddress
 import logging
@@ -55,6 +54,7 @@
     setup_payload,
 )
 from vra.config import AsrMode, Config
+from vra.identity import source_identity
 from vra.rss import RssWriter
 from vra.types import JobKind, JobState
 
@@ -97,7 +97,7 @@ class ProcessBody(BaseModel):
 
 def canonical_for_source(source: str) -> str:
     """Cheap canonical id for a source URL/path before platform resolution."""
-    return "url:" + hashlib.sha1(source.encode("utf-8")).hexdigest()[:16]
+    return source_identity(source)
 
 
 def _iso(value: Any) -> str | None:
@@ -369,6 +369,9 @@ async def process_source(body: ProcessBody) -> dict[str, Any]:
         if body.video_id:
             video_id = body.video_id
         elif body.source_url:
+            body.source_url = body.source_url.strip()
+            if not body.source_url:
+                raise HTTPException(status_code=400, detail="source_url must not be blank")
             await validate_source_url(body.source_url, config)
             canonical = canonical_for_source(body.source_url)
             record = await runtime.store.get_video_by_canonical(canonical)
@@ -387,7 +390,7 @@ async def process_source(body: ProcessBody) -> dict[str, Any]:
 
         job = await runtime.queue.enqueue(
             JobKind.PROCESS,
-            {"video_id": video_id, "force": body.force},
+            {"video_id": video_id, **({"force": True} if body.force else {})},
             video_id=video_id,
         )
         return {"job_id": job.id, "video_id": video_id}
diff --git a/vra/identity.py b/vra/identity.py
new file mode 100644
index 0000000..740d543
--- /dev/null
+++ b/vra/identity.py
@@ -0,0 +1,15 @@
+"""Conservative source identity shared by feed ingestion and manual requests.
+
+Legacy lossy identities are never adopted automatically: their saved artifacts
+may belong to an earlier colliding URL even when source_url now looks correct.
+"""
+from __future__ import annotations
+
+import hashlib
+
+
+def source_identity(source: str) -> str:
+    value = source.strip()
+    if not value:
+        raise ValueError('source must not be empty')
+    return 'source:v2:' + hashlib.sha256(value.encode('utf-8')).hexdigest()
diff --git a/vra/runtime.py b/vra/runtime.py
index 97a6be5..86e41c1 100644
--- a/vra/runtime.py
+++ b/vra/runtime.py
@@ -15,7 +15,12 @@
   stages (yt-dlp download, ffmpeg decode) run concurrently across workers.
 - Inference concurrency is managed by the serving engine (continuous
   batching); the app does not gate it further.
+- Local enqueue wakes idle workers by kind; the one-second fallback still
+  observes due retries and writes from other queue instances/processes.
 - On boot, running jobs stale for > 30 minutes are requeued.
+- Normal worker cancellation records a retry (or terminal failure at the
+  existing attempt limit) before SQLite closes. This is not a heartbeat or
+  recovery guarantee for forced termination, repeated cancellation or DB errors.
 """
 
 from __future__ import annotations
@@ -41,6 +46,7 @@
 from vra.store.db import Database
 from vra.store.queue import SqlJobQueue
 from vra.store.repos import SqlVideoStore
+from vra.types import VideoStatus
 
 log = logging.getLogger(__name__)
 
@@ -55,6 +61,23 @@ async def _run_job(job: Job, coro: Any, queue: SqlJobQueue, store: SqlVideoStore
             await queue.fail(job.id, detail or "processing failed", retry=True)
         else:
             await queue.complete(job.id)
+    except asyncio.CancelledError:
+        # Normal shutdown waits for workers before closing SQLite. Persist the
+        # interruption here so a quick restart cannot strand a fresh running
+        # job behind the startup-only 30-minute stale threshold.
+        detail = "job cancelled during execution"
+        if job.kind is JobKind.PROCESS:
+            video_id = str(job.payload.get("video_id", ""))
+            try:
+                rec = await store.get_video(video_id)
+                if rec is not None and rec.status in (
+                    VideoStatus.ACQUIRING, VideoStatus.SUMMARIZING,
+                ):
+                    await store.set_video_status(video_id, VideoStatus.FAILED, error=detail)
+            except Exception:
+                log.exception("could not mark interrupted video %s as failed", video_id)
+        await queue.fail(job.id, detail, retry=True)
+        raise
     except Exception as exc:
         log.warning("job %s failed: %s", job.id, exc)
         await queue.fail(job.id, str(exc), retry=True)
@@ -76,9 +99,11 @@ async def _ingest_worker(
 ) -> None:
     while not stop.is_set():
         try:
+            wakeup = queue.work_wakeup(JobKind.INGEST)
             job = await queue.claim((JobKind.INGEST,))
             if job is None:
-                await asyncio.sleep(_POLL_INTERVAL_S)
+                with contextlib.suppress(TimeoutError):
+                    await asyncio.wait_for(wakeup(), _POLL_INTERVAL_S)
                 continue
             await _run_job(
                 job,
@@ -101,9 +126,11 @@ async def _process_worker(
 ) -> None:
     while not stop.is_set():
         try:
+            wakeup = queue.work_wakeup(JobKind.PROCESS)
             job = await queue.claim((JobKind.PROCESS,))
             if job is None:
-                await asyncio.sleep(_POLL_INTERVAL_S)
+                with contextlib.suppress(TimeoutError):
+                    await asyncio.wait_for(wakeup(), _POLL_INTERVAL_S)
                 continue
             await _run_job(
                 job,
@@ -131,6 +158,8 @@ class AppRuntime:
     scheduler: PollScheduler
     summarizer: VlmSummarizer
     engine: EngineClient
+    _feed_http: httpx.AsyncClient | None = field(default=None, repr=False, compare=False)
+    _transcriber: OpenAiAsrTranscriber | None = field(default=None, repr=False, compare=False)
     _state: dict[str, Any] = field(default_factory=dict, repr=False, compare=False)
 
     async def start(self) -> None:
@@ -172,9 +201,16 @@ async def close(self) -> None:
         self._state["tasks"] = []
         self._state.pop("stop", None)
         self._state["started"] = False
-        await self.scheduler.stop()
-        await self.summarizer.close()
-        await self.queue.close()
+        # These clients are created by build_runtime, not owned by the
+        # summarizer. Unwind every owned resource even if another close fails.
+        async with contextlib.AsyncExitStack() as cleanup:
+            cleanup.push_async_callback(self.queue.close)
+            if self._feed_http is not None:
+                cleanup.push_async_callback(self._feed_http.aclose)
+            if self._transcriber is not None:
+                cleanup.push_async_callback(self._transcriber.close)
+            cleanup.push_async_callback(self.summarizer.close)
+            await self.scheduler.stop()
 
     def stats(self) -> dict[str, Any]:
         return {
@@ -225,4 +261,6 @@ async def build_runtime(config: Config | None = None) -> AppRuntime:
         scheduler=scheduler,
         summarizer=summarizer,
         engine=engine,
+        _feed_http=http,
+        _transcriber=transcriber,
     )
diff --git a/vra/service/ingester.py b/vra/service/ingester.py
index 56d5ad5..ff8a22f 100644
--- a/vra/service/ingester.py
+++ b/vra/service/ingester.py
@@ -17,7 +17,6 @@
 from __future__ import annotations
 
 import asyncio
-import hashlib
 import ipaddress
 import logging
 import socket
@@ -31,6 +30,7 @@
 
 from vra.config import Config
 from vra.contracts import FeedFetcher, JobQueue, VideoStore
+from vra.identity import source_identity
 from vra.types import IngestResult, JobKind, VideoStatus
 
 log = logging.getLogger(__name__)
@@ -219,20 +219,9 @@ async def _validate_url(self, url: str) -> None:
 # canonical identity + FeedIngester
 # --------------------------------------------------------------------------
 
-def _normalize_url(url: str) -> str:
-    """scheme://host/path form with query/fragment dropped (host lowered)."""
-    parts = urlparse(url)
-    if parts.scheme and parts.netloc:
-        return f"{parts.scheme}://{parts.netloc.lower()}{parts.path}"
-    return parts.path or url
-
-
 def _canonical_id(guid: str | None, source_url: str) -> str:
-    """Absolute-http guid wins; otherwise sha1 of the normalized URL."""
-    if guid and guid.startswith("http"):
-        return guid
-    digest = hashlib.sha1(_normalize_url(source_url).encode("utf-8")).hexdigest()
-    return f"url:{digest[:16]}"
+    """GUID remains metadata; acquisition source determines new identity."""
+    return source_identity(source_url)
 
 
 class FeedIngester:
diff --git a/vra/store/queue.py b/vra/store/queue.py
index 0a2a8fa..d2a2028 100644
--- a/vra/store/queue.py
+++ b/vra/store/queue.py
@@ -18,13 +18,17 @@
   claimable.
 
 Timestamps compare lexicographically — all writers use
-``datetime.now(UTC).isoformat()`` (fixed-width UTC ISO-8601).
+UTC ISO-8601 with an explicit six-digit microsecond field. Legacy stored
+non-UTC deadlines need normalization before replay; serialization only
+normalizes new writes.
 """
 
 from __future__ import annotations
 
+import asyncio
 import json
 import uuid
+from collections.abc import Awaitable, Callable
 from datetime import UTC, datetime, timedelta
 from typing import Any
 
@@ -41,18 +45,20 @@ def _utc_now() -> datetime:
 
 
 def _now_iso() -> str:
-    return datetime.now(UTC).isoformat()
+    return datetime.now(UTC).isoformat(timespec="microseconds")
 
 
 def _iso(value: Any) -> str | None:
-    """Normalize a datetime (naive treated as UTC) to an ISO-8601 string."""
+    """Canonical UTC deadline; naive datetimes/ISO strings are treated as UTC."""
     if value is None:
         return None
-    if isinstance(value, datetime):
-        if value.tzinfo is None:
-            value = value.replace(tzinfo=UTC)
-        return value.isoformat()
-    return str(value)
+    if isinstance(value, str):
+        value = datetime.fromisoformat(value)
+    if not isinstance(value, datetime):
+        raise TypeError("run_at must be a datetime, ISO-8601 string, or None")
+    if value.tzinfo is None:
+        value = value.replace(tzinfo=UTC)
+    return value.astimezone(UTC).isoformat(timespec="microseconds")
 
 
 def _parse_dt(value: Any) -> datetime | None:
@@ -99,6 +105,21 @@ def __init__(self, db: Database, poll_interval_s: float = 1.0) -> None:
         self._db = db
         #: Hint for worker loops polling claim() (runtime-owned consumers).
         self.poll_interval_s = poll_interval_s
+        self._work_signals = {kind: asyncio.Event() for kind in JobKind}
+
+    def work_wakeup(self, kind: JobKind) -> Callable[[], Awaitable[bool]]:
+        """Snapshot a local generation before claim; wait only after an empty claim.
+
+        Enqueue rotates rather than clears the event, so concurrent workers
+        cannot consume another worker's notification. SQLite remains the
+        authority; writes through other instances still need polling fallback.
+        """
+        return self._work_signals[kind].wait
+
+    def _notify_work(self, kind: JobKind) -> None:
+        previous = self._work_signals[kind]
+        self._work_signals[kind] = asyncio.Event()
+        previous.set()
 
     async def enqueue(
         self,
@@ -115,6 +136,7 @@ async def enqueue(
         otherwise inserts a fresh queued job. ``dedup_key`` is accepted for
         caller readability and not persisted.
         """
+        normalized_run_at = _iso(run_at)
         del dedup_key  # advisory only; identity is kind + canonical payload
         payload_json = json.dumps(payload, sort_keys=True, separators=(",", ":"))
         now = _now_iso()
@@ -128,26 +150,32 @@ async def enqueue(
                 """,
                 (kind.value, payload_json),
             )
-            if existing is not None:
-                return _job_from_row(existing)
-            job_id = str(uuid.uuid4())
-            await self._db.execute(
-                """
-                INSERT INTO jobs (
-                    id, kind, state, payload, attempts, max_attempts,
-                    run_at, video_id, created_at, updated_at
+            row = existing
+            if row is None:
+                job_id = str(uuid.uuid4())
+                await self._db.execute(
+                    """
+                    INSERT INTO jobs (
+                        id, kind, state, payload, attempts, max_attempts,
+                        run_at, video_id, created_at, updated_at
+                    )
+                    VALUES (?, ?, 'queued', ?, 0, 3, ?, ?, ?, ?)
+                    """,
+                    (job_id, kind.value, payload_json, normalized_run_at, video_id, now, now),
+                    commit=False,
+                )
+                row = await self._db.fetchone(
+                    "SELECT * FROM jobs WHERE id = ?", (job_id,)
                 )
-                VALUES (?, ?, 'queued', ?, 0, 3, ?, ?, ?, ?)
-                """,
-                (job_id, kind.value, payload_json, _iso(run_at), video_id, now, now),
-                commit=False,
-            )
-            row = await self._db.fetchone(
-                "SELECT * FROM jobs WHERE id = ?", (job_id,)
-            )
         if row is None:  # pragma: no cover - defensive
             raise RuntimeError(f"job insert failed: {job_id}")
-        return _job_from_row(row)
+        job = _job_from_row(row)
+        # Match claim's stored deadline ordering. Future work uses the existing
+        # timed fallback instead of waking every idle worker before it is due.
+        due = row["run_at"] is None or str(row["run_at"]) <= _now_iso()
+        if job.state is JobState.QUEUED and due:
+            self._notify_work(kind)
+        return job
 
     async def claim(self, kinds: tuple[JobKind, ...]) -> Job | None:
         """Atomically claim the oldest due queued job of any given kind."""
@@ -219,8 +247,8 @@ async def fail(self, job_id: str, error: str, *, retry: bool) -> None:
     async def requeue_stale(self, older_than_s: float) -> int:
         """Requeue running jobs with stale updated_at; returns the count."""
         now_dt = _utc_now()
-        cutoff = (now_dt - timedelta(seconds=older_than_s)).isoformat()
-        now = now_dt.isoformat()
+        cutoff = _iso(now_dt - timedelta(seconds=older_than_s))
+        now = _iso(now_dt)
         async with self._db.lock:
             rows = await self._db.execute(
                 """
@@ -231,6 +259,9 @@ async def requeue_stale(self, older_than_s: float) -> int:
                 """,
                 (now, now, cutoff),
             )
+        if rows:
+            for kind in JobKind:
+                self._notify_work(kind)
         return len(rows)
 
     async def list(