diff --git a/gateway/session_sudo_prompt.py b/gateway/session_sudo_prompt.py index 4596f3669b..701f2ed6f5 100644 --- a/gateway/session_sudo_prompt.py +++ b/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. diff --git a/tests/gateway/test_clarify_batch_settlement_parity.py b/tests/gateway/test_clarify_batch_settlement_parity.py new file mode 100644 index 0000000000..0175926341 --- /dev/null +++ b/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 diff --git a/tools/clarify_batch.py b/tools/clarify_batch.py new file mode 100644 index 0000000000..0cb8edf495 --- /dev/null +++ b/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"} diff --git a/tui_gateway/server_requests.py b/tui_gateway/server_requests.py index 0cc4bbf01d..a7ee8e80fe 100644 --- a/tui_gateway/server_requests.py +++ b/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: