fix : stabilisation-1

.github/workflows/ci-test.yml

@@ -212,6 +212,14 @@ jobs:
           fi
           echo "✓ No silent TOML unwrap_or_default"
 
+      - name: Lint — no unwrap() inside with_conn closures
+        run: |
+          # stab-1 — a panic inside a with_conn closure poisons the shared
+          # DB mutex (recovered since 0.8.11, but still one failed request).
+          # Baseline is ZERO in prod code: keep it that way.
+          python3 scripts/ci/test_lint_with_conn_unwrap.py
+          python3 scripts/ci/lint_with_conn_unwrap.py
+
       - name: cargo test
         run: cargo test
         env:

backend/scripts/ci/lint_with_conn_unwrap.py

@@ -0,0 +1,123 @@
+#!/usr/bin/env python3
+"""CI anti-regression lint (stab-1): no `.unwrap()` inside `with_conn`
+closures in production code. A panic there poisons the shared DB mutex
+(recovered since 0.8.11, but still fails the request). Test modules
+(everything after the first `#[cfg(test)]` in a file) are exempt.
+Run from `backend/`: `python3 scripts/ci/lint_with_conn_unwrap.py`."""
+import pathlib
+import re
+import sys
+
+
+def _sanitize(text):
+    """Blank out string/char literals and comments (same length, newlines
+    kept so line numbers survive). Without this, a `)` inside a literal
+    closed the paren scan early and hid unwraps (Codex review); it also
+    prevents a literal ".unwrap()" in a string from false-positiving.
+    Handles: "…" with escapes, raw strings r"…" / r#"…"#…, char literals
+    (without eating lifetimes like 'a), // line and nested /* */ comments."""
+    out = list(text)
+    i, n = 0, len(text)
+
+    def blank(a, b):
+        for k in range(a, min(b, n)):
+            if out[k] != "\n":
+                out[k] = " "
+
+    while i < n:
+        c = text[i]
+        two = text[i : i + 2]
+        if two == "//":
+            j = text.find("\n", i)
+            j = n if j == -1 else j
+            blank(i, j)
+            i = j
+        elif two == "/*":
+            depth, j = 1, i + 2
+            while j < n and depth:
+                if text[j : j + 2] == "/*":
+                    depth += 1
+                    j += 2
+                elif text[j : j + 2] == "*/":
+                    depth -= 1
+                    j += 2
+                else:
+                    j += 1
+            blank(i, j)
+            i = j
+        elif c == '"' or (c == "r" and re.match(r'r#*"', text[i:])):
+            if c == '"':
+                j = i + 1
+                while j < n:
+                    if text[j] == "\\":
+                        j += 2
+                    elif text[j] == '"':
+                        j += 1
+                        break
+                    else:
+                        j += 1
+            else:
+                m = re.match(r'r(#*)"', text[i:])
+                closer = '"' + m.group(1)
+                j = text.find(closer, i + len(m.group(0)))
+                j = n if j == -1 else j + len(closer)
+            blank(i, j)
+            i = j
+        elif c == "'":
+            # Char literal ('x', '\n', '\u{…}') vs lifetime ('a) — a char
+            # literal always has a CLOSING quote within a few chars.
+            m = re.match(r"'(\\(?:u\{[0-9a-fA-F]{1,6}\}|.)|[^\\'])'", text[i:])
+            if m:
+                blank(i, i + m.end())
+                i += m.end()
+            else:
+                i += 1  # lifetime: skip the quote, keep the identifier
+        else:
+            i += 1
+    return "".join(out)
+
+
+def find_violations(text, path="<mem>"):
+    """`path:line` for every `.unwrap()` inside a `with_conn(...)` call.
+
+    Runs on SANITIZED text (strings/comments blanked) and scans to the
+    BALANCED closing paren with no length cap (Codex review, findings 2+3:
+    a 4k cap skipped long closures; literal parens broke the balance).
+    An unbalanced call is a hard error — loud beats a silent blind spot.
+    """
+    viol = []
+    prod = _sanitize(re.split(r"#\[cfg\(test\)\]", text)[0])
+    for m in re.finditer(r"with_conn(?:_blocking)?\s*\(", prod):
+        i, depth, end = m.end() - 1, 0, None
+        for j in range(i, len(prod)):
+            c = prod[j]
+            if c == "(":
+                depth += 1
+            elif c == ")":
+                depth -= 1
+                if depth == 0:
+                    end = j
+                    break
+        if end is None:
+            raise RuntimeError(
+                f"{path}: unbalanced parens after with_conn at offset {i} — lint cannot scan"
+            )
+        for um in re.finditer(r"\.unwrap\(\)", prod[i:end]):
+            line = prod[: i + um.start()].count("\n") + 1
+            viol.append(f"{path}:{line}")
+    return viol
+
+
+def main():
+    viol = []
+    for f in pathlib.Path("src").rglob("*.rs"):
+        viol.extend(find_violations(f.read_text(), str(f)))
+    if viol:
+        print("::error::unwrap() inside a with_conn closure panics the DB request — return an error instead:")
+        print("\n".join(viol))
+        sys.exit(1)
+    print("OK: no unwrap() inside with_conn closures")
+
+
+if __name__ == "__main__":
+    main()

backend/scripts/ci/test_lint_with_conn_unwrap.py

@@ -0,0 +1,96 @@
+#!/usr/bin/env python3
+"""Fixture tests for the with_conn-unwrap CI lint (Codex review: the
+4k-char scan cap silently skipped long closures)."""
+import importlib.util
+import pathlib
+import unittest
+
+_spec = importlib.util.spec_from_file_location(
+    "lint_wcu", pathlib.Path(__file__).parent / "lint_with_conn_unwrap.py")
+_mod = importlib.util.module_from_spec(_spec)
+_spec.loader.exec_module(_mod)
+
+
+class LintWithConnUnwrap(unittest.TestCase):
+    def test_flags_unwrap_placed_beyond_4k_chars(self):
+        # The exact blind spot: a closure body longer than the old cap.
+        filler = '            let _x = "' + "a" * 4200 + '";\n'
+        src = (
+            "fn f(db: &Db) {\n"
+            "    db.with_conn(move |conn| {\n"
+            + filler +
+            "        conn.query_row(q, [], |r| r.get(0)).unwrap();\n"
+            "        Ok(())\n"
+            "    });\n"
+            "}\n"
+        )
+        viol = _mod.find_violations(src, "long.rs")
+        self.assertEqual(len(viol), 1, "unwrap after 4k chars MUST be flagged")
+
+    def test_clean_closure_and_test_module_are_exempt(self):
+        src = (
+            "fn f(db: &Db) {\n"
+            "    db.with_conn(|conn| Ok(conn.count()?));\n"
+            "}\n"
+            "#[cfg(test)]\n"
+            "mod tests {\n"
+            "    fn t(db: &Db) { db.with_conn(|c| Ok(c.count().unwrap())); }\n"
+            "}\n"
+        )
+        self.assertEqual(_mod.find_violations(src, "clean.rs"), [])
+
+    def test_unwrap_outside_the_closure_is_not_flagged(self):
+        src = (
+            "fn f(db: &Db) {\n"
+            "    let v = db.with_conn(|conn| Ok(conn.count()?));\n"
+            "    v.unwrap();\n"  # caller-side unwrap: out of this lint's scope
+            "}\n"
+        )
+        self.assertEqual(_mod.find_violations(src, "outside.rs"), [])
+
+    def test_literal_paren_in_string_does_not_hide_the_unwrap(self):
+        # Codex finding 3 — exact repro: a ")" string literal closed the
+        # scan artificially and the unwrap after it was invisible.
+        src = (
+            "fn f(db: &Db) {\n"
+            "    db.with_conn(|conn| {\n"
+            '        let text = ")";\n'
+            "        conn.query_row(q, [], |r| r.get(0)).unwrap();\n"
+            "        Ok(())\n"
+            "    });\n"
+            "}\n"
+        )
+        self.assertEqual(len(_mod.find_violations(src, "lit.rs")), 1)
+
+    def test_parens_in_comments_raw_strings_and_chars_are_ignored(self):
+        src = (
+            "fn f(db: &Db) {\n"
+            "    db.with_conn(|conn| {\n"
+            "        // closing ) in a comment\n"
+            "        /* nested /* )) */ still ) comment */\n"
+            '        let raw = r#"raw ) text"#;\n'
+            "        let ch = ')';\n"
+            "        let lt: &'static str = x;\n"
+            "        conn.count().unwrap();\n"
+            "        Ok(())\n"
+            "    });\n"
+            "}\n"
+        )
+        self.assertEqual(len(_mod.find_violations(src, "mix.rs")), 1,
+                         "the real unwrap is still flagged, literal parens ignored")
+
+    def test_unwrap_text_inside_a_string_is_not_flagged(self):
+        src = (
+            "fn f(db: &Db) {\n"
+            '    db.with_conn(|conn| { let s = ".unwrap()"; Ok(s.len()) });\n'
+            "}\n"
+        )
+        self.assertEqual(_mod.find_violations(src, "strunwrap.rs"), [])
+
+    def test_unbalanced_call_fails_loudly(self):
+        with self.assertRaises(RuntimeError):
+            _mod.find_violations("db.with_conn(|c| {", "broken.rs")
+
+
+if __name__ == "__main__":
+    unittest.main()

backend/scripts/disc-introspection-mcp.py

@@ -32,6 +32,7 @@
 import json
 import os
 import sys
+import time
 import urllib.error
 import urllib.parse
 import urllib.request
@@ -149,7 +150,11 @@
             "content, agent_type}, …]` to push a whole conversation "
             "history at once. Idempotent on (disc_id, source_msg_id) "
             "— re-pushing the same transcript does NOT duplicate.\n\n"
-            "Returns `{appended, skipped_as_duplicates, diverged}`. "
+            "Returns `{appended, skipped_as_duplicates, diverged, "
+            "last_sort_order}`. ALWAYS use `last_sort_order` as the "
+            "`since_sort_order` of your next `disc_wait_for_peer` — "
+            "NEVER estimate your position (+1 per post drifts under "
+            "concurrent posters and silently skips messages). "
             "`diverged=true` means the Kronn UI was edited after a "
             "previous import — warn the user before more updates."
         ),
@@ -2006,6 +2011,30 @@ def _http(method, path, body=None):
         raise RuntimeError(f"HTTP {e.code}: {body[:500]}")
 
 
+def _http_transport_retry(method, path, attempts=6, delays=(2, 4, 8, 12, 16)):
+    """`_http` with a BOUNDED retry on TRANSPORT failures only (connection
+    refused/reset, remote disconnect, socket timeout) — the signature of a
+    backend restart, e.g. `cargo watch` rebuilding for 30-60s. HTTP errors
+    (4xx/5xx) are application-level and never retried. Safe only for
+    idempotent calls: the caller re-sends the same request verbatim.
+    Total worst-case wait ≈ sum(delays) ≈ 42s + in-flight time."""
+    last_err = None
+    for i in range(attempts):
+        try:
+            return _http(method, path)
+        except RuntimeError:
+            raise  # HTTPError path from _http — application error, no retry
+        except (urllib.error.URLError, ConnectionError, TimeoutError, OSError) as e:
+            last_err = e
+            if i + 1 < attempts:
+                time.sleep(delays[min(i, len(delays) - 1)])
+    raise RuntimeError(
+        f"backend unreachable after {attempts} attempts (~{sum(delays)}s — rebuild in "
+        f"progress?): {last_err}. Nothing is lost: messages persist in the DB — call "
+        "disc_wait_for_peer again with the SAME since_sort_order."
+    )
+
+
 def _http_text(method, path):
     """Variant of `_http` for endpoints that ship raw text (not JSON / not the
     `ApiResponse` envelope) — e.g. `/api/conventions/agents-md-format-v1`
@@ -2406,7 +2435,9 @@ def call_disc_wait_for_peer(args):
         params["exclude_agent_type"] = exclude
     qs = urllib.parse.urlencode(params)
     sep = "?" if qs else ""
-    result = _unwrap(_http("GET", f"/api/discussions/{disc_id}/wait{sep}{qs}"))
+    # Transport-level retry (bounded): a backend restart mid-poll must not
+    # surface as a tool error — the wait is idempotent on since_sort_order.
+    result = _unwrap(_http_transport_retry("GET", f"/api/discussions/{disc_id}/wait{sep}{qs}"))
     # A timed-out wait (no peer activity in the window) is NORMAL in an ongoing
     # collaboration — but literal agents (notably Codex) otherwise read the empty
     # result as "conversation over" and STOP after ~60s. Surface an explicit
@@ -2417,7 +2448,10 @@ def call_disc_wait_for_peer(args):
             "agent may still be thinking. Call disc_wait_for_peer AGAIN to keep "
             "waiting (pass latest_sort_order as since_sort_order). Do NOT stop "
             "or disc_leave() just because the wait timed out — only leave when "
-            "the task is done or the user explicitly says stop."
+            "the task is done or the user explicitly says stop. PACING: apply "
+            "the room's poll_policy (from disc_join/disc_meta) — consecutive "
+            "empty waits back off 30s,30s,1m,1m,2m,2m,4m,4m then cap at 8m; "
+            "reset to 30s as soon as a peer message arrives."
         )
     return result
 

backend/scripts/test_disc_introspection_mcp.py

@@ -1335,6 +1335,57 @@ def setUp(self):
         self.addCleanup(self.env_patch.stop)
         self.mod = _load_module()
 
+    def test_transport_cut_is_retried_with_the_same_request(self):
+        # Passe stab-1 — a backend restart (cargo watch) mid-poll used to
+        # surface as a tool error and drop the agent out of the room. The
+        # bridge now retries transport failures, SAME query string (the
+        # since_sort_order makes the resume idempotent).
+        import urllib.error
+        ok = {
+            "success": True,
+            "data": {"timed_out": True, "messages": [], "latest_sort_order": 12},
+        }
+        calls = []
+
+        def flaky(method, path, body=None):
+            calls.append(path)
+            if len(calls) < 3:
+                raise urllib.error.URLError(ConnectionRefusedError(61, "refused"))
+            return ok
+
+        with mock.patch.object(self.mod, "_http", side_effect=flaky), \
+             mock.patch.object(self.mod.time, "sleep") as mock_sleep:
+            result = self.mod.call_disc_wait_for_peer({"since_sort_order": 12})
+        self.assertEqual(len(calls), 3, "two failures then success")
+        self.assertTrue(all(p == calls[0] for p in calls), "identical request each attempt")
+        self.assertTrue(result["timed_out"])
+        self.assertEqual(mock_sleep.call_count, 2, "bounded backoff between attempts")
+
+    def test_transport_retry_is_bounded_and_names_the_resume_contract(self):
+        import urllib.error
+        with mock.patch.object(
+            self.mod, "_http",
+            side_effect=urllib.error.URLError(ConnectionRefusedError(61, "refused")),
+        ) as mock_http, mock.patch.object(self.mod.time, "sleep"):
+            with self.assertRaises(RuntimeError) as ctx:
+                self.mod.call_disc_wait_for_peer({"since_sort_order": 5})
+        self.assertEqual(mock_http.call_count, 6, "bounded — never infinite")
+        msg = str(ctx.exception)
+        self.assertIn("since_sort_order", msg, "the error must teach the resume contract")
+        self.assertIn("unreachable", msg)
+
+    def test_http_application_errors_are_never_retried(self):
+        # A 4xx/5xx is an app-level answer, not a transport cut — retrying
+        # would hammer the backend and mask real errors.
+        with mock.patch.object(
+            self.mod, "_http",
+            side_effect=RuntimeError("HTTP 404: nope"),
+        ) as mock_http, mock.patch.object(self.mod.time, "sleep") as mock_sleep:
+            with self.assertRaises(RuntimeError):
+                self.mod.call_disc_wait_for_peer({"since_sort_order": 5})
+        self.assertEqual(mock_http.call_count, 1, "no retry on HTTP errors")
+        mock_sleep.assert_not_called()
+
     def test_forwards_since_and_timeout_in_query_string(self):
         with mock.patch.object(self.mod, "_http") as mock_http:
             mock_http.return_value = {