refactor(gateway,tui-gateway): one settlement rule for batch clarify in both registries

Main's sidecar (tui_gateway/server_requests.py) and the session owner
(gateway/session_sudo_prompt.py) each computed a batch clarify's terminal
{answers, outcome} in their own code, and the owner's copy had already
drifted twice (Esc reported as submitted, timeout dropping locked
answers). Both now call tools.clarify_batch.settle_batch(locked,
frame_result, ended=...), a pure function: the last lock, a timeout or an
interrupt/cancel keep the locks with that outcome; a closing frame with
an answers dict lays them over the locks; anything else is a cancel-all
that keeps the locks as cancelled.

One sidecar change comes with it: a closing frame that names its own
outcome (ClarifyResult documents timed_out for the classic CLI's own
deadline) now keeps it, where the sidecar used to overwrite it with
submitted. The owner already honoured it.

Test: tests/gateway/test_clarify_batch_settlement_parity.py drives both
real registries through the same six endings after one lock (red: no
module, then the sidecar's frame_names_its_outcome case; green).

Review finding by @alt-glitch on #140472
Refs #134889 (TUI-18)
Refs #140337

gateway/session_sudo_prompt.py

@@ -15,6 +15,8 @@ import logging
 import threading
 import uuid
 
+from tools.clarify_batch import settle_batch
+
 logger = logging.getLogger(__name__)
 
 SUDO_TIMEOUT_S = 120  # main: ``_ask("sudo", ..., timeout=120)``
@@ -39,7 +41,6 @@ class _Pending:
 
 _lock = threading.Lock()
 _open: dict[str, _Pending] = {}
-_OUTCOMES = ("submitted", "timed_out", "cancelled")
 
 
 def owned_by(authority, session_id, events) -> bool:
@@ -71,20 +72,6 @@ def is_response_frame(frame) -> bool:
             and frame["id"].startswith("srq-") and ("result" in frame or "error" in frame))
 
 
-def _settle_batch(locked, result):
-    """A batch clarify's terminal result: answers locked early belong to the final set whatever ends
-    the wait, as main's ``server_requests`` merged them. A closing frame with an ``answers`` dict lays
-    its answers over the locks (the frame wins per question) and keeps its own outcome (``submitted``
-    when it names none); a frame without one (Ink Esc, Desktop Skip) or an error is a cancel-all."""
-    if isinstance(result, dict) and isinstance(result.get("answers"), dict):
-        outcome = result.get("outcome")
-        if outcome not in _OUTCOMES:
-            outcome = "submitted"
-        return {**result, "answers": {**locked, **result["answers"]}, "outcome": outcome}
-    outcome = result.get("outcome") if isinstance(result, dict) else None
-    return {"answers": locked, "outcome": outcome if outcome in ("timed_out", "cancelled") else "cancelled"}
-
-
 def resolve(frame, *, session_ok=lambda session_id, events: True) -> bool:
     """Settle the open request *frame* answers; False when nothing (here) waits for that id.
     ``session_ok(session_id, events)`` lets the caller refuse an answer from a connection not entitled
@@ -98,7 +85,8 @@ def resolve(frame, *, session_ok=lambda session_id, events: True) -> bool:
         locked = dict(pending.locked) if pending.qids is not None else None
     result = frame.get("result") if "error" not in frame else None
     if locked is not None:
-        result = _settle_batch(locked, result)
+        # One rule with main's sidecar registry: the frame's answers over the locks, or a cancel-all.
+        result = settle_batch(locked, result)
     value = result.get("value") if isinstance(result, dict) else ""
     pending.value = value if isinstance(value, str) else ""
     pending.result = result if isinstance(result, dict) else None
@@ -145,7 +133,7 @@ def lock_answer(request_id, question_id, answer, *, session_ok=lambda session_id
         remaining = [qid for qid in pending.qids if qid not in pending.locked]
         if not remaining:
             _open.pop(request_id, None)
-            pending.result = {"answers": dict(pending.locked), "outcome": "submitted"}
+            pending.result = settle_batch(pending.locked, ended="submitted")
     if not remaining:
         pending.event.set()
     return remaining
@@ -174,8 +162,8 @@ def _ask(events, session_id, method, params, *, cancelled, timeout, qids=None):
         if still_open and pending.qids is not None:
             # A batch keeps every answer locked before the wait ended (main's sidecar: "locked
             # answers survive a timeout"); sudo/secret have no question ids and stay unanswered.
-            pending.result = {"answers": dict(pending.locked),
-                              "outcome": "cancelled" if reason == "interrupted" else "timed_out"}
+            pending.result = settle_batch(pending.locked,
+                                          ended="cancelled" if reason == "interrupted" else "timed_out")
     if still_open:
         # The same teardown main's renderers key on (``request.cancel``): the card closes and a
         # timed-out password prompt says so.

tests/gateway/test_clarify_batch_settlement_parity.py

@@ -0,0 +1,115 @@
+"""One settlement rule for a batch clarify, whichever registry holds it.
+
+Main's sidecar (``tui_gateway/server_requests.py``) and the session owner (``gateway/session_sudo_prompt.py``)
+each keep open server→client requests, and each used to compute a batch's terminal ``{answers, outcome}``
+in its own code, which drifted twice (#140472 review). Both now call ``tools.clarify_batch.settle_batch``.
+These cases drive each real registry through the same ending, one ``clarify.lock`` on ``q0`` first, and
+require the same result.
+"""
+import threading
+import time
+
+import pytest
+
+from tools.clarify_batch import settle_batch
+
+QUESTIONS = [{'qid': 'q0', 'question': 'colour?', 'choices': ['red', 'blue'], 'multi_select': False},
+             {'qid': 'q1', 'question': 'size?', 'choices': None, 'multi_select': False}]
+QIDS = ['q0', 'q1']
+
+# ending -> (how the wait ends after q0 is locked, expected result)
+CASES = {
+    'cancel_all_frame': ({'result': {}}, {'answers': {'q0': 'blue'}, 'outcome': 'cancelled'}),
+    'closing_frame_with_tail': ({'result': {'answers': {'q1': 'large'}}},
+                                {'answers': {'q0': 'blue', 'q1': 'large'}, 'outcome': 'submitted'}),
+    'frame_names_its_outcome': ({'result': {'answers': {'q1': 'large'}, 'outcome': 'timed_out'}},
+                                {'answers': {'q0': 'blue', 'q1': 'large'}, 'outcome': 'timed_out'}),
+    'frame_overrides_a_lock': ({'result': {'answers': {'q0': 'red', 'q1': 'large'}}},
+                               {'answers': {'q0': 'red', 'q1': 'large'}, 'outcome': 'submitted'}),
+    'last_lock': ('lock', {'answers': {'q0': 'blue', 'q1': 'large'}, 'outcome': 'submitted'}),
+    'timeout': ('timeout', {'answers': {'q0': 'blue'}, 'outcome': 'timed_out'}),
+}
+
+
+def _wait_for(predicate, timeout=5.0):
+    deadline = time.monotonic() + timeout
+    while not predicate():
+        assert time.monotonic() < deadline
+        time.sleep(0.01)
+
+
+def _sidecar(monkeypatch, ending):
+    from tui_gateway import server_requests
+    frames = []
+    monkeypatch.setattr(server_requests, '_write', frames.append)
+    monkeypatch.setattr(server_requests, '_emit', lambda *a: None)
+    monkeypatch.setattr(server_requests, '_answerable', lambda sid: True)
+    server_requests.reset_for_tests()
+    out = {}
+    wire = {'questions': QUESTIONS}
+    timeout = 0.5 if ending == 'timeout' else 10
+    worker = threading.Thread(target=lambda: out.update(
+        result=server_requests.send('clarify', 's1', wire, timeout=timeout, qids=QIDS)))
+    worker.start()
+    try:
+        _wait_for(lambda: frames)
+        rid = frames[0]['id']
+        assert server_requests.lock_answer(rid, 'q0', 'blue') == ['q1']
+        if ending == 'lock':
+            assert server_requests.lock_answer(rid, 'q1', 'large') == []
+        elif ending != 'timeout':
+            assert server_requests.resolve_response({'jsonrpc': '2.0', 'id': rid, **ending})
+        worker.join(5)
+        return out['result']
+    finally:
+        server_requests.reset_for_tests()
+
+
+def _owner(ending):
+    from gateway.session_events import SessionEvents
+    from gateway.session_sudo_prompt import ask_viewer_result, lock_answer, resolve
+
+    class Peer:
+        frames = []
+
+        def write(self, frame):
+            self.frames.append(frame)
+            return True
+
+        def close(self):
+            pass
+    events, peer = SessionEvents(), Peer()
+    peer.frames = []
+    events.fanout.attach(peer)
+    out = {}
+    timeout = 0.5 if ending == 'timeout' else 10
+    worker = threading.Thread(target=lambda: out.update(result=ask_viewer_result(
+        events, 's1', 'clarify', {'questions': QUESTIONS}, timeout=timeout, qids=QIDS)))
+    worker.start()
+    _wait_for(lambda: any(f.get('method') == 'clarify' for f in peer.frames))
+    rid = next(f['id'] for f in peer.frames if f.get('method') == 'clarify')
+    allow = lambda *a: True  # entitlement is covered in test_clarify_lock_owner
+    assert lock_answer(rid, 'q0', 'blue', session_ok=allow) == ['q1']
+    if ending == 'lock':
+        assert lock_answer(rid, 'q1', 'large', session_ok=allow) == []
+    elif ending != 'timeout':
+        assert resolve({'jsonrpc': '2.0', 'id': rid, **ending})
+    worker.join(5)
+    return out['result']
+
+
+@pytest.mark.parametrize('case', sorted(CASES))
+def test_both_registries_settle_a_locked_batch_the_same_way(case, monkeypatch):
+    ending, expected = CASES[case]
+    assert _sidecar(monkeypatch, ending) == expected, 'sidecar'
+    assert _owner(ending) == expected, 'owner'
+
+
+def test_the_rule_itself():
+    locked = {'q0': 'blue'}
+    assert settle_batch(locked, ended='cancelled') == {'answers': {'q0': 'blue'}, 'outcome': 'cancelled'}
+    assert settle_batch(locked, {}) == {'answers': {'q0': 'blue'}, 'outcome': 'cancelled'}
+    assert settle_batch(locked, None) == {'answers': {'q0': 'blue'}, 'outcome': 'cancelled'}  # error frame
+    assert settle_batch(locked, {'answers': {}, 'outcome': 'bogus'}) == {'answers': locked, 'outcome': 'submitted'}
+    # The caller's dict is never aliased into the result.
+    assert settle_batch(locked, ended='timed_out')['answers'] is not locked

tools/clarify_batch.py

@@ -0,0 +1,31 @@
+"""The terminal result of a batch clarify, shared by every registry that holds one.
+
+A batch (``clarify`` server request with question ids) is answered per question with ``clarify.lock``
+and ends one of four ways: the last lock, a closing response frame, a timeout, or an interrupt/cancel.
+Main's TUI sidecar (``tui_gateway/server_requests.py``) and the session owner
+(``gateway/session_sudo_prompt.py``) both settle batches; each used to compute ``{answers, outcome}``
+itself and they drifted. Both call :func:`settle_batch`, so the model is told the same thing on every
+surface.
+"""
+from __future__ import annotations
+
+OUTCOMES = ("submitted", "timed_out", "cancelled")
+
+
+def settle_batch(locked: dict, frame_result=None, *, ended: str | None = None) -> dict:
+    """``{answers, outcome}`` for a batch whose locks so far are *locked*.
+
+    - ``ended`` (``"timed_out"`` / ``"cancelled"``): the wait ended with no frame (timeout, interrupt,
+      a ``cancel()``), or ``"submitted"`` for the last lock. The locked answers stand.
+    - Otherwise *frame_result* is the closing frame's ``result``. A dict with an ``answers`` dict lays
+      those answers over the locks (the frame wins per question) and keeps the frame's own outcome when it
+      names a known one (main's classic CLI reports ``timed_out``), else ``submitted``. Anything else (Ink
+      Esc / Desktop Skip send ``{}``; an error frame gives ``None``) is a cancel-all: locks kept, ``cancelled``.
+    """
+    if ended is not None:
+        return {"answers": dict(locked), "outcome": ended if ended in OUTCOMES else "cancelled"}
+    if isinstance(frame_result, dict) and isinstance(frame_result.get("answers"), dict):
+        outcome = frame_result.get("outcome")
+        return {**frame_result, "answers": {**locked, **frame_result["answers"]},
+                "outcome": outcome if outcome in OUTCOMES else "submitted"}
+    return {"answers": dict(locked), "outcome": "cancelled"}

tui_gateway/server_requests.py

@@ -43,6 +43,8 @@ import time
 import uuid
 from typing import Any, Callable
 
+from tools.clarify_batch import settle_batch
+
 logger = logging.getLogger(__name__)
 
 
@@ -221,7 +223,7 @@ def send(method: str, sid: str, params: dict, *, timeout: float | None,
     if timed_out:
         _emit_cancel(req, "timeout")
         if req.qids is not None:
-            return {"answers": locked, "outcome": "timed_out"}
+            return settle_batch(locked, ended="timed_out")
     return None
 
 
@@ -302,16 +304,11 @@ def resolve_response(frame: dict, transport: Any = None) -> bool:
         else:
             result = frame.get("result")
             req.result = result if isinstance(result, dict) else {}
-            if req.qids and "answers" in req.result:
+            if req.qids:
                 # Batch clarify: answers locked early via clarify.lock belong to the final set even when
-                # the closing response only carries the tail the user answered last.
-                answers = req.result.get("answers")
-                merged = dict(req.locked)
-                if isinstance(answers, dict):
-                    merged.update(answers)
-                req.result = {**req.result, "answers": merged, "outcome": "submitted"}
-            elif req.qids:
-                req.result = {"answers": dict(req.locked), "outcome": "cancelled"}
+                # the closing response only carries the tail the user answered last (one rule with the
+                # session owner's registry: ``tools.clarify_batch``).
+                req.result = settle_batch(req.locked, req.result)
             req.answered = True
     if req.on_result is not None:
         req.on_result(req.result)
@@ -332,7 +329,7 @@ def lock_answer(request_id: str, question_id: str, answer: str | None) -> list[s
         req.locked[question_id] = answer
         remaining = [qid for qid in req.qids if qid not in req.locked]
         if not remaining:
-            req.result, req.answered = {"answers": dict(req.locked), "outcome": "submitted"}, True
+            req.result, req.answered = settle_batch(req.locked, ended="submitted"), True
             _open.pop(request_id, None)
     if not remaining:
         req.event.set()
@@ -350,7 +347,7 @@ def cancel(sid: str | None = None, reason: str = "interrupted", *, method: str |
         for req in targets:
             _open.pop(req.id, None)
             if req.qids is not None:
-                req.result, req.answered = {"answers": dict(req.locked), "outcome": "cancelled"}, True
+                req.result, req.answered = settle_batch(req.locked, ended="cancelled"), True
             else:
                 req.result, req.answered = None, False
             if req.writing: