diff --git a/gateway/session_refine.py b/gateway/session_refine.py index e2957e067a..524266b403 100644 --- a/gateway/session_refine.py +++ b/gateway/session_refine.py @@ -52,6 +52,7 @@ async def refine_local_session(runner, source, focus): return t('gateway.refine.started', focus_suffix=tail) task_id = f'refine_{uuid.uuid4().hex[:6]}' holder = [None] + loop = asyncio.get_running_loop() def publish(message): current = authority.sessions.get(ref.session_id) @@ -63,6 +64,9 @@ async def refine_local_session(runner, source, focus): review_body() finally: # whatever raised first, the session's guard is released finish_background_review_run(live.review, run) + from gateway.session_retirement import retire_after_review + with contextlib.suppress(RuntimeError): # the loop already closed + loop.call_soon_threadsafe(retire_after_review, authority, ref, live) return '' # the summary (if any) was published above; an empty review.summary draws nothing def review_body(): diff --git a/gateway/session_retirement.py b/gateway/session_retirement.py index 5e514de210..d591f68d67 100644 --- a/gateway/session_retirement.py +++ b/gateway/session_retirement.py @@ -84,6 +84,12 @@ def retire_at_idle(authority, ref, live): return False +def retire_after_review(authority, ref, live): + """A cold ``/refine`` review ended (loop side): a close it pinned retires the conversation now.""" + if authority.sessions.get(ref.session_id) is live and live.closed_stamp is not None: + retire_at_idle(authority, ref, live) + + def restore_retired(authority, session_id): """Bring back a conversation ``retire_idle`` let go, from its receipt. A local one keeps its frozen policy on the local adapter until it is deleted; an API one keeps its binding receipt. diff --git a/tests/gateway/test_local_closed_retirement.py b/tests/gateway/test_local_closed_retirement.py index a311cee10e..7ba50c8bb3 100644 --- a/tests/gateway/test_local_closed_retirement.py +++ b/tests/gateway/test_local_closed_retirement.py @@ -355,3 +355,41 @@ async def test_restored_conversation_retire_timer_firing_mid_close_keeps_the_clo release.set() db.close() + +@pytest.mark.asyncio +async def test_conversation_closed_during_a_refine_review_retires_once_the_review_ends(tmp_path, monkeypatch): + """The review pins the closed conversation only while it runs: its end retires it.""" + import threading + from contextlib import nullcontext + + from agent import background_review + from gateway import session_refine, session_tui_side + _runner, authority, db = await _owner(tmp_path, monkeypatch) + finish = threading.Event() + threads = [] + monkeypatch.setattr(session_tui_side, '_side_scope', lambda *a: nullcontext()) + monkeypatch.setattr(session_tui_side, '_side_agent', lambda *a, **k: type('A', (), { + 'valid_tool_names': set(), 'release_clients': lambda self: None})()) + monkeypatch.setattr(background_review, 'spawn_background_review_thread', + lambda *a, **k: (lambda: finish.wait(10), '')) + + def spawn(_authority, _ref, _task_id, _event_type, body, **_kw): + threads.append(threading.Thread(target=body, daemon=True)) + threads[-1].start() + monkeypatch.setattr(session_tui_side, '_spawn', spawn) + try: + ref = _create(authority) + db.append_message(ref.session_id, 'user', 'hello') + live = authority.sessions[ref.session_id] + reply = await session_refine.refine_local_session(authority.runner, live.source, None) + assert threads, reply + await _close(authority, ref) + await async_wait_until(lambda: live.closed_stamp is not None, timeout=10, what="the close to settle") + assert authority.sessions.get(ref.session_id) is live # the running review pins it + finish.set() + threads[0].join(10) + await async_wait_until(lambda: ref.session_id not in authority.sessions, timeout=10, + what="the closed conversation to retire once its review ended") + finally: + finish.set() + db.close()