fix(diagnostics): deep-review round 1 — a send that cannot race, lose or repeat itself

The send, the draft and the screenshots take the session's row lock, so an
autosave or a paste during a send waits for it and then sees it, and two sends
cannot open two issues. A save keeps the delivery's progress; a failed edit
that shows the screenshots is finished by the retry; an answer that is not the
tracker's API (a redirect, a proxy page) is refused, not a crash that loses the
issue's record. Only an attempt that opened an issue counts toward the pace.

Screenshots: deleting one after the send is refused, a user's storage is
capped across sessions, new sessions and zip exports are paced, images are
stored uncompressed in the zip. One liveness rule, one session view, one image
type table.

Logging: a session's elevation is tracked apart from the user's own, so the
browser's own requests (no header) no longer lower the floor a session needs;
Stop ends it at once.

Composer: an image pasted anywhere is taken; what is still unsaved is saved on
close, and "saved" is shown only when it is; the draft is held while it is
sent; a failed screenshot read says so instead of reading as none; focus moves
to what stays; the preview follows the text size. Pill: the step prompt and a
running Stop survive the pill moving between hosts, the prompt keeps the
dialog's Tab trap, the toast region lets clicks through, other tabs hear of a
Start or Stop, the last seconds are sent before the cap, a capture with no
frame is an error.

backend/tests/test_debug_sessions.py

@@ -10,6 +10,7 @@ Four parts:
   * screenshots: judged by their bytes, capped in size and count, served back only to their owner.
 """
 
+import contextvars
 import json
 import logging
 from datetime import datetime, timedelta, timezone
@@ -274,6 +275,31 @@ def test_a_header_naming_anything_else_is_ignored_not_refused(gated, db, client,
         assert getattr(line, "elevated", False) is False, value
 
 
+def test_a_header_naming_the_own_session_past_its_cap_is_ignored(gated, db, client, make_user):
+    _signed_in(client, make_user)
+    sid = _start(client)["id"]
+    _expire(db, sid)  # past its cap, not yet swept: still open in the table
+    gated.records.clear()
+    client.get("/api/debug-sessions", headers={"x-debug-session": sid})
+    (line,) = _request_lines(gated)
+    assert getattr(line, "debug_session_id", None) is None
+    assert getattr(line, "elevated", False) is False
+
+
+def test_a_request_without_the_header_keeps_the_sessions_floor(gated):
+    """The browser's own requests (an <img>, a <video>) carry no header: they must not lower the
+    root level a concurrent request of the session still needs."""
+    root = logging.getLogger()
+    until = datetime.now(timezone.utc) + timedelta(minutes=30)
+    request = contextvars.copy_context()  # set_user writes the context: keep it this test's own
+    request.run(logs.set_user, 5, "TRACE", until, "a" * 16)
+    assert root.level == logs.resolve_level("TRACE")
+    request.run(logs.set_user, 5, "INFO", None, None)  # the same user, a request without the header
+    assert root.level == logs.resolve_level("TRACE")
+    logs.end_session_elevation("a" * 16)  # Stop
+    assert root.level == logging.INFO
+
+
 def test_a_session_never_lowers_the_users_own_level():
     now = datetime.now(timezone.utc)
     user = User(diagnostics_level="TRACE", diagnostics_until=now + timedelta(hours=3))
@@ -454,6 +480,75 @@ def test_a_screenshot_can_be_removed_and_is_then_gone(recording):
     assert r.status_code == 404 and r.json()["detail"]["reason"] == "screenshot"
 
 
+def test_a_sent_session_keeps_its_screenshots(db, recording):
+    client, sid = recording
+    shot = _upload(client, sid, _PNG).json()["screenshot"]["id"]
+    db.get(DebugSession, sid).issue_url = "https://git.example/o/r/issues/1"
+    db.commit()
+    # The issue carries it already, and a half-sent one names its screenshots by their order.
+    r = client.delete(f"/api/debug-sessions/{sid}/screenshots/{shot}")
+    assert r.status_code == 409 and r.json()["detail"]["reason"] == "screenshot_already_sent"
+
+
+def test_a_users_screenshots_are_capped_across_their_sessions(recording, monkeypatch):
+    client, sid = recording
+    monkeypatch.setattr(debugsession, "SCREENSHOT_USER_MAX_BYTES", len(_PNG) * 2)
+    assert _upload(client, sid, _PNG).status_code == 200
+    client.post(f"/api/debug-sessions/{sid}/stop")
+    other = _start(client)["id"]
+    assert _upload(client, other, _PNG).status_code == 200
+    r = _upload(client, other, _PNG)
+    assert r.status_code == 422 and r.json()["detail"]["reason"] == "screenshot_storage_full"
+
+
+def test_new_sessions_are_paced_but_joining_one_is_not(db, make_user, as_user, monkeypatch):
+    from app.ratelimit import RateLimiter
+    from app.routes import debug_sessions as routes
+
+    monkeypatch.setattr(routes, "_new_sessions", RateLimiter(max_events=1, window_seconds=3600))
+    client = as_user(make_user())
+    sid = _start(client)["id"]
+    assert _start(client)["id"] == sid  # another tab joining
+    client.post(f"/api/debug-sessions/{sid}/stop")
+    r = client.post("/api/debug-sessions", json={"level": "DEBUG"})
+    assert r.status_code == 429 and r.json()["detail"]["reason"] == "debug_session_rate_limited"
+
+
+def test_two_starts_at_once_join_the_one_that_won(db, make_user, as_user, monkeypatch):
+    user = make_user()
+    client = as_user(user)
+    won = _start(client)["id"]
+
+    def raced(db_, user_, level):
+        # The loser of the race: past the "is one open?" read, its insert meets the index.
+        db_.add(
+            DebugSession(
+                id="0" * 16,
+                user_id=user_.id,
+                level=level,
+                started_at=datetime.now(timezone.utc),
+                until=datetime.now(timezone.utc) + timedelta(minutes=5),
+            )
+        )
+        db_.flush()
+
+    monkeypatch.setattr(debugsession, "start", raced)
+    r = client.post("/api/debug-sessions", json={"level": "DEBUG"})
+    assert r.status_code == 200, r.text
+    assert r.json()["session"]["id"] == won
+
+
+def test_the_export_is_paced(recording, monkeypatch):
+    from app.ratelimit import RateLimiter
+    from app.routes import debug_sessions as routes
+
+    client, sid = recording
+    monkeypatch.setattr(routes, "_exports", RateLimiter(max_events=1, window_seconds=60))
+    assert client.get(f"/api/debug-sessions/{sid}/export").status_code == 200
+    r = client.get(f"/api/debug-sessions/{sid}/export")
+    assert r.status_code == 429 and r.json()["detail"]["reason"] == "export_rate_limited"
+
+
 def test_the_upload_path_left_out_of_the_evidence_is_the_real_one():
     assert debugsession.INGEST_PATH in {r.path for r in diagnostics_routes.router.routes}
 
@@ -565,6 +660,9 @@ class FakeGitea:
         self.has_issues = True
         self.token_ok = True
         self.fail_asset: str | None = None
+        self.fail_patch = False
+        self.redirect = False  # an http:// address behind a proxy that sends it to https
+        self.garbage_asset = False  # a 2xx that is not the API's answer (a proxy's page)
         self.patched_body: str | None = None
 
     def handler(self, request: httpx.Request) -> httpx.Response:
@@ -572,6 +670,8 @@ class FakeGitea:
         self.calls.append((request.method, path))
         if request.headers.get("authorization") != "token t0ken" or not self.token_ok:
             return httpx.Response(401, json={"message": "invalid username, password or token"})
+        if self.redirect:
+            return httpx.Response(301, headers={"location": "https://git.example/"}, text="moved")
         if request.method == "GET" and path == "/repos/o/r":
             return httpx.Response(
                 200,
@@ -591,10 +691,16 @@ class FakeGitea:
             if name == self.fail_asset:
                 self.fail_asset = None
                 return httpx.Response(500, json={"message": "boom"})
+            if self.garbage_asset:
+                self.garbage_asset = False
+                return httpx.Response(200, text="<html>sign in</html>")
             return httpx.Response(
                 201, json={"name": name, "browser_download_url": f"https://git.example/a/{name}"}
             )
         if request.method == "PATCH" and path == "/repos/o/r/issues/7":
+            if self.fail_patch:
+                self.fail_patch = False
+                return httpx.Response(502, json={"message": "bad gateway"})
             self.patched_body = json.loads(request.content)["body"]
             return httpx.Response(201, json={"number": 7})
         if request.method == "POST" and path == "/markdown":
@@ -698,6 +804,53 @@ def test_a_send_that_stops_half_way_keeps_the_issue_and_the_retry_finishes_it(
     assert len(gitea_fake.uploads()) == 3  # client.log, the failed server.log, then server.log
 
 
+def test_a_failed_edit_that_shows_the_screenshots_is_finished_by_the_retry(
+    gitea_fake, db, client, make_user
+):
+    gitea_fake.fail_patch = True
+    _signed_in(client, make_user)
+    sid = _start(client)["id"]
+    _upload(client, sid, _PNG)
+    _ready(client, sid)
+    r = client.post(f"/api/debug-sessions/{sid}/issue")
+    assert r.status_code == 422 and r.json()["detail"]["reason"] == "issues_unreachable"
+    # Every file is up, but the body does not show the screenshot yet: the send is not finished.
+    assert client.get(f"/api/debug-sessions/{sid}").json()["pending_files"] == ["screenshot-1.png"]
+    r = client.post(f"/api/debug-sessions/{sid}/issue")
+    assert r.status_code == 200 and r.json()["session"]["pending_files"] == []
+    assert "![screenshot-1.png](https://git.example/a/screenshot-1.png)" in gitea_fake.patched_body
+    assert len(gitea_fake.uploads()) == 3  # nothing uploaded twice
+
+
+def test_an_answer_that_is_not_the_api_is_refused_and_the_issue_stays_linked(
+    gitea_fake, db, client, make_user
+):
+    gitea_fake.garbage_asset = True
+    _signed_in(client, make_user)
+    sid = _start(client)["id"]
+    _ready(client, sid)
+    r = client.post(f"/api/debug-sessions/{sid}/issue")
+    assert r.status_code == 422 and r.json()["detail"]["reason"] == "issues_rejected"
+    assert client.get(f"/api/debug-sessions/{sid}").json()["issue_url"] is not None
+    assert client.post(f"/api/debug-sessions/{sid}/issue").status_code == 200
+    # Still the one issue.
+    assert [m for m, p in gitea_fake.calls if p == "/repos/o/r/issues"] == ["POST"]
+
+
+def test_a_redirecting_tracker_is_refused_not_a_crash(gitea_fake, db, make_user, as_user):
+    gitea_fake.redirect = True
+    admin = as_user(make_user(role="admin"))
+    r = admin.post("/api/admin/config/test-issues")
+    assert r.status_code == 422 and r.json()["detail"]["reason"] == "issues_rejected"
+
+
+def test_the_delivery_survives_a_save_of_the_draft():
+    session = DebugSession(draft={"title": "t", "delivery": {"number": 7, "pending": ["a.log"]}})
+    debugsession.save_draft(session, {"title": "t2", "body": "b", "excluded": [], "form": {}})
+    assert session.draft["title"] == "t2"
+    assert debugsession.unfinished(session) == ["a.log"]
+
+
 def test_a_report_needs_a_title_and_a_wrong_token_says_so(gitea_fake, db, client, make_user):
     _signed_in(client, make_user)
     sid = _start(client)["id"]
@@ -723,7 +876,27 @@ def test_new_issues_are_paced_per_user(gitea_fake, db, make_user, as_user, monke
     second = _start(client)["id"]
     _ready(client, second)
     r = client.post(f"/api/debug-sessions/{second}/issue")
-    assert r.status_code == 429 and r.json()["detail"]["reason"] == "issues_rate_limited"
+    assert r.status_code == 429 and r.json()["detail"] == {
+        "code": "refused",
+        "reason": "issues_rate_limited",
+        "limit": 1,
+    }
+
+
+def test_an_attempt_refused_before_the_issue_opened_is_not_counted(
+    gitea_fake, db, make_user, as_user, monkeypatch
+):
+    from app.ratelimit import RateLimiter
+    from app.routes import debug_sessions as routes
+
+    monkeypatch.setattr(routes, "_new_issues", RateLimiter(max_events=1, window_seconds=3600))
+    client = as_user(make_user())
+    sid = _start(client)["id"]
+    _ready(client, sid)
+    gitea_fake.token_ok = False
+    assert client.post(f"/api/debug-sessions/{sid}/issue").status_code == 422
+    gitea_fake.token_ok = True
+    assert client.post(f"/api/debug-sessions/{sid}/issue").status_code == 200
 
 
 def test_the_preview_is_the_trackers_own_rendering(gitea_fake, db, make_user, as_user):