From a04588740bea9378fe035b02a4aecf67434463c2 Mon Sep 17 00:00:00 2001
From: agent_coder <agent_coder@vvzvlad.xyz>
Date: Sun, 5 Jul 2026 10:04:47 +0300
Subject: [PATCH 1/6] perf(stability): batch SQLite access-time writes out of
 the /media hot path (#5)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Stage 5. A /media cache hit no longer touches SQLite: it records the access
timestamp into a module-level accumulator (a dict write on the event loop,
cheap and atomic), and a supervised 60s background task flushes the whole
batch in one executemany UPDATE. This removes both per-hit write sites — the
awaited to_thread in download_media_file and the fire-and-forget create_task
in the pre-semaphore fast path — so under active RSS polling the threadpool is
no longer starved by per-request access-time UPDATEs.

- file_io: add update_media_file_access_bulk_sync (one connection, executemany;
  empty batch is a no-op).
- api_server: _access_updates accumulator + _flush_access_updates (snapshot-
  then-clear atomically before the await so writes during the flush land in the
  fresh dict; re-queue the batch with setdefault on write failure so a fresher
  concurrent write is never clobbered and no access-time is lost) +
  _access_flush_loop under _supervised + a final flush on shutdown, ordered
  after the loop task is cancelled and before the io threadpool is shut down.
- Keys use str(channel) to match the TEXT channel column (a str/int mix would
  make the UPDATE WHERE silently never match, evicting still-used files).
- tests/test_stage5_sqlite.py: 7 tests (no-sync-write hot path, str-key
  discipline, hit->flush->DB, empty no-op, snapshot-then-clear race, re-queue-
  without-clobbering-fresh, bulk SQL).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 api_server.py               |  94 +++++++++++----
 file_io.py                  |  21 ++++
 tests/test_stage5_sqlite.py | 229 ++++++++++++++++++++++++++++++++++++
 3 files changed, 323 insertions(+), 21 deletions(-)
 create mode 100644 tests/test_stage5_sqlite.py

diff --git a/api_server.py b/api_server.py
index 6201bcf..e98b0a8 100644
--- a/api_server.py
+++ b/api_server.py
@@ -40,7 +40,8 @@ from rss_generator import generate_channel_rss, generate_channel_html
 from post_parser import PostParser
 from url_signer import verify_media_digest, generate_media_digest
 from file_io import (DB_PATH, init_db_sync, get_all_media_file_ids_sync,
-                     update_media_file_access_sync, remove_media_file_ids_sync,
+                     update_media_file_access_sync, update_media_file_access_bulk_sync,
+                     remove_media_file_ids_sync,
                      get_mime_type_sync, set_mime_type_sync)
 
 # Global python-magic instance for MIME type detection
@@ -139,6 +140,57 @@ async def _supervised(factory, name: str, min_restart_interval: float = 60.0):
             await asyncio.sleep(min_restart_interval - elapsed)
 
 
+# Access-time write accumulator. A /media cache hit used to touch SQLite on the hot path
+# (a threadpool hop + connect + UPDATE per request), which starves the threadpool under
+# active RSS polling. Instead a cache hit just records the timestamp here — a dict write
+# on the single-threaded event loop is cheap and atomic — and a periodic background task
+# flushes the whole batch to SQLite in one executemany. Keys use str(channel) to match the
+# TEXT channel column; mixing str/int would make the UPDATE's WHERE silently never match,
+# so the access-time would stop advancing and the file would fall out of the 20-day cache.
+ACCESS_FLUSH_INTERVAL = 60  # seconds between access-time flushes
+_access_updates: dict[tuple[str, int, str], float] = {}
+
+
+async def _flush_access_updates() -> None:
+    """Flush the accumulated access timestamps to SQLite in one bulk UPDATE.
+
+    Snapshot-then-clear atomically on the loop: capture the current dict reference and
+    replace the module global with a fresh empty dict in ONE synchronous step (before any
+    await), so cache-hit writes arriving DURING the flush land in the new dict and are not
+    lost. The bulk UPDATE runs off-loop via asyncio.to_thread. An empty batch is a no-op.
+    """
+    global _access_updates
+    if not _access_updates:
+        return
+    pending = _access_updates
+    _access_updates = {}
+    entries = [(channel, post_id, file_unique_id, added)
+               for (channel, post_id, file_unique_id), added in pending.items()]
+    try:
+        await asyncio.to_thread(update_media_file_access_bulk_sync, DB_PATH, entries)
+    except Exception:
+        # Bulk write failed: re-queue this batch so the access-times are not lost (a lost
+        # timestamp would eventually evict a still-used file from the 20-day cache). Use
+        # setdefault so any FRESHER write accumulated during the flush is never overwritten
+        # by our stale snapshot. Runs on the loop with no await before the mutation, so this
+        # is race-free. Re-raise so the flush loop logs it.
+        for key, added in pending.items():
+            _access_updates.setdefault(key, added)
+        raise
+
+
+async def _access_flush_loop() -> None:
+    """Periodically flush the access-time accumulator (runs under _supervised)."""
+    while True:
+        await asyncio.sleep(ACCESS_FLUSH_INTERVAL)
+        try:
+            await _flush_access_updates()
+        except Exception as e:
+            # Log and keep looping: a transient SQLite error must not drop the batch's
+            # successors. (_supervised still restarts us if this ever raises out.)
+            logger.error(f"access_flush_error: {e}")
+
+
 @asynccontextmanager
 async def lifespan(_: FastAPI):
     setup_logging(Config["log_level"])
@@ -161,9 +213,11 @@ async def lifespan(_: FastAPI):
     # CRITICAL and restarted, so a crash can no longer silently stop cache sweeping or downloads.
     background_task = asyncio.create_task(_supervised(cache_media_files, "cache_media_files"))
     worker_task = asyncio.create_task(_supervised(background_download_worker, "background_download_worker"))
+    access_flush_task = asyncio.create_task(_supervised(_access_flush_loop, "access_flush_loop"))
     yield
     background_task.cancel() # Cleanup
     worker_task.cancel()
+    access_flush_task.cancel()
     try:
         await background_task
     except asyncio.CancelledError:
@@ -172,6 +226,17 @@ async def lifespan(_: FastAPI):
         await worker_task
     except asyncio.CancelledError:
         pass
+    try:
+        await access_flush_task
+    except asyncio.CancelledError:
+        pass
+    # Final flush AFTER the loop task is cancelled (no race with a loop-driven flush) and
+    # BEFORE the threadpool is shut down (to_thread still has its executor), so the last
+    # <=ACCESS_FLUSH_INTERVAL seconds of access-times are persisted on shutdown.
+    try:
+        await _flush_access_updates()
+    except Exception as e:
+        logger.error(f"access_flush_shutdown_error: {e}")
     await client.stop()
     # Shut the io threadpool down so its threads don't linger past a reload/restart.
     io_executor.shutdown(wait=False)
@@ -538,16 +603,11 @@ async def download_media_file(channel: Union[str, int], post_id: int, file_uniqu
                 logger.error(f"cleanup_error: Failed to remove zero-size cached file {cache_path}: {e}")
             # Do not raise error here, proceed to download below
         else:
-            # File exists and is not zero size, update access timestamp and return
+            # File exists and is not zero size, record access timestamp and return.
+            # Record into the accumulator instead of touching SQLite on the hot path; the
+            # background flush persists it. Key channel as str(channel) — see _access_updates.
             logger.info(f"Found cached media file: {cache_path}")
-            try:
-                await asyncio.to_thread(
-                    update_media_file_access_sync,
-                    DB_PATH, str(channel), post_id, file_unique_id,
-                    datetime.now().timestamp()
-                )
-            except Exception as e:
-                logger.error(f"Failed to update timestamp for {channel}/{post_id}/{file_unique_id}: {str(e)}")
+            _access_updates[(str(channel), post_id, file_unique_id)] = datetime.now().timestamp()
             return cache_path, False
 
     file_id = await find_file_id_in_message(message, file_unique_id)
@@ -1074,17 +1134,9 @@ async def get_media(channel: str, post_id: int, file_unique_id: str, request: Re
             if os.path.exists(cache_path) and os.path.getsize(cache_path) > 0:
                 # File is already in cache — skip semaphore and serve directly
                 logger.info(f"pre_semaphore_cache_hit: {channel}/{post_id}/{file_unique_id}")
-                # Fire-and-forget timestamp update with error handling to avoid silent failures
-                async def _update_access(_ch, _pid, _fid):
-                    try:
-                        await asyncio.to_thread(
-                            update_media_file_access_sync,
-                            DB_PATH, str(_ch), _pid, _fid,
-                            datetime.now().timestamp()
-                        )
-                    except Exception as _e:
-                        logger.warning(f"Failed to update access time for {_ch}/{_pid}/{_fid}: {_e}")
-                asyncio.create_task(_update_access(channel, post_id, file_unique_id))
+                # Record the access time into the accumulator instead of firing a per-hit
+                # SQLite write. Key channel as str(channel) — see _access_updates.
+                _access_updates[(str(channel), post_id, file_unique_id)] = datetime.now().timestamp()
                 return await prepare_file_response(cache_path, request=request,
                                                    media_key=(str(channel), post_id, file_unique_id))
 
diff --git a/file_io.py b/file_io.py
index f7ed43d..fbdfb89 100644
--- a/file_io.py
+++ b/file_io.py
@@ -96,6 +96,27 @@ def update_media_file_access_sync(db_path: str, channel: str, post_id: int, file
         )
 
 
+def update_media_file_access_bulk_sync(db_path: str, entries: List[tuple]) -> None:
+    """Update the access timestamp for multiple existing media file ID records.
+
+    entries: iterable of (channel, post_id, file_unique_id, added) tuples.
+    Uses executemany (one connection, one commit) so a batch of cache-hit access
+    updates costs a single SQLite transaction instead of one connect+UPDATE per hit.
+    Rows that do not exist are simply not matched by the WHERE clause (no-op), mirroring
+    the single-row update_media_file_access_sync. An empty batch is a no-op.
+    """
+    if not entries:
+        return
+    with _db_connection(db_path) as conn:
+        conn.executemany(
+            "UPDATE media_file_ids SET added = ? WHERE channel = ? AND post_id = ? AND file_unique_id = ?",
+            # Reorder each (channel, post_id, file_unique_id, added) tuple to match the
+            # UPDATE's placeholder order (added first, then the WHERE key columns).
+            [(added, channel, post_id, file_unique_id)
+             for (channel, post_id, file_unique_id, added) in entries],
+        )
+
+
 def get_all_media_file_ids_sync(db_path: str) -> List[dict]:
     """Return all rows from media_file_ids as a list of dicts."""
     with _db_connection(db_path) as conn:
diff --git a/tests/test_stage5_sqlite.py b/tests/test_stage5_sqlite.py
new file mode 100644
index 0000000..8366b21
--- /dev/null
+++ b/tests/test_stage5_sqlite.py
@@ -0,0 +1,229 @@
+# flake8: noqa
+# pylint: disable=protected-access, missing-function-docstring, missing-class-docstring
+# pylint: disable=redefined-outer-name, logging-fstring-interpolation, line-too-long
+# pylance: disable=reportMissingImports, reportMissingModuleSource
+"""
+Stage 5 (SQLite access-time batching) tests.
+
+Covers:
+- 5.1 accumulator: a /media cache hit records into api_server._access_updates and does NOT
+      call update_media_file_access_sync (no synchronous SQLite write on the hot path).
+- 5.1 flush: seeding _access_updates + running the flush once writes the accumulated
+      timestamps to a real temp DB (hit -> flush -> value updated in DB), and clears the dict.
+- 5.1 bulk fn update_media_file_access_bulk_sync: empty no-op, multi-row, and updating an
+      EXISTING row's `added` (WHERE matches on the str channel key).
+- 5.1 snapshot-then-clear: an update arriving DURING the flush lands in the fresh dict and
+      is not lost.
+- gotcha: str(channel) key discipline — an int-ish channel on the hot path keys the
+      accumulator (and thus the UPDATE) by the string form.
+"""
+import os
+import sys
+import sqlite3
+import asyncio
+from types import SimpleNamespace
+
+import pytest
+
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
+sys.modules['config'] = __import__('tests.mock_config', fromlist=['get_settings'])
+
+import api_server
+from file_io import init_db_sync, update_media_file_access_bulk_sync
+
+
+def _fake_plain_message():
+    """A non-poll, non-video message so download_media_file takes the normal cache flow."""
+    return SimpleNamespace(media=None, video=None, empty=False)
+
+
+@pytest.fixture(autouse=True)
+def _clean_accumulator():
+    """Each test starts with an empty accumulator and restores it afterwards."""
+    api_server._access_updates = {}
+    yield
+    api_server._access_updates = {}
+
+
+# --------------------------------------------------------------------------- #
+# 5.1 hot path: cache hit records into the accumulator, no synchronous SQLite.
+# --------------------------------------------------------------------------- #
+@pytest.mark.asyncio
+async def test_cache_hit_records_accumulator_no_sqlite(tmp_path, monkeypatch):
+    monkeypatch.chdir(tmp_path)
+    channel, post_id, fid = "testchan", 5, "fidHIT"
+    cache_dir = tmp_path / "data" / "cache" / channel / str(post_id)
+    cache_dir.mkdir(parents=True)
+    (cache_dir / fid).write_bytes(b"cached-bytes")
+
+    async def fake_get(_cid, _pid):
+        return _fake_plain_message()
+    monkeypatch.setattr(api_server.client, "safe_get_messages", fake_get)
+
+    # Spy: the single-row synchronous updater must NOT be called on the hot path.
+    called = []
+    monkeypatch.setattr(api_server, "update_media_file_access_sync",
+                        lambda *a, **k: called.append(a))
+
+    path, delete_after = await api_server.download_media_file(channel, post_id, fid)
+
+    assert path == str(cache_dir / fid)
+    assert delete_after is False
+    assert called == []  # no synchronous SQLite access-write happened
+    assert (channel, post_id, fid) in api_server._access_updates
+
+
+# --------------------------------------------------------------------------- #
+# gotcha: str(channel) key discipline on the hot path (int-ish channel).
+# --------------------------------------------------------------------------- #
+@pytest.mark.asyncio
+async def test_cache_hit_keys_channel_as_str(tmp_path, monkeypatch):
+    monkeypatch.chdir(tmp_path)
+    channel_int, post_id, fid = 12345, 7, "fidINT"
+    cache_dir = tmp_path / "data" / "cache" / str(channel_int) / str(post_id)
+    cache_dir.mkdir(parents=True)
+    (cache_dir / fid).write_bytes(b"x")
+
+    async def fake_get(_cid, _pid):
+        return _fake_plain_message()
+    monkeypatch.setattr(api_server.client, "safe_get_messages", fake_get)
+
+    await api_server.download_media_file(channel_int, post_id, fid)
+
+    assert ("12345", post_id, fid) in api_server._access_updates      # string form
+    assert (channel_int, post_id, fid) not in api_server._access_updates  # never the raw int
+
+
+# --------------------------------------------------------------------------- #
+# 5.1 flush: hit -> flush -> the accumulated timestamp is written to the DB.
+# --------------------------------------------------------------------------- #
+@pytest.mark.asyncio
+async def test_flush_writes_accumulated_timestamps(tmp_path, monkeypatch):
+    db = str(tmp_path / "flush.db")
+    init_db_sync(db)
+    # Seed two existing rows with an old timestamp.
+    conn = sqlite3.connect(db)
+    conn.executemany(
+        "INSERT INTO media_file_ids (channel, post_id, file_unique_id, added) VALUES (?,?,?,?)",
+        [("chA", 1, "fA", 1.0), ("chB", 2, "fB", 2.0)],
+    )
+    conn.commit()
+    conn.close()
+
+    monkeypatch.setattr(api_server, "DB_PATH", db)
+    api_server._access_updates = {
+        ("chA", 1, "fA"): 111.0,
+        ("chB", 2, "fB"): 222.0,
+    }
+
+    await api_server._flush_access_updates()
+
+    # Dict cleared after flush.
+    assert api_server._access_updates == {}
+
+    conn = sqlite3.connect(db)
+    rows = dict(((c, p, f), a) for c, p, f, a in
+                conn.execute("SELECT channel, post_id, file_unique_id, added FROM media_file_ids"))
+    conn.close()
+    assert rows[("chA", 1, "fA")] == 111.0
+    assert rows[("chB", 2, "fB")] == 222.0
+
+
+@pytest.mark.asyncio
+async def test_flush_empty_is_noop(monkeypatch):
+    api_server._access_updates = {}
+    # Must not raise and must not touch the threadpool/DB.
+    monkeypatch.setattr(api_server, "update_media_file_access_bulk_sync",
+                        lambda *a, **k: pytest.fail("bulk sync should not run for an empty batch"))
+    await api_server._flush_access_updates()
+
+
+# --------------------------------------------------------------------------- #
+# 5.1 snapshot-then-clear: an update arriving DURING the flush is not lost.
+# --------------------------------------------------------------------------- #
+@pytest.mark.asyncio
+async def test_snapshot_then_clear_does_not_lose_late_update(tmp_path, monkeypatch):
+    db = str(tmp_path / "race.db")
+    init_db_sync(db)
+    monkeypatch.setattr(api_server, "DB_PATH", db)
+
+    late_key = ("chLate", 9, "fLate")
+
+    def fake_bulk(_db, entries):
+        # Simulates a cache hit landing WHILE the flush's to_thread runs: because the flush
+        # already replaced the module dict with a fresh one, this write goes into the NEW dict.
+        api_server._access_updates[late_key] = 999.0
+
+    monkeypatch.setattr(api_server, "update_media_file_access_bulk_sync", fake_bulk)
+
+    api_server._access_updates = {("chA", 1, "fA"): 111.0}
+    await api_server._flush_access_updates()
+
+    # The snapshot (chA) was flushed and cleared; the late update survives in the new dict.
+    assert api_server._access_updates == {late_key: 999.0}
+
+
+# --------------------------------------------------------------------------- #
+# 5.1 re-queue on failure: a failed bulk write is not lost; setdefault keeps a
+# fresher write that arrived during the flush.
+# --------------------------------------------------------------------------- #
+@pytest.mark.asyncio
+async def test_failed_flush_requeues_batch_without_clobbering_fresh(tmp_path, monkeypatch):
+    db = str(tmp_path / "fail.db")
+    init_db_sync(db)
+    monkeypatch.setattr(api_server, "DB_PATH", db)
+
+    stale_key = ("chStale", 1, "fS")   # only in the failing snapshot
+    fresh_key = ("chFresh", 2, "fF")   # re-written FRESHER during the flush
+
+    def fake_bulk(_db, _entries):
+        # A newer cache hit for fresh_key lands in the fresh dict WHILE the bulk write runs,
+        # then the write fails. The re-queue must restore stale_key but must NOT overwrite
+        # the newer fresh_key value (setdefault, not assignment).
+        api_server._access_updates[fresh_key] = 999.0
+        raise sqlite3.OperationalError("database is locked")
+
+    monkeypatch.setattr(api_server, "update_media_file_access_bulk_sync", fake_bulk)
+
+    api_server._access_updates = {stale_key: 1.0, fresh_key: 2.0}
+    with pytest.raises(sqlite3.OperationalError):
+        await api_server._flush_access_updates()
+
+    # stale_key restored with its snapshot value; fresh_key keeps the NEWER value (not 2.0).
+    assert api_server._access_updates == {stale_key: 1.0, fresh_key: 999.0}
+
+
+# --------------------------------------------------------------------------- #
+# 5.1 bulk fn: empty no-op, multi-row, and existing-row update via str channel key.
+# --------------------------------------------------------------------------- #
+def test_bulk_access_update_real_sql(tmp_path):
+    db = str(tmp_path / "bulk.db")
+    init_db_sync(db)
+
+    # Empty batch is a no-op (no crash).
+    update_media_file_access_bulk_sync(db, [])
+
+    # Seed existing rows.
+    conn = sqlite3.connect(db)
+    conn.executemany(
+        "INSERT INTO media_file_ids (channel, post_id, file_unique_id, added) VALUES (?,?,?,?)",
+        [("chA", 1, "fA", 1.0), ("chB", 2, "fB", 2.0)],
+    )
+    conn.commit()
+    conn.close()
+
+    # Multi-row update of EXISTING rows: WHERE matches on the str channel key.
+    update_media_file_access_bulk_sync(db, [
+        ("chA", 1, "fA", 500.0),
+        ("chB", 2, "fB", 600.0),
+    ])
+
+    conn = sqlite3.connect(db)
+    rows = dict(((c, p, f), a) for c, p, f, a in
+                conn.execute("SELECT channel, post_id, file_unique_id, added FROM media_file_ids"))
+    # A row keyed by an int channel does NOT match the string "chA" WHERE (documents the gotcha).
+    n = conn.execute("SELECT COUNT(*) FROM media_file_ids WHERE channel = 1").fetchone()[0]
+    conn.close()
+    assert rows[("chA", 1, "fA")] == 500.0
+    assert rows[("chB", 2, "fB")] == 600.0
+    assert n == 0

From d0801ef0ffe59f55f55850011c8959019b469194 Mon Sep 17 00:00:00 2001
From: agent_coder <agent_coder@vvzvlad.xyz>
Date: Sun, 5 Jul 2026 10:18:36 +0300
Subject: [PATCH 2/6] =?UTF-8?q?fix(stability):=20stage=205=20review=20roun?=
 =?UTF-8?q?d=201=20=E2=80=94=20DoD=20guard=20on=20get=5Fmedia=20hot=20path?=
 =?UTF-8?q?=20+=20correct=20affinity=20comment?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Review findings (both low, no bugs; accumulator design adversarially confirmed):
- test-coverage: the DoD's hottest changed site — get_media's pre-semaphore
  cache-hit — had no direct zero-SQLite guard (the spy test only exercised
  download_media_file), so a regression re-introducing a per-hit write into
  the get_media branch would pass green. Add a mirror spy test through
  get_media asserting the accumulator is written and update_media_file_access_sync
  is NOT called. Verified site-specific: neutering only the get_media write
  reds the new test while the download_media_file test stays green.
- documentation: the _access_updates comment claimed a str/int key mix "would
  make the WHERE silently never match" — empirically false: channel is a TEXT
  column, so a bound int is affinity-coerced and DOES match. Reword to say we
  key str(channel) to stay consistent with the stored form rather than lean on
  SQLite's implicit coercion (the code was already correct on both sites).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 api_server.py               |  8 +++++---
 tests/test_stage5_sqlite.py | 34 ++++++++++++++++++++++++++++++++++
 2 files changed, 39 insertions(+), 3 deletions(-)

diff --git a/api_server.py b/api_server.py
index e98b0a8..6b497af 100644
--- a/api_server.py
+++ b/api_server.py
@@ -144,9 +144,11 @@ async def _supervised(factory, name: str, min_restart_interval: float = 60.0):
 # (a threadpool hop + connect + UPDATE per request), which starves the threadpool under
 # active RSS polling. Instead a cache hit just records the timestamp here — a dict write
 # on the single-threaded event loop is cheap and atomic — and a periodic background task
-# flushes the whole batch to SQLite in one executemany. Keys use str(channel) to match the
-# TEXT channel column; mixing str/int would make the UPDATE's WHERE silently never match,
-# so the access-time would stop advancing and the file would fall out of the 20-day cache.
+# flushes the whole batch to SQLite in one executemany. Keys use str(channel) to stay
+# consistent with the string form written at insert time and to not lean on SQLite's
+# implicit column-affinity coercion (the channel column is TEXT, so a bound int would be
+# coerced and still match — but we key the accumulator by the same type we store, rather
+# than depend on that).
 ACCESS_FLUSH_INTERVAL = 60  # seconds between access-time flushes
 _access_updates: dict[tuple[str, int, str], float] = {}
 
diff --git a/tests/test_stage5_sqlite.py b/tests/test_stage5_sqlite.py
index 8366b21..bd1692c 100644
--- a/tests/test_stage5_sqlite.py
+++ b/tests/test_stage5_sqlite.py
@@ -73,6 +73,40 @@ async def test_cache_hit_records_accumulator_no_sqlite(tmp_path, monkeypatch):
     assert (channel, post_id, fid) in api_server._access_updates
 
 
+# --------------------------------------------------------------------------- #
+# 5.1 hot path (DoD guard): get_media's pre-semaphore cache-hit — the hottest
+# changed site, the one this PR exists for — records into the accumulator and
+# does NO synchronous SQLite access-write. Mirrors the download_media_file spy
+# so a regression re-introducing a per-hit write into THIS branch goes red.
+# --------------------------------------------------------------------------- #
+@pytest.mark.asyncio
+async def test_get_media_pre_semaphore_cache_hit_no_sqlite(tmp_path, monkeypatch):
+    monkeypatch.chdir(tmp_path)
+    channel, post_id, fid = "gmchan", 11, "fidGM"
+    cache_dir = tmp_path / "data" / "cache" / channel / str(post_id)
+    cache_dir.mkdir(parents=True)
+    (cache_dir / fid).write_bytes(b"cached-bytes")
+
+    # Bypass the HMAC digest gate and the FileResponse machinery — the test targets
+    # only the access-time write on the pre-semaphore cache-hit branch.
+    monkeypatch.setattr(api_server, "verify_media_digest", lambda *a, **k: True)
+    sentinel = object()
+    async def fake_prepare(cache_path, request=None, media_key=None):
+        return sentinel
+    monkeypatch.setattr(api_server, "prepare_file_response", fake_prepare)
+
+    # Spy: the single-row synchronous updater must NOT be called on the hot path.
+    called = []
+    monkeypatch.setattr(api_server, "update_media_file_access_sync",
+                        lambda *a, **k: called.append(a))
+
+    resp = await api_server.get_media(channel, post_id, fid, request=object(), digest="x")
+
+    assert resp is sentinel
+    assert called == []  # no synchronous SQLite access-write on the hottest path
+    assert (channel, post_id, fid) in api_server._access_updates
+
+
 # --------------------------------------------------------------------------- #
 # gotcha: str(channel) key discipline on the hot path (int-ish channel).
 # --------------------------------------------------------------------------- #

From 7d6ee0271d5152c4674bf5cf855cacb53e2d6081 Mon Sep 17 00:00:00 2001
From: agent_coder <agent_coder@vvzvlad.xyz>
Date: Sun, 5 Jul 2026 10:34:06 +0300
Subject: [PATCH 3/6] =?UTF-8?q?feat(stability):=20stage=206=20=E2=80=94=20?=
 =?UTF-8?q?lightweight=20/ping=20healthcheck=20that=20never=20touches=20Te?=
 =?UTF-8?q?legram=20(#6)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The container healthcheck hit /rss/...?limit=1 (5s timeout): on a cold cache RSS
generation exceeds 5s, or a hung TG RPC makes it hang, so docker/autoheal restarts
the container mid-download and corrupts temp files. Replace it with /ping, which
reflects process/loop liveness (answers instantly, always) plus TG liveness read
from the watchdog's last-probe data — issuing ZERO Telegram RPC.

- telegram_client: public watchdog_last_ok_age() — seconds since the last successful
  watchdog probe (None if never). Pure read of the Stage-1 _wd_last_ok_monotonic
  field; no RPC.
- api_server: /ping route (no token, no TG RPC, no SQLite, no fs scan). healthy =
  connected and (age is None or age < threshold). age is None right after boot =>
  healthy (don't kill before the first probe). connected coerced to bool so the JSON
  "connected" field is always a bool (pre-start reports false, never null).
- config: TG_PING_UNHEALTHY_AFTER knob, default interval*(failures+1)+timeout = 250s
  (how long until the watchdog itself gives up), env-overridable.
- dockercompose.yml: healthcheck -> curl -sf http://127.0.0.1:80/ping, interval 5m,
  timeout 5s, retries 3, start_period 30s. Old /rss check removed, not left behind.
- tests: 10 (healthy/stale/disconnected/fresh-boot/pre-start-null/no-token +
  the anti-regression zero-TG-RPC spy across all branches + the accessor).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 api_server.py                    |  34 +++++-
 config.py                        |  11 ++
 dockercompose.yml                |  11 +-
 telegram_client.py               |  11 ++
 tests/mock_config.py             |   1 +
 tests/test_stage6_healthcheck.py | 173 +++++++++++++++++++++++++++++++
 6 files changed, 237 insertions(+), 4 deletions(-)
 create mode 100644 tests/test_stage6_healthcheck.py

diff --git a/api_server.py b/api_server.py
index 6b497af..bba4f43 100644
--- a/api_server.py
+++ b/api_server.py
@@ -33,7 +33,7 @@ from pyrogram import errors
 from pyrogram.types import Message
 from pyrogram.enums import MessageMediaType
 from fastapi import FastAPI, HTTPException, Response, Request
-from fastapi.responses import HTMLResponse, FileResponse
+from fastapi.responses import HTMLResponse, FileResponse, JSONResponse
 from telegram_client import TelegramClient
 from config import get_settings, setup_logging
 from rss_generator import generate_channel_rss, generate_channel_html
@@ -1065,6 +1065,38 @@ async def get_raw_post_json(channel: str, post_id: int, request: Request, token:
         raise HTTPException(status_code=500, detail=error_message) from e
 
 
+@app.get("/ping")
+async def ping() -> JSONResponse:
+    """Lightweight liveness probe for the container healthcheck.
+
+    Reflects process/event-loop liveness (always answers in microseconds) and TG liveness
+    from the watchdog's last-probe data. It MUST NOT issue any Telegram RPC (no get_me,
+    no safe_get_messages), touch SQLite, or walk the filesystem — that is the whole point:
+    it stays instant and truthful even while a real TG RPC is hung. It only reads the
+    already-recorded watchdog timestamp and the is_connected bool.
+    """
+    age = client.watchdog_last_ok_age()          # seconds since last OK probe, None if never
+    # is_connected is None before client.start() and a bool afterwards; coerce so the JSON
+    # "connected" field is always a bool (never null) and the pre-start window reports false.
+    connected = bool(client.client.is_connected)
+    threshold = Config["tg_ping_unhealthy_after"]
+    # age is None right after boot: the watchdog hasn't run its first probe yet. Treat that
+    # as healthy (gate on connected only) so a freshly-started container is not killed before
+    # its first probe cycle — otherwise start_period would have to cover a full watchdog interval.
+    # Note: when the watchdog is DISABLED (TG_WATCHDOG_ENABLED=false) age stays None forever,
+    # so /ping degenerates to a pure connectivity check and cannot detect a stale-but-connected
+    # ("zombie") session — that TG-liveness signal only exists while the watchdog runs.
+    healthy = connected and (age is None or age < threshold)
+    return JSONResponse(
+        {
+            "status": "ok" if healthy else "degraded",
+            "connected": connected,
+            "last_probe_age_s": round(age, 1) if age is not None else None,
+            "threshold_s": threshold,
+        },
+        status_code=200 if healthy else 503,
+    )
+
 @app.get("/health")
 @app.get("/health/{token}")
 async def health_check(request: Request, token: str | None = None) -> Response:
diff --git a/config.py b/config.py
index a8fc395..3cc8763 100644
--- a/config.py
+++ b/config.py
@@ -123,6 +123,17 @@ def get_settings() -> dict[str, Any]:
         "tg_watchdog_heartbeat_every": _parse_int_env("TG_WATCHDOG_HEARTBEAT_EVERY", 30),
         "tg_disconnect_flap_limit": _parse_int_env("TG_DISCONNECT_FLAP_LIMIT", 3),
         "tg_disconnect_flap_window": _parse_int_env("TG_DISCONNECT_FLAP_WINDOW", 120),
+        # /ping reports TG as unhealthy once the last successful watchdog probe is older than
+        # this many seconds. Default is derived from the watchdog cadence: it is roughly how
+        # long the watchdog itself would take to give up and trigger a restart —
+        # interval * (failures + 1) + timeout. With the defaults (60,3,10) that is 250s, so a
+        # transient slow probe never flaps /ping, but a genuinely stuck session (no successful
+        # probe for ~4 min) surfaces as 503 before/around the time the watchdog restarts.
+        "tg_ping_unhealthy_after": _parse_int_env(
+            "TG_PING_UNHEALTHY_AFTER",
+            _parse_int_env("TG_WATCHDOG_INTERVAL", 60) * (_parse_int_env("TG_WATCHDOG_FAILURES", 3) + 1)
+            + _parse_int_env("TG_WATCHDOG_TIMEOUT", 10),
+        ),
         # Media download timeout scales with file size (large videos): the per-download
         # timeout is clamped to [min, max] seconds, with an effective floor of
         # `media_download_min_speed` bytes/s (timeout ≈ file_size / min_speed).
diff --git a/dockercompose.yml b/dockercompose.yml
index c22711a..4fa8bda 100644
--- a/dockercompose.yml
+++ b/dockercompose.yml
@@ -53,10 +53,15 @@ services:
       com.centurylinklabs.watchtower.enable: "true"   
       autoheal: true
     healthcheck:
-      test: ["CMD", "curl", "-s", "-f", "-o", "/dev/null", "http://127.0.0.1:80/rss/vvzvlad_lytdybr?limit=1"]
-      interval: 30m
+      # Lightweight process/loop liveness probe. /ping never touches Telegram or the
+      # filesystem, so it answers instantly even while a TG RPC is hung — unlike the old
+      # /rss?limit=1 check, which could exceed the 5s timeout on a cold cache and get the
+      # container restarted mid-download (corrupted temp files). TG liveness is now judged
+      # by the in-process watchdog, which /ping reports via its last-probe age.
+      test: ["CMD", "curl", "-sf", "http://127.0.0.1:80/ping"]
+      interval: 5m
       timeout: 5s
-      retries: 2
+      retries: 3
       start_period: 30s
       start_interval: 5s
 
diff --git a/telegram_client.py b/telegram_client.py
index da9793e..6d4f547 100644
--- a/telegram_client.py
+++ b/telegram_client.py
@@ -259,6 +259,17 @@ class TelegramClient:
             # Emergency termination
             os._exit(1)
 
+    def watchdog_last_ok_age(self) -> float | None:
+        """Seconds since the last successful watchdog probe, or None if none succeeded yet.
+
+        Reads only the already-recorded monotonic timestamp set by the watchdog loop; it
+        never issues a Telegram RPC, so it is safe to call from the hot /ping path even
+        while a real RPC is hung.
+        """
+        if self._wd_last_ok_monotonic is None:
+            return None
+        return time.monotonic() - self._wd_last_ok_monotonic
+
     async def safe_get_messages(self, channel_id, post_id, max_retries=2):
         """Wrapper with retry logic for auth errors"""
         for attempt in range(max_retries):
diff --git a/tests/mock_config.py b/tests/mock_config.py
index d6bacfe..cfed846 100644
--- a/tests/mock_config.py
+++ b/tests/mock_config.py
@@ -33,6 +33,7 @@ def get_settings():
         "tg_watchdog_heartbeat_every": 30,
         "tg_disconnect_flap_limit": 3,
         "tg_disconnect_flap_window": 120,
+        "tg_ping_unhealthy_after": 250,
         "media_download_timeout_min": 120,
         "media_download_timeout_max": 1800,
         "media_download_min_speed": 256 * 1024,
diff --git a/tests/test_stage6_healthcheck.py b/tests/test_stage6_healthcheck.py
new file mode 100644
index 0000000..8f54b63
--- /dev/null
+++ b/tests/test_stage6_healthcheck.py
@@ -0,0 +1,173 @@
+# flake8: noqa
+# pylint: disable=protected-access, missing-function-docstring, missing-class-docstring
+# pylint: disable=redefined-outer-name, logging-fstring-interpolation, line-too-long
+# pylance: disable=reportMissingImports, reportMissingModuleSource
+"""
+Stage 6 (lightweight /ping healthcheck) regression tests.
+
+Covers:
+- /ping returns 200 "ok" when connected and the last probe is recent (age < threshold).
+- /ping returns 503 "degraded" when connected but the last probe is stale (age > threshold).
+- /ping returns 503 "degraded" when disconnected, regardless of probe age.
+- /ping returns 200 "ok" on a fresh boot (age is None) while connected — a freshly-started
+  container must NOT be killed before the watchdog's first probe.
+- ANTI-REGRESSION (the critical invariant): /ping issues ZERO Telegram RPC. A spy on the
+  fake client's get_me / safe_get_messages proves neither is ever called.
+- TelegramClient.watchdog_last_ok_age(): None when never probed; a positive float afterwards.
+"""
+import os
+import sys
+import time
+
+import pytest
+
+# Add project root to sys.path and mock the config module (same pattern as the other tests).
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
+sys.modules['config'] = __import__('tests.mock_config', fromlist=['get_settings'])
+
+from fastapi.testclient import TestClient
+
+import api_server
+from telegram_client import TelegramClient
+
+
+# --------------------------------------------------------------------------- #
+# Fakes
+# --------------------------------------------------------------------------- #
+class _FakeKurigramClient:
+    """Stands in for TelegramClient.client — exposes is_connected and RPC spies."""
+    def __init__(self, is_connected=True):
+        self.is_connected = is_connected
+        self.get_me_calls = 0
+
+    async def get_me(self):
+        # If /ping ever touches this, the whole point of the endpoint is defeated.
+        self.get_me_calls += 1
+        raise AssertionError("/ping must never call get_me()")
+
+
+class _FakeTelegramClient:
+    """Stands in for api_server.client with a controllable probe age + RPC spies."""
+    def __init__(self, age, is_connected=True):
+        self._age = age
+        self.client = _FakeKurigramClient(is_connected=is_connected)
+        self.safe_get_messages_calls = 0
+
+    def watchdog_last_ok_age(self):
+        return self._age
+
+    async def safe_get_messages(self, *args, **kwargs):
+        self.safe_get_messages_calls += 1
+        raise AssertionError("/ping must never call safe_get_messages()")
+
+
+@pytest.fixture
+def patch_client(monkeypatch):
+    """Return a factory that installs a fake api_server.client and yields a TestClient."""
+    def _install(age, is_connected=True):
+        fake = _FakeTelegramClient(age=age, is_connected=is_connected)
+        monkeypatch.setattr(api_server, "client", fake)
+        return fake, TestClient(api_server.app)
+    return _install
+
+
+THRESHOLD = api_server.Config["tg_ping_unhealthy_after"]  # 250 in mock_config
+
+
+# --------------------------------------------------------------------------- #
+# /ping endpoint behavior
+# --------------------------------------------------------------------------- #
+def test_ping_healthy_connected_recent(patch_client):
+    fake, tc = patch_client(age=THRESHOLD - 10, is_connected=True)
+    r = tc.get("/ping")
+    assert r.status_code == 200
+    body = r.json()
+    assert body["status"] == "ok"
+    assert body["connected"] is True
+    assert body["last_probe_age_s"] == round(THRESHOLD - 10, 1)
+    assert body["threshold_s"] == THRESHOLD
+    assert fake.client.get_me_calls == 0
+
+
+def test_ping_degraded_stale_probe(patch_client):
+    fake, tc = patch_client(age=THRESHOLD + 100, is_connected=True)
+    r = tc.get("/ping")
+    assert r.status_code == 503
+    body = r.json()
+    assert body["status"] == "degraded"
+    assert body["connected"] is True
+    assert fake.client.get_me_calls == 0
+
+
+def test_ping_degraded_disconnected(patch_client):
+    # Even with a fresh probe age, a disconnected client is unhealthy.
+    fake, tc = patch_client(age=1.0, is_connected=False)
+    r = tc.get("/ping")
+    assert r.status_code == 503
+    body = r.json()
+    assert body["status"] == "degraded"
+    assert body["connected"] is False
+
+
+def test_ping_degraded_disconnected_even_when_age_none(patch_client):
+    fake, tc = patch_client(age=None, is_connected=False)
+    r = tc.get("/ping")
+    assert r.status_code == 503
+    assert r.json()["status"] == "degraded"
+
+
+def test_ping_fresh_boot_age_none_is_healthy(patch_client):
+    # Right after boot the watchdog hasn't probed yet (age None); connected => healthy.
+    fake, tc = patch_client(age=None, is_connected=True)
+    r = tc.get("/ping")
+    assert r.status_code == 200
+    body = r.json()
+    assert body["status"] == "ok"
+    assert body["last_probe_age_s"] is None
+    assert body["threshold_s"] == THRESHOLD
+
+
+def test_ping_pre_start_connected_none_is_degraded_bool(patch_client):
+    # Before client.start(), Kurigram's is_connected is None. /ping must not 500: it coerces
+    # to a bool, so "connected" is false (never null) and the endpoint reports 503 degraded.
+    fake, tc = patch_client(age=None, is_connected=None)
+    r = tc.get("/ping")
+    assert r.status_code == 503
+    body = r.json()
+    assert body["status"] == "degraded"
+    assert body["connected"] is False  # bool, not null
+    assert fake.client.get_me_calls == 0
+
+
+def test_ping_issues_no_tg_rpc(patch_client):
+    """The critical invariant: /ping never issues any Telegram RPC in any branch."""
+    for age, connected in [(1.0, True), (THRESHOLD + 500, True), (None, True), (1.0, False)]:
+        fake, tc = patch_client(age=age, is_connected=connected)
+        tc.get("/ping")
+        assert fake.client.get_me_calls == 0, f"get_me called (age={age}, connected={connected})"
+        assert fake.safe_get_messages_calls == 0, f"safe_get_messages called (age={age}, connected={connected})"
+
+
+def test_ping_route_needs_no_token(patch_client):
+    # /ping is unauthenticated by design (no token path variant); it just works.
+    fake, tc = patch_client(age=1.0, is_connected=True)
+    assert tc.get("/ping").status_code == 200
+
+
+# --------------------------------------------------------------------------- #
+# TelegramClient.watchdog_last_ok_age accessor
+# --------------------------------------------------------------------------- #
+def test_watchdog_last_ok_age_none_when_never_probed():
+    tgc = TelegramClient()
+    assert tgc._wd_last_ok_monotonic is None
+    assert tgc.watchdog_last_ok_age() is None
+
+
+def test_watchdog_last_ok_age_positive_after_probe():
+    tgc = TelegramClient()
+    tgc._wd_last_ok_monotonic = time.monotonic() - 5
+    age = tgc.watchdog_last_ok_age()
+    assert age is not None
+    assert age >= 5.0
+    # Sanity: a plausible upper bound so we didn't accidentally read the wrong field.
+    assert age < 60.0

From 1cc592bef6f6fa01ec4aa809f5dd3b0fcbb32e10 Mon Sep 17 00:00:00 2001
From: agent_coder <agent_coder@vvzvlad.xyz>
Date: Sun, 5 Jul 2026 10:47:29 +0300
Subject: [PATCH 4/6] =?UTF-8?q?fix(stability):=20stage=206=20review=20roun?=
 =?UTF-8?q?d=201=20=E2=80=94=20/ping=20degrades=20to=20pure=20connectivity?=
 =?UTF-8?q?=20when=20watchdog=20disabled?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Review finding (low, real): the comment claimed /ping degenerates to a pure
connectivity check when TG_WATCHDOG_ENABLED=false, but _wd_last_ok_monotonic is
also stamped by _restart_client (on the disconnect-flap path, which runs before
the watchdog-enabled gate), so with the watchdog off one flap sets age and nothing
ever refreshes it — age grows unbounded past the threshold and /ping returns 503 on
a live connection, spuriously failing the container healthcheck and triggering an
autoheal restart after every flap.

Fix: gate the staleness branch on the watchdog being enabled —
  healthy = connected and (not Config["tg_watchdog_enabled"] or age is None or age < threshold)
so with the watchdog disabled /ping is a pure connectivity check (matching the
intent), and correct the comment to note a flap-restart can stamp age even when the
watchdog is off. New test test_ping_watchdog_disabled_stale_age_still_healthy:
watchdog off + connected + stale age => 200 ok. Adversarially validated — reverting
the gate reds the new test (503) while the watchdog-ON stale-probe test stays green.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 api_server.py                    | 16 ++++++++++++----
 tests/test_stage6_healthcheck.py | 15 +++++++++++++++
 2 files changed, 27 insertions(+), 4 deletions(-)

diff --git a/api_server.py b/api_server.py
index bba4f43..6fbe2ea 100644
--- a/api_server.py
+++ b/api_server.py
@@ -1083,10 +1083,18 @@ async def ping() -> JSONResponse:
     # age is None right after boot: the watchdog hasn't run its first probe yet. Treat that
     # as healthy (gate on connected only) so a freshly-started container is not killed before
     # its first probe cycle — otherwise start_period would have to cover a full watchdog interval.
-    # Note: when the watchdog is DISABLED (TG_WATCHDOG_ENABLED=false) age stays None forever,
-    # so /ping degenerates to a pure connectivity check and cannot detect a stale-but-connected
-    # ("zombie") session — that TG-liveness signal only exists while the watchdog runs.
-    healthy = connected and (age is None or age < threshold)
+    #
+    # The staleness branch (age >= threshold => degraded) is only meaningful while the watchdog
+    # is running to refresh age. With the watchdog DISABLED (TG_WATCHDOG_ENABLED=false) nothing
+    # refreshes age — yet a disconnect-flap restart can still stamp it once (see _restart_client,
+    # which runs before the watchdog-enabled gate), after which age only grows. Letting that
+    # stale age drive /ping to 503 would spuriously fail the container healthcheck on a live
+    # connection and trigger an autoheal restart. So gate staleness on the watchdog being on;
+    # with it off, /ping is a pure connectivity check (no zombie-session detection — that
+    # TG-liveness signal only exists while the watchdog runs).
+    healthy = connected and (
+        not Config["tg_watchdog_enabled"] or age is None or age < threshold
+    )
     return JSONResponse(
         {
             "status": "ok" if healthy else "degraded",
diff --git a/tests/test_stage6_healthcheck.py b/tests/test_stage6_healthcheck.py
index 8f54b63..2a1d5ff 100644
--- a/tests/test_stage6_healthcheck.py
+++ b/tests/test_stage6_healthcheck.py
@@ -99,6 +99,21 @@ def test_ping_degraded_stale_probe(patch_client):
     assert fake.client.get_me_calls == 0
 
 
+def test_ping_watchdog_disabled_stale_age_still_healthy(patch_client, monkeypatch):
+    # With the watchdog OFF, nothing refreshes age (a disconnect-flap restart can stamp it
+    # once, then it only grows). A stale age must NOT drive /ping to 503 on a live connection
+    # — that would spuriously fail the healthcheck and trigger an autoheal restart. So with the
+    # watchdog disabled, /ping is a pure connectivity check: connected + stale age => healthy.
+    monkeypatch.setitem(api_server.Config, "tg_watchdog_enabled", False)
+    fake, tc = patch_client(age=THRESHOLD + 100, is_connected=True)
+    r = tc.get("/ping")
+    assert r.status_code == 200
+    body = r.json()
+    assert body["status"] == "ok"
+    assert body["connected"] is True
+    assert fake.client.get_me_calls == 0
+
+
 def test_ping_degraded_disconnected(patch_client):
     # Even with a fresh probe age, a disconnected client is unhealthy.
     fake, tc = patch_client(age=1.0, is_connected=False)

From cff6e61d2f96222623240bce630f5d6af3a9c8bc Mon Sep 17 00:00:00 2001
From: agent_coder <agent_coder@vvzvlad.xyz>
Date: Sun, 5 Jul 2026 11:06:09 +0300
Subject: [PATCH 5/6] =?UTF-8?q?test(stability):=20stage=207=20=E2=80=94=20?=
 =?UTF-8?q?end-to-end=20verification=20(integration=20tests=20+=20report)?=
 =?UTF-8?q?=20(#7)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Caps the stacked stability work (stages 1-6). Verification only — ZERO production
code change (git diff against fix/stage-6-healthcheck touches only these two files).

- tests/test_stage7_integration.py (8 cross-stage integration tests via TestClient +
  a mocked TelegramClient, no network): Range at the /media route level (206/206/416
  through get_media -> prepare_file_response -> FileResponse); /ping stays prompt and
  issues zero TG RPC while a slow op is parked; in-flight dedup shares one download and
  drains _inflight after completion AND after request cancellation (no stuck key / hung
  waiter); str(channel) access-time hit -> flush -> the bulk UPDATE lands on the seeded
  row (mutation-verified: transposing the key columns reds it).
- docs/stability-verification.md: per-stage DoD -> evidence mapping (test or "operator
  observation" for prod-only items), the exact manual curl/lsof scenarios for the
  operator to run post-deploy, the diag-log signals to watch, and the per-stage
  independent-commit rollback plan.

Full integrated suite: 260 passed (252 baseline + 8). The prod deploy + live diag-log
observation (plan items 3-4) are the operator's call — this stage does not deploy.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 docs/stability-verification.md   | 138 ++++++++++++
 tests/test_stage7_integration.py | 349 +++++++++++++++++++++++++++++++
 2 files changed, 487 insertions(+)
 create mode 100644 docs/stability-verification.md
 create mode 100644 tests/test_stage7_integration.py

diff --git a/docs/stability-verification.md b/docs/stability-verification.md
new file mode 100644
index 0000000..79958e4
--- /dev/null
+++ b/docs/stability-verification.md
@@ -0,0 +1,138 @@
+# Стадия 7 — сквозная верификация плана стабилизации
+
+Дата: 2026-07-05. Ветка: `fix/stage-7-verification` (на базе `fix/stage-6-healthcheck`,
+кумулятивно содержит стадии 1–6). Эта стадия **ничего не меняет в поведении** прод-кода —
+только автоматизирует проверки и фиксирует, что должен наблюдать оператор после деплоя.
+
+## Итог автотестов
+
+Каноничный прогон (изоляция `config` через `sys.modules`): `python -m pytest tests/`.
+
+```
+260 passed
+```
+
+Раскладка по стадиям (все зелёные):
+
+| Стадия | Файл тестов | Кол-во | Что покрывает |
+|--------|-------------|--------|----------------|
+| 1 — анти-зависания | `tests/test_stage1_hangs.py` | 6 | таймаут RPC-гейта без утечки пермита; воркер переживает Exception/FloodWait, `task_done` сбалансирован; отмена в spacing-ожидании не теряет пермит |
+| 2 — статика/большие видео | `tests/test_stage2_static.py` | 15 | атомарный `_download_atomic` (publish-on-rename, чистка `.part.` при таймауте/zero-size/гонке); дедуп конкурентных скачиваний; FloodWait→429; touch mtime у `temp_*`; свипер чистит `.part.`+legacy `.tmp.`; баланс HTTP-семафора |
+| 3 — FileResponse | `tests/test_stage3_fileresponse.py` | 19 | матрица Range (0-499/500-/-500/за EOF→416/мусор→400/мульти-range→206 multipart); сохранены mtime-touch, delete_after-BackgroundTask, MIME-кэш; чистый ASGI-логгер |
+| 4 — гигиена event loop | `tests/test_stage4_eventloop.py` | 19 | ленивый `raw_message`; вынос side-effect IO из `process_message` (bulk upsert media id); рендер-пайплайн ушёл в поток без create_task/get_running_loop; XSS вычищен во всех 4 выводах ровно одним проходом |
+| 5 — батчинг SQLite | `tests/test_stage5_sqlite.py` | 8 | кэш-хит пишет в аккумулятор, а не в SQLite; flush→DB; snapshot-then-clear не теряет поздние апдейты; re-queue при сбое; str(channel)-ключи |
+| 6 — healthcheck | `tests/test_stage6_healthcheck.py` | 11 | `/ping` 200/503 по connected+age; ноль TG RPC; watchdog отключён→чистая проверка коннекта; `watchdog_last_ok_age()` |
+| 7 — интеграция | `tests/test_stage7_integration.py` | 8 | сквозные сценарии (см. ниже) |
+| — регрессия парсера | `tests/test_postparser_*.py` | 174 | заголовки/флаги/автор (существовавшие до плана) |
+
+## Новые интеграционные тесты стадии 7 (DoD → сквозное доказательство)
+
+`tests/test_stage7_integration.py` драйвит **реальные точки входа** (`get_media`, `ping`,
+flush), чтобы поймать регрессии, которые проявляются только при взаимодействии стадий:
+
+1. **Range на уровне маршрута `/media`** — `test_media_route_range_prefix_0_99`,
+   `_range_suffix_last_100`, `_range_unsatisfiable_416`. Прогоняют `get_media` (кэш-хит) →
+   `prepare_file_response` → `FileResponse` через реальный ASGI: `bytes=0-99`→206 с точным
+   Content-Range/длиной и байтами, `bytes=-100`→206 (суффикс), `bytes=999999999-`→416 (`*/size`).
+   Ловит: если кэш-хит перестанет доходить до FileResponse (ре-буферизация тела, ручные
+   заголовки) или сломается связка digest-гейта — Range перестанет работать. (Стадия 3
+   пинит это на `prepare_file_response` напрямую; здесь — сшивка стадий 2+3.)
+2. **`/ping` быстр и без RPC при висящей операции** — `test_ping_prompt_and_rpc_free_while_slow_op_pending`
+   и `_reports_degraded_promptly_while_slow_op_pending`. Пока фейковая медленная корутина
+   (модель зависшего hot-path) висит на `Event`, `ping()` возвращает 200/503 корректно и
+   **до** завершения медленной операции, при нуле вызовов `get_me`/`safe_get_messages`.
+   Ловит: рекаплинг `/ping` к TG RPC или к любому awaitable, который может застопориться.
+3. **Дедуп + очистка при disconnect через реальный `get_media`** —
+   `test_get_media_concurrent_shares_one_download_and_drains_registry` (2 конкурентных
+   запроса → одна скачка, `_inflight` пуст) и `_request_cancel_does_not_stick_registry_or_hang_sibling`
+   (отмена запроса-«клиента» не оставляет застрявший ключ и не вешает соседа). Ловит:
+   возврат скачивания в корутину запроса (отмена убила бы download) или потерю `finally`-pop.
+4. **str(channel)-ключ access-time end-to-end** — `test_media_cache_hit_flush_updates_str_keyed_db_row`.
+   Кэш-хит `/media` записывает str-ключ, flush обновляет ту же строку в реальной SQLite
+   (hit→flush→DB). Сшивает две половины: ключ, который пишет hot-path, и WHERE, по которому
+   апдейтит flush. Ловит любое расхождение ключа аккумулятора и WHERE bulk-UPDATE (напр.
+   перестановку колонок в SQL — проверено мутацией: тест краснеет, `added` остаётся stale).
+   (int/str-хазард самого канала живёт в `download_media_file` и закрыт стадией 5 —
+   `test_cache_hit_keys_channel_as_str`; здесь покрыта связка get_media+flush.)
+
+## Соответствие DoD стадий → доказательство
+
+- **DoD 1** «ни один путь не ждёт TG без таймаута; воркер не умирает молча» →
+  `test_stage1_hangs.py` (гейт-таймаут, живучесть воркера). Живая проверка supervision
+  под нагрузкой — **наблюдение оператора** (лог `supervised_task_crashed/…_exited`).
+- **DoD 2** «клиенту никогда не отдаётся неполный файл; флуд→429; обрезки не живут >часа» →
+  `test_stage2_static.py` целиком + интеграционный дедуп-тест стадии 7.
+- **DoD 3** «Range-тесты зелёные; поведение эндпоинта эквивалентно (± RFC-допущения)» →
+  `test_stage3_fileresponse.py` + Range на уровне `/media` (стадия 7). «Нет потока-на-чанк»
+  — **наблюдение оператора** (нагрузочный запрос большого файла, число io-потоков).
+- **DoD 4** «генерация 100-сообщ. фида не блокирует луп (параллельный /ping <100 мс); XSS
+  зелёный; media id сохраняются» → `test_stage4_eventloop.py` (рендер в потоке, XSS,
+  bulk upsert) + `/ping`-decoupling стадии 7. Живой замер «<100 мс на проде» —
+  **наблюдение оператора**.
+- **DoD 5** «на кэш-хит /media ноль обращений к SQLite; фоновая запись раз в минуту» →
+  `test_stage5_sqlite.py` + str-ключ end-to-end стадии 7.
+- **DoD 6** «во время зависшего TG RPC /ping отвечает мгновенно (503); контейнер не
+  рестартится от медленного фида» → `test_stage6_healthcheck.py` + `/ping`-under-slow-op
+  стадии 7. Отсутствие autoheal-рестартов на холодном кэше — **наблюдение оператора**.
+
+## Ручные curl-сценарии для оператора (пост-деплой, локально в контейнере)
+
+Подставить реальный `{channel}/{post_id}/{fid}/{digest}` (валидная подпись обязательна).
+
+```bash
+# 1. Range-тройка (ожидания: 206 / 206 / 416)
+curl -s -D- -o /dev/null -H "Range: bytes=0-99"      "http://127.0.0.1:80/media/{ch}/{pid}/{fid}/{digest}"
+curl -s -D- -o /dev/null -H "Range: bytes=-100"       "http://127.0.0.1:80/media/{ch}/{pid}/{fid}/{digest}"
+curl -s -D- -o /dev/null -H "Range: bytes=999999999-" "http://127.0.0.1:80/media/{ch}/{pid}/{fid}/{digest}"
+
+# 2. Параллельные запросы одного БОЛЬШОГО видео (>100 MB), пока не в кэше:
+#    оба должны получить ПОЛНЫЙ файл (одинаковый размер), без частичной отдачи.
+URL="http://127.0.0.1:80/media/{ch}/{pid}/{bigfid}/{digest}"
+curl -s -o /tmp/a "$URL" & curl -s -o /tmp/b "$URL" & wait
+ls -l /tmp/a /tmp/b     # размеры совпадают и равны полному файлу; на диске нет *.part.*
+
+# 3. Обрыв на середине стрима — нет утечки тасков/фд:
+#    считать fd до/после и убедиться, что не растут монотонно.
+lsof -p $(pgrep -f api_server) | wc -l          # baseline
+timeout 2 curl -s -o /dev/null "$URL"; sleep 5  # оборвать скачку на середине (несколько раз)
+lsof -p $(pgrep -f api_server) | wc -l          # не выросло относительно baseline
+
+# 4. Генерация большого фида + параллельный /ping (<100 мс во время генерации):
+curl -s -o /dev/null "http://127.0.0.1:80/rss/{big_channel}" &
+for i in $(seq 1 20); do curl -s -o /dev/null -w "%{time_total}\n" "http://127.0.0.1:80/ping"; done
+wait   # все замеры /ping должны быть заметно < 0.100 s
+```
+
+> Сценарии 2 и 3 (реальная скачка большого видео + реальный подсчёт fd через `lsof`) в
+> headless-тестах **намеренно не подделаны** — им нужен живой сервер, реальные загрузки и
+> реальные файловые дескрипторы. Дедуп-инвариант и disconnect-очистка доказаны на уровне
+> реестра/`get_media` (тесты стадии 7 №3), но «нет утечки fd на проде» проверяет оператор.
+
+## Diag-логи для наблюдения после деплоя
+
+Ожидаемая динамика (сравнить с до-деплойным baseline):
+
+- `diag_semaphore_wait` — ожидание HTTP-семафора должно **упасть** (реже/меньше секунд).
+- `diag_download_timing` — время скачивания стабилизируется; нет длинных «зависаний».
+- `diag_sanitize_slow` — почти **исчезнуть** (один sanitize на выходную границу, стадия 4).
+- `watchdog: heartbeat` — **продолжаются** штатно (живость TG-сессии).
+- На момент FloodWait — **нет всплеска 404** на `/media`; вместо этого `media_flood_wait`
+  и ответы **429** с `Retry-After` (стадия 2.3).
+- Признаки supervision: любые `supervised_task_crashed` / `supervised_task_exited` на
+  CRITICAL — сигнал разобраться, но задача при этом перезапускается (не тихая смерть).
+
+## План отката (rollback)
+
+Каждая стадия — отдельный, независимо ревертируемый коммит на своей ветке
+(`fix/stage-1-hangs` … `fix/stage-6-healthcheck`); порядок стадий = порядок деплоя.
+
+- Стадии 1 и 2 — низкорисковые, деплоятся первыми; при регрессе откатываются по одной
+  (`git revert <commit>` соответствующей стадии) без затрагивания остальных.
+- Стадия 3 (FileResponse) независима — реверт возвращает прежний ручной стриминг.
+- Стадия 4 требует, чтобы 4.2 откатывалась не позже 4.3 (иначе рендер-в-потоке остался бы
+  без безопасного пути сохранения media id) — откатывать стадию 4 целиком.
+- Стадии 5 и 6 независимы, откатываются по отдельности.
+- Стадия 7 — только тесты и этот документ: реверт ничего не меняет в проде.
+
+Прод-деплой, живое наблюдение diag-логов и обновление прод-compose (healthcheck→`/ping`,
+вне репозитория) — **зона ответственности оператора** (пункты 3–4 плана стадии 7).
diff --git a/tests/test_stage7_integration.py b/tests/test_stage7_integration.py
new file mode 100644
index 0000000..a950e23
--- /dev/null
+++ b/tests/test_stage7_integration.py
@@ -0,0 +1,349 @@
+# flake8: noqa
+# pylint: disable=protected-access, missing-function-docstring, missing-class-docstring
+# pylint: disable=redefined-outer-name, logging-fstring-interpolation, line-too-long
+# pylance: disable=reportMissingImports, reportMissingModuleSource
+"""
+Stage 7 — cross-stage END-TO-END integration tests.
+
+Unlike the per-stage suites (which pin one seam in isolation), these drive the real
+public entry points (`get_media`, `ping`, the access-time flush) so a regression that
+only shows up when the stages interact goes red. Every scenario here maps to one of the
+plan's "Стадия 7 — сквозные ручные сценарии"; the ones that genuinely need a running
+server + real downloads + lsof (fd-leak counting) are NOT faked here — they stay in
+docs/stability-verification.md for the operator's prod observation.
+
+Automated here:
+- Range semantics at the /media ROUTE level (get_media -> prepare_file_response ->
+  FileResponse driven through real ASGI): bytes=0-99 -> 206, bytes=-100 -> 206,
+  bytes=999999999- -> 416. (Stage 3 pins these on prepare_file_response directly; this
+  adds the end-to-end assertion that get_media's cache-hit branch reaches FileResponse
+  with a live Range still honored — i.e. stages 2+3 wired together.)
+- /ping (stage 6) stays fast + correct while a deliberately-slow op is in flight, issuing
+  ZERO Telegram RPC — proving the healthcheck is decoupled from the hot/blocked paths.
+- In-flight dedup + disconnect cleanup (stages 1/2) through the REAL get_media entry
+  point: concurrent requests share one download and the _inflight registry drains; a
+  cancelled request (client disconnect) leaves neither a stuck key nor a hung sibling.
+- str(channel) access-time key consistency (stage 5) end-to-end: a /media cache hit for
+  an int-ish channel records the str-keyed timestamp, and the flush UPDATE matches the
+  str-keyed DB row (hit -> flush -> DB), the exact affinity gotcha the plan warns about.
+"""
+import os
+import sys
+import time
+import sqlite3
+import asyncio
+from types import SimpleNamespace
+
+import pytest
+
+# Add project root to sys.path and mock the config module (same pattern as the other tests).
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
+sys.modules['config'] = __import__('tests.mock_config', fromlist=['get_settings'])
+
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+import api_server
+from pyrogram import errors
+from file_io import init_db_sync
+
+
+# 2048 deterministic bytes so Range slices are byte-checkable.
+BODY = bytes(range(256)) * 8
+SIZE = len(BODY)
+
+
+def _media_app():
+    """A bare app that mounts the REAL get_media (and ping) with NO lifespan, so
+    client.start() never runs — a pure cache hit never touches Telegram, and FileResponse
+    computes Range/206/416 at send time, which only happens when driven through ASGI."""
+    app = FastAPI()
+    app.add_api_route("/media/{channel}/{post_id}/{file_unique_id}/{digest}", api_server.get_media, methods=["GET"])
+    app.add_api_route("/media/{channel}/{post_id}/{file_unique_id}", api_server.get_media, methods=["GET"])
+    return app
+
+
+def _seed_cache(tmp_path, channel, post_id, fid, body=BODY):
+    cache_dir = tmp_path / "data" / "cache" / str(channel) / str(post_id)
+    cache_dir.mkdir(parents=True, exist_ok=True)
+    (cache_dir / fid).write_bytes(body)
+    return cache_dir / fid
+
+
+# --------------------------------------------------------------------------- #
+# Range semantics at the /media ROUTE level (stages 2 + 3 wired together).
+# Plan scenario: `curl -H "Range: bytes=0-99" / "bytes=-100" / "bytes=999999999-"`.
+# Regression caught: any change that makes get_media's cache-hit branch stop reaching
+# FileResponse (e.g. re-buffering the body, hand-rolling headers, dropping the media_key
+# path) or that breaks the digest gate wiring — the Range would stop being honored.
+# --------------------------------------------------------------------------- #
+def test_media_route_range_prefix_0_99(monkeypatch, tmp_path):
+    monkeypatch.chdir(tmp_path)
+    _seed_cache(tmp_path, "chan", 3, "fidR")
+    monkeypatch.setattr(api_server, "verify_media_digest", lambda url, digest: True)
+    # Keep the MIME path DB-free; the FileResponse/Range machinery is what we exercise.
+    monkeypatch.setattr(api_server, "get_mime_type_sync", lambda *a, **k: None)
+    monkeypatch.setattr(api_server, "set_mime_type_sync", lambda *a, **k: None)
+
+    c = TestClient(_media_app())
+    r = c.get("/media/chan/3/fidR/anydigest", headers={"Range": "bytes=0-99"})
+    assert r.status_code == 206
+    assert r.headers["content-range"] == f"bytes 0-99/{SIZE}"
+    assert r.headers["content-length"] == "100"
+    assert r.content == BODY[:100]
+    assert r.headers["accept-ranges"] == "bytes"
+
+
+def test_media_route_range_suffix_last_100(monkeypatch, tmp_path):
+    monkeypatch.chdir(tmp_path)
+    _seed_cache(tmp_path, "chan", 3, "fidS")
+    monkeypatch.setattr(api_server, "verify_media_digest", lambda url, digest: True)
+    monkeypatch.setattr(api_server, "get_mime_type_sync", lambda *a, **k: None)
+    monkeypatch.setattr(api_server, "set_mime_type_sync", lambda *a, **k: None)
+
+    c = TestClient(_media_app())
+    r = c.get("/media/chan/3/fidS/anydigest", headers={"Range": "bytes=-100"})
+    assert r.status_code == 206
+    assert r.headers["content-range"] == f"bytes {SIZE - 100}-{SIZE - 1}/{SIZE}"
+    assert r.headers["content-length"] == "100"
+    assert r.content == BODY[-100:]
+
+
+def test_media_route_range_unsatisfiable_416(monkeypatch, tmp_path):
+    monkeypatch.chdir(tmp_path)
+    _seed_cache(tmp_path, "chan", 3, "fidU")
+    monkeypatch.setattr(api_server, "verify_media_digest", lambda url, digest: True)
+    monkeypatch.setattr(api_server, "get_mime_type_sync", lambda *a, **k: None)
+    monkeypatch.setattr(api_server, "set_mime_type_sync", lambda *a, **k: None)
+
+    c = TestClient(_media_app())
+    r = c.get("/media/chan/3/fidU/anydigest", headers={"Range": "bytes=999999999-"})
+    assert r.status_code == 416
+    # Starlette's 416 Content-Range is `*/size` (documented stage-3 RFC-7233 delta).
+    assert r.headers["content-range"] == f"*/{SIZE}"
+
+
+# --------------------------------------------------------------------------- #
+# /ping (stage 6) stays fast + correct while a slow op is in flight, zero TG RPC.
+# Plan scenario: "Генерация фида на 100+ сообщений + параллельный /ping -> ping < 100 мс".
+# We model the concurrent slow op as a coroutine parked on an Event that is NEVER set
+# during the ping, and assert ping resolves while it is still pending AND touches no RPC.
+# Regression caught: re-coupling /ping to a TG RPC (get_me / safe_get_messages) or to any
+# awaitable that a blocked hot path could stall — the ping would no longer return promptly
+# or the spy counts would go non-zero.
+# --------------------------------------------------------------------------- #
+class _FakeKurigram:
+    def __init__(self, is_connected=True):
+        self.is_connected = is_connected
+        self.get_me_calls = 0
+
+    async def get_me(self):
+        self.get_me_calls += 1
+        raise AssertionError("/ping must never call get_me()")
+
+
+class _FakeTelegramClient:
+    def __init__(self, age, is_connected=True):
+        self._age = age
+        self.client = _FakeKurigram(is_connected=is_connected)
+        self.safe_get_messages_calls = 0
+
+    def watchdog_last_ok_age(self):
+        return self._age
+
+    async def safe_get_messages(self, *a, **k):
+        self.safe_get_messages_calls += 1
+        raise AssertionError("/ping must never call safe_get_messages()")
+
+
+async def test_ping_prompt_and_rpc_free_while_slow_op_pending(monkeypatch):
+    threshold = api_server.Config["tg_ping_unhealthy_after"]
+    fake = _FakeTelegramClient(age=threshold - 10, is_connected=True)
+    monkeypatch.setattr(api_server, "client", fake)
+
+    # A concurrent slow operation (a stand-in for a hung feed/RPC hot path) parked on an
+    # Event we deliberately never set for the duration of the ping.
+    gate = asyncio.Event()
+
+    async def slow_op():
+        await gate.wait()
+
+    slow = asyncio.create_task(slow_op())
+    await asyncio.sleep(0)  # let slow_op start and park on the gate
+
+    # The real proof of decoupling: ping() returns under a tight deadline while the slow op
+    # is parked, AND issues zero TG RPC. wait_for reds if ping ever blocks; the spies (which
+    # raise if touched) red if ping recouples to any RPC. `not slow.done()` is only a sanity
+    # check that ping did not somehow drive the parked op — the gate keeps it pending anyway.
+    resp = await asyncio.wait_for(api_server.ping(), timeout=1.0)
+
+    assert resp.status_code == 200            # correct health while connected + fresh
+    assert not slow.done()                    # sanity: slow op still parked, ping did not await it
+    assert fake.client.get_me_calls == 0      # decoupled: zero TG RPC
+    assert fake.safe_get_messages_calls == 0
+
+    gate.set()
+    await slow
+
+
+async def test_ping_reports_degraded_promptly_while_slow_op_pending(monkeypatch):
+    threshold = api_server.Config["tg_ping_unhealthy_after"]
+    fake = _FakeTelegramClient(age=threshold + 100, is_connected=True)  # stale probe
+    monkeypatch.setattr(api_server, "client", fake)
+
+    gate = asyncio.Event()
+
+    async def slow_op():
+        await gate.wait()
+
+    slow = asyncio.create_task(slow_op())
+    await asyncio.sleep(0)
+
+    resp = await asyncio.wait_for(api_server.ping(), timeout=1.0)
+
+    assert resp.status_code == 503            # stale watchdog probe -> degraded, still instant
+    assert not slow.done()                    # sanity: slow op still parked, ping did not await it
+    assert fake.client.get_me_calls == 0
+    assert fake.safe_get_messages_calls == 0
+
+    gate.set()
+    await slow
+
+
+# --------------------------------------------------------------------------- #
+# In-flight dedup + disconnect cleanup (stages 1/2) through the REAL get_media entry.
+# Plan scenario: "Параллельные запросы одного большого видео -> нет частичной отдачи" and
+# "Отключение клиента на середине стрима -> нет утечки тасков/фд".
+# The pure-unit stage-2 tests pin _download_deduped directly; these drive get_media so the
+# HTTP semaphore + dedup registry + serve path are proven wired together.
+# Regression caught: moving the download back into the request coroutine (so a disconnect
+# cancels it), or dropping the finally-pop, would leave a stuck _inflight key / hung sibling.
+# --------------------------------------------------------------------------- #
+async def test_get_media_concurrent_shares_one_download_and_drains_registry(monkeypatch, tmp_path):
+    monkeypatch.chdir(tmp_path)
+    api_server._inflight.clear()
+    monkeypatch.setattr(api_server, "verify_media_digest", lambda url, digest: True)
+    # Serve step is not under test here; keep it to a sentinel so we assert on dedup + registry.
+    sentinel = object()
+
+    async def fake_prepare(file_path, request=None, delete_after=False, media_key=None):
+        return sentinel
+
+    monkeypatch.setattr(api_server, "prepare_file_response", fake_prepare)
+
+    calls = []
+
+    async def slow_dl(channel, post_id, fid):
+        calls.append(fid)
+        await asyncio.sleep(0.05)  # real overlap window for the two requests
+        return (f"/final/{fid}", False)
+
+    monkeypatch.setattr(api_server, "download_media_file", slow_dl)
+
+    req = SimpleNamespace(headers={})
+    t1 = asyncio.create_task(api_server.get_media("chan", 9, "vfid", request=req, digest="x"))
+    t2 = asyncio.create_task(api_server.get_media("chan", 9, "vfid", request=req, digest="x"))
+    r1, r2 = await asyncio.gather(t1, t2)
+
+    assert r1 is sentinel and r2 is sentinel
+    assert calls == ["vfid"]            # exactly ONE real download, shared by both requests
+    assert not api_server._inflight     # registry drained — no forever-busy key
+
+
+async def test_get_media_request_cancel_does_not_stick_registry_or_hang_sibling(monkeypatch, tmp_path):
+    monkeypatch.chdir(tmp_path)
+    api_server._inflight.clear()
+    monkeypatch.setattr(api_server, "verify_media_digest", lambda url, digest: True)
+    sentinel = object()
+
+    async def fake_prepare(file_path, request=None, delete_after=False, media_key=None):
+        return sentinel
+
+    monkeypatch.setattr(api_server, "prepare_file_response", fake_prepare)
+
+    gate = asyncio.Event()
+    calls = []
+
+    async def held_dl(channel, post_id, fid):
+        calls.append(fid)
+        await gate.wait()  # hold the shared download open until we release it
+        return (f"/final/{fid}", False)
+
+    monkeypatch.setattr(api_server, "download_media_file", held_dl)
+
+    req = SimpleNamespace(headers={})
+    t1 = asyncio.create_task(api_server.get_media("chan", 9, "cfid", request=req, digest="x"))
+    await asyncio.sleep(0.02)  # t1 registers the future + starts the detached download
+    t2 = asyncio.create_task(api_server.get_media("chan", 9, "cfid", request=req, digest="x"))
+    await asyncio.sleep(0.02)
+
+    # First client disconnects: its request coroutine is cancelled mid-wait.
+    t1.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await t1
+
+    # The detached download is unaffected; releasing it resolves the surviving request.
+    gate.set()
+    r2 = await asyncio.wait_for(t2, timeout=2.0)
+    assert r2 is sentinel
+    assert calls == ["cfid"]           # download ran exactly once (not restarted)
+    assert not api_server._inflight     # registry drained despite the disconnect
+
+
+# --------------------------------------------------------------------------- #
+# str(channel) access-time key consistency (stage 5) END-TO-END: hit -> flush -> DB.
+# Plan gotcha #9: "Ключи SQLite: channel всегда str(...)"; if the key the hot path RECORDS
+# and the key the flush UPDATEs by ever diverge, the timestamp silently stops updating and
+# the file eventually falls out of the cache. Stage 5 pins hit and flush separately; this
+# stitches them: the /media cache-hit records a (str-channel) key, and the flush's bulk
+# UPDATE must land on that exact DB row. (get_media's route channel is already a str, so the
+# str() there is a defensive no-op; the live int/str hazard is in download_media_file, pinned
+# by stage-5's test_cache_hit_keys_channel_as_str — this test covers the get_media+flush leg.)
+# Regression caught: any accumulator-key vs UPDATE-WHERE inconsistency (e.g. a transposed
+# key-column order in the bulk SQL, verified to turn this test red) leaves `added` stale.
+# --------------------------------------------------------------------------- #
+async def test_media_cache_hit_flush_updates_str_keyed_db_row(monkeypatch, tmp_path):
+    monkeypatch.chdir(tmp_path)
+    api_server._access_updates = {}
+
+    channel, post_id, fid = "12345", 7, "fidINT"  # int-ish channel, passed as the route str
+    _seed_cache(tmp_path, channel, post_id, fid)
+
+    db = str(tmp_path / "access.db")
+    init_db_sync(db)
+    # Seed the row keyed by the STRING channel with an old access time.
+    conn = sqlite3.connect(db)
+    conn.execute(
+        "INSERT INTO media_file_ids (channel, post_id, file_unique_id, added) VALUES (?,?,?,?)",
+        (channel, post_id, fid, 1.0),
+    )
+    conn.commit()
+    conn.close()
+    monkeypatch.setattr(api_server, "DB_PATH", db)
+
+    monkeypatch.setattr(api_server, "verify_media_digest", lambda url, digest: True)
+    sentinel = object()
+
+    async def fake_prepare(file_path, request=None, media_key=None):
+        return sentinel
+
+    monkeypatch.setattr(api_server, "prepare_file_response", fake_prepare)
+
+    before = time.time()
+    resp = await api_server.get_media(channel, post_id, fid, request=SimpleNamespace(headers={}), digest="x")
+    assert resp is sentinel
+
+    # Hot path recorded the str-keyed access time (never the raw int form).
+    assert (channel, post_id, fid) in api_server._access_updates
+
+    # Flush the accumulator; the bulk UPDATE must match the str-keyed row and refresh `added`.
+    await api_server._flush_access_updates()
+    assert api_server._access_updates == {}
+
+    conn = sqlite3.connect(db)
+    added = conn.execute(
+        "SELECT added FROM media_file_ids WHERE channel = ? AND post_id = ? AND file_unique_id = ?",
+        (channel, post_id, fid),
+    ).fetchone()[0]
+    conn.close()
+    assert added >= before  # refreshed from the stale 1.0 to ~now via the str-keyed WHERE

From 3a133c349dc5c8f841e190344085a63143c4ca50 Mon Sep 17 00:00:00 2001
From: agent_coder <agent_coder@vvzvlad.xyz>
Date: Sun, 5 Jul 2026 11:17:49 +0300
Subject: [PATCH 6/6] =?UTF-8?q?docs(stability):=20stage=207=20review=20rou?=
 =?UTF-8?q?nd=201=20=E2=80=94=20correct=20rollback=20granularity=20to=20pe?=
 =?UTF-8?q?r-stage?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Review finding (low, doc-only): the rollback plan said each stage is one
independently-revertable commit, but each stage is actually two commits on its
branch (feature + review-round fix, the latter often fixing a real bug), so
reverting a single commit would orphan the review-round fix. Reword the unit of
rollback to the whole STAGE (branch/PR) and spell out the revert command per merge
strategy: squash-merge -> revert the one squash commit; merge-commit -> revert -m 1
the merge; linear history -> revert the full range of the stage's commits.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 docs/stability-verification.md | 18 ++++++++++++++----
 1 file changed, 14 insertions(+), 4 deletions(-)

diff --git a/docs/stability-verification.md b/docs/stability-verification.md
index 79958e4..699cf6d 100644
--- a/docs/stability-verification.md
+++ b/docs/stability-verification.md
@@ -123,11 +123,21 @@ wait   # все замеры /ping должны быть заметно < 0.100
 
 ## План отката (rollback)
 
-Каждая стадия — отдельный, независимо ревертируемый коммит на своей ветке
-(`fix/stage-1-hangs` … `fix/stage-6-healthcheck`); порядок стадий = порядок деплоя.
+Единица отката — **стадия целиком**, не отдельный коммит. Каждая стадия живёт на своей
+ветке/PR (`fix/stage-1-hangs` … `fix/stage-7-verification`) и, как правило, состоит из
+**двух коммитов**: feature-коммит + фикс review-раунда (последний нередко чинит реальный
+баг — напр. fail-closed XSS в стадии 4, watchdog-gate в стадии 6). Поэтому `git revert`
+одного коммита осиротит фикс review-раунда и даст несогласованный откат. Откатывать нужно
+на гранулярности стадии; порядок стадий = порядок деплоя.
 
-- Стадии 1 и 2 — низкорисковые, деплоятся первыми; при регрессе откатываются по одной
-  (`git revert <commit>` соответствующей стадии) без затрагивания остальных.
+- **Как откатывать стадию** — зависит от того, как ветки влиты в `fix/stability`/`main`:
+  - если стадия влита **squash-merge** (один коммит на стадию) — `git revert <squash-commit>`
+    откатывает её целиком, однозначно;
+  - если стадия влита **merge-коммитом** — `git revert -m 1 <merge-commit>` откатывает весь
+    вклад ветки (оба коммита) разом;
+  - если история линейная (rebase/fast-forward) — реверт **всех** коммитов стадии
+    (`git revert <feature>..<review-fix>` включительно), а не одного.
+- Стадии 1 и 2 — низкорисковые, деплоятся первыми; откатываются независимо от остальных.
 - Стадия 3 (FileResponse) независима — реверт возвращает прежний ручной стриминг.
 - Стадия 4 требует, чтобы 4.2 откатывалась не позже 4.3 (иначе рендер-в-потоке остался бы
   без безопасного пути сохранения media id) — откатывать стадию 4 целиком.