diff --git a/tests/test_coordinator_idle_observer.py b/tests/test_coordinator_idle_observer.py index 4ef02c09..83eb5929 100644 --- a/tests/test_coordinator_idle_observer.py +++ b/tests/test_coordinator_idle_observer.py @@ -25,6 +25,8 @@ from turnstone.console.coordinator_idle_observer import ( CoordinatorIdleObserver, ) from turnstone.core.metacognition import ( + NUDGE_CHILD_RUNNING_LINE, + NUDGE_CHILD_STOPPED_LINE, NUDGE_CHILD_STOPPED_STATES, NUDGE_IDLE_TASKS_CHILD_SLOT, NUDGE_IDLE_TASKS_ID_SLOT, @@ -1467,9 +1469,7 @@ class TestTasksBodyChildrenFacts: _type, text, _meta = self._fire(mgr, storage, ws) - assert ( - "Child child-a has stopped — wait_for_workstream returns immediately for it." - ) in text + assert (NUDGE_CHILD_STOPPED_LINE.format(ws_id="child-a").removeprefix(chr(10))) in text # The observed state renders as itself — never the hedge, never # the other state's line. assert "Child child-a is still running" not in text @@ -1493,12 +1493,9 @@ class TestTasksBodyChildrenFacts: _type, text, _meta = self._fire(mgr, storage, ws) if state in NUDGE_CHILD_STOPPED_STATES: - assert ( - "Child child-a has stopped — wait_for_workstream returns immediately for it." - in text - ) + assert NUDGE_CHILD_STOPPED_LINE.format(ws_id="child-a").removeprefix(chr(10)) in text else: - assert "Child child-a is still running; check before redoing anything it owns." in text + assert NUDGE_CHILD_RUNNING_LINE.format(ws_id="child-a").removeprefix(chr(10)) in text @pytest.mark.parametrize("state", ["closed", "deleted"]) def test_terminal_child_rows_read_as_childless(self, coord_setup, state): @@ -1815,12 +1812,8 @@ class TestExecutableCalls: # stop and the immediate wait — so the mixed state renders as # two facts, never one hedge covering both. lines = text.split(chr(10)) - assert lines[1] == ( - "Child child-a is still running; check before redoing anything it owns." - ) - assert lines[2] == ( - "Child child-b has stopped — wait_for_workstream returns immediately for it." - ) + assert lines[1] == (NUDGE_CHILD_RUNNING_LINE.format(ws_id="child-a").removeprefix(chr(10))) + assert lines[2] == (NUDGE_CHILD_STOPPED_LINE.format(ws_id="child-b").removeprefix(chr(10))) def test_the_blocked_branch_is_a_call_not_prose(self, coord_setup): """MUTATION CONTROL: reverting the wait to prose fails here. diff --git a/tests/test_eval_nudges.py b/tests/test_eval_nudges.py index a86817d6..fdec53b9 100644 --- a/tests/test_eval_nudges.py +++ b/tests/test_eval_nudges.py @@ -21,6 +21,8 @@ import pytest from turnstone.console.coordinator_idle_observer import _ACTIVE_CHILD_STATES from turnstone.core.metacognition import ( + NUDGE_CHILD_RUNNING_LINE, + NUDGE_CHILD_STOPPED_LINE, NUDGE_IDLE_TASKS_CHILD_DOOR, NUDGE_IDLE_TASKS_CHILD_SLOT, NUDGE_IDLE_TASKS_ID_SLOT, @@ -281,6 +283,12 @@ class TestWorldSeeding: out = client.list_nodes() ids = {n.get("node_id") for n in out.get("nodes", [])} assert "node-t" in ids + # The FILTERED read is the sharp edge: filter values are + # JSON-encoded before comparison, so the seed must store them + # encoded exactly as production writers do or this matches + # nothing and the world is hollow for a filtering model. + filtered = client.list_nodes(filters={"os": "linux"}) + assert {n.get("node_id") for n in filtered.get("nodes", [])} == {"node-t"} def test_a_worldless_cell_seeds_nothing_and_raises_nothing(self, eval_storage): from turnstone.core.memory import list_structured_memories @@ -1038,6 +1046,39 @@ class TestSweepValidation: ] ) + def test_world_memory_row_missing_content_is_refused(self): + """The field refusals are the only guard in front of + ``_seed_world``'s bare indexing — a regression here becomes a + mid-sweep KeyError after the canary spends round-trips. Own + test, NOT a ``_trip_cells`` entry: that helper is zipped + ``strict=True`` against the check table.""" + import re + + cells = [ + { + "id": "X_world_bad", + "arms": [ARM_NUDGE], + "tasks": [_OPEN_TASK], + "world": {"memory": [{"name": "x"}]}, + } + ] + with pytest.raises(SystemExit, match=re.escape("world.memory[0].content")): + _validate_cells(cells) + + def test_world_node_row_with_empty_id_is_refused(self): + import re + + cells = [ + { + "id": "X_world_bad", + "arms": [ARM_NUDGE], + "tasks": [_OPEN_TASK], + "world": {"nodes": [{"node_id": ""}]}, + } + ] + with pytest.raises(SystemExit, match=re.escape("world.nodes[0].node_id")): + _validate_cells(cells) + def test_a_later_bad_cell_still_refuses(self): """The scan covers the whole list, not just its head.""" for bad in ( @@ -1233,7 +1274,7 @@ class TestStimulus: env = _envelope({"id": "tsk_1", "title": "audit auth.py", "status": "pending"}) body = render_tasks_body(env, children=[("ws-c1", "running")]) assert body.startswith("You still have 1 open task: 0 in_progress, 1 pending.") - assert "Child ws-c1 is still running; check before redoing anything it owns." in body + assert NUDGE_CHILD_RUNNING_LINE.format(ws_id="ws-c1").removeprefix(chr(10)) in body assert "needs_user" in body # The seeded id reaches the block AND the branch calls; the # seeded title reaches neither. @@ -1308,13 +1349,11 @@ class TestStimulus: return str(turns[1]["content"]) assert "Child " not in _body([]) - assert "Child ws-c1 is still running; check before redoing anything it owns." in _body( + assert NUDGE_CHILD_RUNNING_LINE.format(ws_id="ws-c1").removeprefix(chr(10)) in _body( [_LIVE_CHILD] ) idle_body = _body([{"ws_id": "ws-c1", "name": "auditor", "state": "idle"}]) - assert ( - "Child ws-c1 has stopped — wait_for_workstream returns immediately for it." - ) in idle_body + assert (NUDGE_CHILD_STOPPED_LINE.format(ws_id="ws-c1").removeprefix(chr(10))) in idle_body assert "Child " not in _body([{"ws_id": "ws-c1", "name": "auditor", "state": "closed"}]) # No hedge about an observed state, in any cell class. for children in ([], [_LIVE_CHILD], [{"ws_id": "ws-c1", "state": "idle"}]): @@ -1474,7 +1513,7 @@ class TestBodyOverrideSkip: assert rendered.startswith( "You still have 1 open task: 0 in_progress, 1 pending." + chr(10) - + "Child ws-c1 is still running; check before redoing anything it owns." + + NUDGE_CHILD_RUNNING_LINE.format(ws_id="ws-c1").removeprefix(chr(10)) ) assert "record the link and wait instead of redoing its work" in rendered assert "child_ws_id='ws-c1'" in rendered diff --git a/tests/test_idle_nudge_wake_integration.py b/tests/test_idle_nudge_wake_integration.py index 68d4d047..11e445df 100644 --- a/tests/test_idle_nudge_wake_integration.py +++ b/tests/test_idle_nudge_wake_integration.py @@ -32,7 +32,10 @@ from tests._helpers import wait_until as _wait_until from tests.test_session_manager import FakeStorage from turnstone.core import session_worker from turnstone.core.idle_nudge_watcher import IdleNudgeWatcher, wake_workstream_if_pending -from turnstone.core.metacognition import NUDGE_IDLE_TASKS_CHILD_DOOR +from turnstone.core.metacognition import ( + NUDGE_CHILD_RUNNING_LINE, + NUDGE_IDLE_TASKS_CHILD_DOOR, +) from turnstone.core.session import ChatSession from turnstone.core.session_manager import SessionManager from turnstone.core.trajectory import dicts_from_turns, turn_from_dict @@ -573,9 +576,7 @@ def test_coord_idle_with_children_and_open_tasks_delivers_both(coord_mgr, tmp_db # whose empty half is the test below. The fact line renders the # OBSERVED state (registered running) with the full id, end to # end through the real storage round-trip. - assert ( - "Child child-a is still running; check before redoing anything it owns." in tasks_text - ) + assert NUDGE_CHILD_RUNNING_LINE.format(ws_id="child-a").removeprefix(chr(10)) in tasks_text assert "may still be running" not in tasks_text # CO-DELIVERY COHERENCE, end to end: both bodies in one drain now # name the same child, from two independent storage reads. The @@ -1223,9 +1224,9 @@ def test_interjection_handoff_dispatches_exactly_one_real_send(tmp_db): """The handoff's dispatch shape, pinned at the session boundary: a queued interjection plus queued wake nudges produce ONE ``send`` call carrying the interjection text with ``from_wake`` unset — never - ``send("")`` (measured at this HEAD: that shape appends an empty - untagged user turn and delivers the interjection one assistant turn - late via the flush seam).""" + ``send("")`` (that shape appends an empty untagged user turn and + delivers the interjection one assistant turn late via the flush + seam).""" from tests._helpers import make_chat_session session = make_chat_session() @@ -1252,3 +1253,156 @@ def test_interjection_handoff_dispatches_exactly_one_real_send(tmp_db): assert len(session._nudge_queue) == 0 assert session._queued_messages == {} assert session._wake_source_tag == "" + + +def test_interjection_handoff_contains_generation_cancelled(tmp_db): + """A Stop landing inside the handed-off interjection send must not + escape the wake worker: ``GenerationCancelled`` is a BaseException + ``session_worker`` does not catch. Deliberately NO restore on this + arm — the Stop supersedes the queued words — so the queue stays + empty and the wake entries stay dropped.""" + from tests._helpers import make_chat_session + from turnstone.core.session import GenerationCancelled + + session = make_chat_session() + session._nudge_queue.enqueue("idle_tasks", "open tasks remain", "wake") + session._nudge_queue.enqueue("idle_children", "children active", "wake") + session.queue_message("urgent question") + + def _cancelled_send(*a: Any, **k: Any) -> None: + raise GenerationCancelled() + + session.send = _cancelled_send # type: ignore[method-assign] + session.deliver_wake_nudge_from_queue() # must not raise + + assert len(session._nudge_queue) == 0 + assert session._queued_messages == {} + + +def test_interjection_handoff_restores_the_queue_when_send_raises(tmp_db): + """Any escape other than a cancel happened in send's preamble, + before the user turn was appended — the popped items must be + restored (ids and priorities intact) and the failure must surface, + so the next send's flush seams deliver the user's words instead of + a log line being their gravestone.""" + from tests._helpers import make_chat_session + + session = make_chat_session() + session._nudge_queue.enqueue("idle_tasks", "open tasks remain", "wake") + _c, _p, msg_id = session.queue_message("!!!do not lose this") + + def _exploding_send(*a: Any, **k: Any) -> None: + raise RuntimeError("preamble failure") + + session.send = _exploding_send # type: ignore[method-assign] + with pytest.raises(RuntimeError, match="preamble failure"): + session.deliver_wake_nudge_from_queue() + + # Restored verbatim: same id, same cleaned text, same priority. + assert msg_id in session._queued_messages + text, priority = session._queued_messages[msg_id] + assert text == "do not lose this" + assert priority == "important" + + +def test_interjection_handoff_does_not_restore_after_a_late_raise(tmp_db): + """send can raise from its LATE handlers, after the user turn was + appended and persisted — restoring there would deliver the user's + words twice at the next flush seam. The length snapshot is the + discriminator: an appended turn means no restore.""" + from tests._helpers import make_chat_session + + session = make_chat_session() + session._nudge_queue.enqueue("idle_tasks", "open tasks remain", "wake") + session.queue_message("delivered then failed") + + def _append_then_raise(text: str, *a: Any, **k: Any) -> None: + session.messages.append({"role": "user", "content": text}) + raise RuntimeError("late failure") + + session.send = _append_then_raise # type: ignore[method-assign] + with pytest.raises(RuntimeError, match="late failure"): + session.deliver_wake_nudge_from_queue() + + # The turn reached history; the queue must NOT get it back. + assert session._queued_messages == {} + + +def test_a_retraction_during_the_handoff_is_honoured_by_the_restore(tmp_db): + """The DELETE route can land while the dispatcher holds the popped + items — it finds the id gone and answers as already-sent. A + failure-path restore must not resurrect the message the user just + cancelled.""" + from tests._helpers import make_chat_session + + session = make_chat_session() + session._nudge_queue.enqueue("idle_tasks", "open tasks remain", "wake") + _c, _p, keep_id = session.queue_message("keep this one") + _c, _p, retract_id = session.queue_message("cancel this one") + + def _retract_mid_send(text: str, *a: Any, **k: Any) -> None: + session.dequeue_message(retract_id) + raise RuntimeError("preamble failure") + + session.send = _retract_mid_send # type: ignore[method-assign] + with pytest.raises(RuntimeError, match="preamble failure"): + session.deliver_wake_nudge_from_queue() + + assert keep_id in session._queued_messages + assert retract_id not in session._queued_messages + + +def test_interjection_handoff_skips_the_pop_when_budget_exhausted(tmp_db): + """On the budget latch ``send`` refuses without appending a turn + unless a human approves, and a wake is unattended — the handoff + must not pop (the message stays queued for the user's next real + send) and must fall through to the wake drain so the worker's exit + convergence holds.""" + from tests._helpers import make_chat_session + + session = make_chat_session() + session._nudge_queue.enqueue("idle_tasks", "open tasks remain", "wake") + _c, _p, msg_id = session.queue_message("held message") + session._budget_exhausted = True + + sends: list[tuple[Any, ...]] = [] + + def _recording_send(*a: Any, **k: Any) -> None: + sends.append((a, k)) + + session.send = _recording_send # type: ignore[method-assign] + session.deliver_wake_nudge_from_queue() + + # No interjection dispatch; the wake drain path ran instead (the + # wake-eligible entry was drained toward the synthetic wake send). + assert msg_id in session._queued_messages + assert all(args != ("held message",) for args, _k in sends) + assert any(a == ("",) for a, _k in sends) + + +def test_interjection_handoff_falls_through_on_content_free_items(tmp_db): + """A bare priority marker ('!!!') renders as nothing deliverable: + the handoff must not spend the seam on a content-free user turn — + the husk is discarded and the wake proceeds normally, nudges + intact.""" + from tests._helpers import make_chat_session + + session = make_chat_session() + session._nudge_queue.enqueue("idle_tasks", "open tasks remain", "wake") + session.queue_message("!!!") + session.queue_message(" ") + + sends: list[tuple[Any, ...]] = [] + + def _recording_send(*a: Any, **k: Any) -> None: + sends.append((a, k)) + + session.send = _recording_send # type: ignore[method-assign] + session.deliver_wake_nudge_from_queue() + + # The husks were consumed, no interjection turn was dispatched, and + # the wake path ran: the ONLY send is the wake's synthetic "" one — + # a reverted husk skip would dispatch a rendered husk turn here and + # fail the equality, so this line is the live mutation control. + assert session._queued_messages == {} + assert [a for a, _k in sends] == [("",)] diff --git a/tests/test_metacognition.py b/tests/test_metacognition.py index 6187a0ab..d0b928ec 100644 --- a/tests/test_metacognition.py +++ b/tests/test_metacognition.py @@ -4,6 +4,9 @@ import pytest from turnstone.core.metacognition import ( MEMORY_NUDGE_TYPES, + NUDGE_CHILD_OVERFLOW_LINE, + NUDGE_CHILD_RUNNING_LINE, + NUDGE_CHILD_STOPPED_LINE, NUDGE_CHILD_STOPPED_STATES, NUDGE_COMPLETION, NUDGE_CORRECTION, @@ -833,7 +836,7 @@ class TestFormatIdleTasksNudge: against the same constant with NO fact lines. """ opener = "You still have 2 open tasks: 1 in_progress, 1 pending." - facts = chr(10) + "Child child-a is still running; check before redoing anything it owns." + facts = NUDGE_CHILD_RUNNING_LINE.format(ws_id="child-a") block = chr(10) * 2 + " - tsk_1 (in_progress)" + chr(10) + " - tsk_2 (pending)" childful = ( NUDGE_IDLE_TASKS_TAIL.replace(NUDGE_IDLE_TASKS_OPEN_LIST_SLOT, block, 1) @@ -903,9 +906,7 @@ class TestFormatIdleTasksNudge: OBSERVED fact line, full id: the caller read the state this same event, so the body states it rather than hedging about it.""" out = self._fmt(children=[("child-a", "running")]) - assert ( - chr(10) + "Child child-a is still running; check before redoing anything it owns." - ) in out + assert (NUDGE_CHILD_RUNNING_LINE.format(ws_id="child-a")) in out @pytest.mark.parametrize("state", ["thinking", "running", "attention"]) def test_every_non_stopped_live_state_reads_as_still_running(self, state): @@ -925,20 +926,67 @@ class TestFormatIdleTasksNudge: finished" disjunct, now attached to the child it is true of, and the check it invites finds whatever is actually there.""" out = self._fmt(children=[("child-a", state)]) - assert ( - chr(10) + "Child child-a has stopped — wait_for_workstream returns immediately for it." - ) in out + assert (NUDGE_CHILD_STOPPED_LINE.format(ws_id="child-a")) in out assert "is still running" not in out.split(chr(10))[1] + def test_fact_lines_cap_at_the_display_cap_with_a_counts_overflow(self): + """The body is a persistent system turn replayed every request, + so the fact block is bounded exactly as the roster body is: + display-capped lines plus one counts-only overflow line (no + ids — an id-less summary cannot dangle an unusable handle). + The wait slot keeps its own larger handle cap.""" + n = NUDGE_IDLE_CHILDREN_DISPLAY_CAP + 3 + children = [(f"{i:032x}", "running") for i in range(n)] + out = self._fmt(children=children) + lines = out.split(chr(10)) + fact_lines = [ln for ln in lines if ln.startswith("Child ")] + assert len(fact_lines) == NUDGE_IDLE_CHILDREN_DISPLAY_CAP + assert NUDGE_CHILD_OVERFLOW_LINE.format(n=3).removeprefix(chr(10)) in lines + # No id from the overflowed rows appears anywhere in the body's + # fact block (the wait call may still carry them — its cap is + # larger by design). + overflowed = {f"{i:032x}" for i in range(NUDGE_IDLE_CHILDREN_DISPLAY_CAP, n)} + for ws_id in overflowed: + assert all(not ln.startswith(f"Child {ws_id}") for ln in lines) + + def test_all_unusable_child_ids_render_the_childless_body(self): + """The door and the fact lines key on ONE condition — the + USABLE list. A children list whose every id the sanitiser + rejects must render the childless body: keeping the branch + would ship the raw template slot into a system turn with zero + fact lines above it.""" + out = self._fmt(children=[("bad", "running"), ("", "idle")]) + assert "Child " not in out + assert "waiting on a child workstream" not in out + assert NUDGE_IDLE_TASKS_CHILD_SLOT not in out + + def test_mixed_state_overflow_makes_no_state_claim(self): + """The overflow line summarises rows the fact lines above may + have just called stopped — any state adjective would + reclassify them, so the line carries a count and nothing + else.""" + n = NUDGE_IDLE_CHILDREN_DISPLAY_CAP + 2 + children = [(f"{i:032x}", "idle" if i % 2 else "running") for i in range(n)] + out = self._fmt(children=children) + overflow = NUDGE_CHILD_OVERFLOW_LINE.format(n=2).removeprefix(chr(10)) + assert overflow in out.split(chr(10)) + for word in ("live", "running", "active"): + assert word not in overflow + + def test_an_alterable_child_ws_id_is_dropped_never_mangled(self): + """The children projection takes the same alteration check as + the open-row fields: a ws_id the sanitiser would change is + dropped whole, so no forged or mangled handle can reach the + fact lines or the door slots.""" + out = self._fmt(children=[("bad", "running"), ("child-ok", "running")]) + assert "bad" not in out + assert NUDGE_CHILD_RUNNING_LINE.format(ws_id="child-ok") in out + def test_mixed_children_render_one_fact_line_each_in_read_order(self): out = self._fmt(children=[("child-a", "running"), ("child-b", "idle")]) lines = out.split(chr(10)) - assert lines[1] == ( - "Child child-a is still running; check before redoing anything it owns." - ) - assert lines[2] == ( - "Child child-b has stopped — wait_for_workstream returns immediately for it." - ) + assert lines[1] == (NUDGE_CHILD_RUNNING_LINE.format(ws_id="child-a").removeprefix(chr(10))) + assert lines[2] == (NUDGE_CHILD_STOPPED_LINE.format(ws_id="child-b").removeprefix(chr(10))) def test_no_hedge_survives_about_an_observed_state(self): """MUTATION CONTROL for the retired caveat (ruled 2026-07-29): @@ -991,9 +1039,7 @@ class TestFormatIdleTasksNudge: .replace(NUDGE_IDLE_TASKS_WAIT_SLOT, wait_call(["child-a"]), 1) .replace(NUDGE_IDLE_TASKS_CHILD_SLOT, "child-a") ) - fact_line = ( - chr(10) + "Child child-a is still running; check before redoing anything it owns." - ) + fact_line = NUDGE_CHILD_RUNNING_LINE.format(ws_id="child-a") childful = self._fmt(children=[("child-a", "running")]) assert rendered_door in childful assert dropped == childful.replace(fact_line, "", 1).replace(rendered_door, "", 1) diff --git a/turnstone/console/coordinator_idle_observer.py b/turnstone/console/coordinator_idle_observer.py index 3e6eb84c..4c898cbe 100644 --- a/turnstone/console/coordinator_idle_observer.py +++ b/turnstone/console/coordinator_idle_observer.py @@ -31,10 +31,11 @@ nothing about children at all. Either way a consistent pair is two true statements, not a contradiction. One ordering caveat, accepted: a cross-bracket pair (an older queued ``idle_children`` surviving into a bracket that enqueues ``idle_tasks``) delivers children-first by -seq; both entries are still predicate-valid, so that is a tuning miss, -not a correctness one. If evals show small models fumbling even the ordered pair, the -named upgrade path is a single combined checkpoint type selected at -produce time — do NOT reintroduce a cross-domain fire gate. +seq; both entries are still predicate-valid, so that is a tuning +miss, not a correctness one. If small models measurably fumble even +the ordered pair, the named upgrade path is a single combined +checkpoint type selected at produce time — do NOT reintroduce a +cross-domain fire gate. The class decides three behaviours, each ruled at its site: diff --git a/turnstone/core/metacognition.py b/turnstone/core/metacognition.py index a362876c..7b6d0952 100644 --- a/turnstone/core/metacognition.py +++ b/turnstone/core/metacognition.py @@ -384,18 +384,18 @@ NUDGE_IDLE_CHILDREN_HEADER = "These child workstreams are still active:" # reported running (check before redoing what it owns), a stopped # one is reported stopped (``wait_for_workstream`` returns # immediately for it) — the protective points the hedge carried, -# each now attached to the child it is true of, and nothing about -# results, whose existence no read observed. Deleting either children-bearing element -# from the CHILDREN-PRESENT body reopens the -# resume-over-live-children hazard the old cross-domain fire gate -# existed for. On a coordinator with no children both are measured -# noise (children prose on a childless coordinator measurably -# induces a pointless ``list_workstreams`` round-trip) and both -# are omitted — an -# omission asserts nothing about children at all. A body with -# NOTHING about children in it is also what keeps the children -# question a clean single factor when the body is measured: does -# it need to mention children at all? +# each now attached to the child it is true of, and nothing +# about results, whose existence no read observed. Deleting +# either children-bearing element from the CHILDREN-PRESENT body +# reopens the resume-over-live-children hazard the old +# cross-domain fire gate existed for. On a coordinator with no +# children both are measured noise (children prose on a childless +# coordinator measurably induces a pointless ``list_workstreams`` +# round-trip) and both are omitted — an omission asserts nothing +# about children at all. A body with NOTHING about children in +# it is also what keeps the children question a clean single +# factor when the body is measured: does it need to mention +# children at all? # THE SLOTS. Each is a literal that the shipped tail contains and # :func:`format_idle_tasks_nudge` substitutes at render time, exactly as @@ -422,11 +422,12 @@ NUDGE_IDLE_CHILDREN_HEADER = "These child workstreams are still active:" # honest recovery — the harness holds no id and says so. # * :data:`NUDGE_IDLE_TASKS_CHILD_SLOT` is a pure TEMPLATE VARIABLE. # No production body can contain it. The branch it sits in renders -# only when the caller passed child rows, and it is substituted -# whenever it renders; the one state that would leave it -# unsubstituted is a live child row with an empty ``ws_id`` (not -# producible — every creation path mints through ``uuid4().hex`` or -# ``secrets.token_hex``). The INDETERMINATE-read state that once +# only when the caller passed at least one USABLE child row (id +# non-empty and unaltered by the sanitiser — the same filter that +# builds the fact lines), and it is substituted whenever it +# renders: a children list whose every id fails the filter cuts +# the branch whole, so no state can leave the slot unsubstituted +# in a rendered body. The INDETERMINATE-read state that once # rendered it is no longer even expressible: the formatter takes a # required children list, and a failed read renders no body at all # (the observer fails its whole event closed). @@ -555,6 +556,29 @@ NUDGE_CHILD_STOPPED_STATES: frozenset[str] = frozenset( {WorkstreamState.IDLE.value, WorkstreamState.ERROR.value} ) +# The per-child fact-line templates, one per state class — named +# constants like every other piece of body text in this module, so a +# reword has ONE anchor (tests assert through these, never through +# retyped literals) and the two forms cannot drift from what the +# formatter renders. ``{ws_id}`` is the only slot; each line carries +# its own leading newline, the join idiom the fact block builds with. +NUDGE_CHILD_RUNNING_LINE = ( + "\nChild {ws_id} is still running; check before redoing anything it owns." +) +NUDGE_CHILD_STOPPED_LINE = ( + "\nChild {ws_id} has stopped — wait_for_workstream returns immediately for it." +) + +# Overflow line for a fact block that hits the display cap — the same +# cap-and-summarise idiom :func:`format_idle_children_nudge` ships, +# minted as a constant for the same one-anchor reason as the lines +# above. Counts only — no ids, and NO state claim: the unshown rows +# mix running and stopped children, so any state adjective here would +# reclassify some of them against the fact lines above (an id-less +# summary also cannot dangle an unusable handle; the wait call above +# the branches already carries handles up to its own larger cap). +NUDGE_CHILD_OVERFLOW_LINE = "\n...and {n} more child workstream(s) not listed here." + # ASCII control chars + Unicode steering vectors (bidi-override, # zero-width, line/paragraph separators, BOM, tag chars). Treated @@ -706,8 +730,9 @@ def format_idle_children_nudge(children: list[dict[str, Any]]) -> str: """Render the ``idle_children`` reminder body — ids and states ONLY. *children* is a list of dicts carrying at least ``ws_id`` and - ``state`` as strings — both producers' row projections satisfy it; every OTHER key a row carries - (``name`` above all) is ignored, which is why the annotation is + ``state`` as strings — both producers' row projections satisfy it; + every OTHER key a row carries (``name`` above all) is ignored, + which is why the annotation is ``dict[str, Any]`` rather than a narrower shape that only the projections happen to hold. Returns raw text *without* any envelope; the nudge is emitted as a first-class @@ -998,13 +1023,29 @@ def format_idle_tasks_nudge( return "" tail = NUDGE_IDLE_TASKS_TAIL - # ONE condition, BOTH children-bearing elements. The per-child fact - # lines (built below, beside the opener) and the blocked-on-a-child - # branch are removed together or kept together, because a body that - # omits the facts about children while keeping the instruction about - # children is the defect this conditional exists to fix, one block - # lower down. The cut is literal-anchored, never positional. - if not children: + # A falsy ws_id would render ``child_ws_id=''`` and a fact line + # with no handle — populated in appearance and unrunnable in fact, + # which is the shape this whole feature exists to remove — and an + # id the sanitiser would ALTER is dropped for the same + # drop-never-mangle rule the open-row fields take below: this is a + # public surface, and a ws_id carrying a newline or bracket would + # forge sibling fact lines inside a system turn. Neither is + # producible today (every creation path mints through + # ``uuid4().hex`` or ``secrets.token_hex``), so both are guards + # against a future writer rather than live branches. + live = [(ws_id, state) for ws_id, state in children if ws_id and sanitize_name(ws_id) == ws_id] + + # ONE condition, EVERY children-bearing element. The per-child + # fact lines (built below, beside the opener), the + # blocked-on-a-child branch, and that branch's slots all key on the + # USABLE list: a body that omits the facts about children while + # keeping the instruction about children is the defect this + # conditional exists to fix, and a children list whose every id the + # sanitiser rejects is childless FOR RENDERING purposes — keeping + # the branch would ship the raw template slot into a system turn + # with zero fact lines above it, exactly that defect in its worst + # dress. The cut is literal-anchored, never positional. + if not live: tail = tail.replace(NUDGE_IDLE_TASKS_CHILD_DOOR, "", 1) # BOTH row fields take the alteration check — see the docstring's @@ -1027,17 +1068,6 @@ def format_idle_tasks_nudge( else: tail = tail.replace(NUDGE_IDLE_TASKS_OPEN_LIST_SLOT, "", 1) - # A falsy ws_id would render ``child_ws_id=''`` and a fact line with - # no handle — populated in appearance and unrunnable in fact, which - # is the shape this whole feature exists to remove. Not producible - # today (every creation path mints through ``uuid4().hex`` or - # ``secrets.token_hex``), so this is a guard against a future writer - # rather than a live branch; it is one comprehension and it keeps - # "rendered ⇒ actionable" true by construction rather than by an - # argument about id minting three modules away. Filtered for - # RENDERING only: the branch cut above read the raw list, so a child - # row that EXISTS without a usable id still keeps the branch. - live = [(ws_id, state) for ws_id, state in children if ws_id] if live: # The wait call FIRST: its slot literal contains the child slot, # so substituting the scalar first would consume the bytes this @@ -1054,17 +1084,27 @@ def format_idle_tasks_nudge( # harness-rendered, the tail carries the typed branches. Each line # states the observed fact (the caller's read, this same event) and # pairs it with the protection that is true FOR THAT STATE — no - # "may" about a state the read returned, no invented context. The - # one "may" that remains is honest: whether a stopped child's - # results were collected is the thing the read did not observe. + # "may" about a state the read returned, no invented context, and + # nothing about results: the stopped line asserts the stop and the + # immediate wait, because whether the child produced anything is a + # thing no read observed and the cheap check finds whatever is + # there. CAPPED at the roster formatter's display cap with a + # counts-only overflow line — the body is a persistent system turn + # replayed on every request, and an uncapped block would grow with + # every finished-but-unclosed child a coordinator accumulates; the + # wait slot above keeps its own larger cap (display-capped prose, + # handle-capped call, the split the roster body already ships). + shown = live[:NUDGE_IDLE_CHILDREN_DISPLAY_CAP] facts = "".join( ( - f"\nChild {ws_id} has stopped — wait_for_workstream returns immediately for it." + NUDGE_CHILD_STOPPED_LINE if state in NUDGE_CHILD_STOPPED_STATES - else f"\nChild {ws_id} is still running; check before redoing anything it owns." - ) - for ws_id, state in live + else NUDGE_CHILD_RUNNING_LINE + ).format(ws_id=ws_id) + for ws_id, state in shown ) + if len(live) > len(shown): + facts += NUDGE_CHILD_OVERFLOW_LINE.format(n=len(live) - len(shown)) split = ", ".join(f"{open_counts[status]} {status}" for status in sorted(open_counts)) noun = "task" if total == 1 else "tasks" diff --git a/turnstone/core/session.py b/turnstone/core/session.py index c91afc18..a3252d6b 100644 --- a/turnstone/core/session.py +++ b/turnstone/core/session.py @@ -1762,6 +1762,13 @@ class ChatSession: # Queued user turns never carry attachments — see # ``AttachmentsNotQueueableError`` for the role-ordering reason — # so the entry tuple is just ``(cleaned, priority)``. + # Ids retracted while a dispatcher held the popped items — the + # DELETE route can land during the handoff's in-flight send, find + # the id already popped, and answer "already sent"; a restore + # that resurrected it would deliver a message the user explicitly + # cancelled. Written under ``_queued_lock``; cleared at each pop + # (a new window) and consumed by the restore. + self._retracted_while_popped: set[str] = set() self._queued_messages: collections.OrderedDict[str, tuple[str, str]] = ( collections.OrderedDict() ) @@ -9779,9 +9786,20 @@ class ChatSession: return cleaned, priority, msg_id def dequeue_message(self, msg_id: str) -> bool: - """Remove a queued message by ID. Returns True if removed.""" + """Remove a queued message by ID. Returns True if removed. + + A miss is RECORDED, not just reported: during the wake handoff's + in-flight send the items live in the dispatcher's hands, so the + user's retraction cannot reach the queue — the ledger lets a + failure-path restore honour it instead of resurrecting a message + the user cancelled. (A miss for an id that was simply already + delivered records harmlessly: the ledger is cleared at each pop + and consulted only by the restore.) + """ with self._queued_lock: popped = self._queued_messages.pop(msg_id, None) + if popped is None: + self._retracted_while_popped.add(msg_id) return popped is not None def flush_queued_messages(self) -> bool: @@ -9881,27 +9899,70 @@ class ChatSession: self._append_user_turn(content, ()) return True - def _pop_queued_messages_text(self) -> str: - """Atomically drain ``_queued_messages`` and render the combined text. + def _pop_queued_messages(self) -> dict[str, tuple[str, str]]: + """Atomically drain ``_queued_messages``, returning the raw items. - The one rendering of the interjection queue as USER-turn content - — ``[IMPORTANT]``-prefixed per item priority, items joined by - blank lines — shared by :meth:`_flush_queued_messages` (the - in-send flush seams) and the wake path's interjection handoff in - :meth:`deliver_wake_nudge_from_queue`, so the two dispatch shapes - cannot drift. Returns ``""`` when nothing was queued. The pop - happens under ``_queued_lock`` and is destructive: the caller - owns delivery of whatever comes back. + The pop happens under ``_queued_lock`` and is destructive: the + caller owns delivery of whatever comes back, and a caller whose + delivery can fail restores the SAME mapping via + :meth:`_restore_queued_messages` — ids and priorities intact, so + the queued-id / send-id correspondence the delete route and the + pending rows rely on survives a failed dispatch. + """ + with self._queued_lock: + items = dict(self._queued_messages) + self._queued_messages.clear() + self._retracted_while_popped.clear() + return items + + def _restore_queued_messages(self, items: dict[str, tuple[str, str]]) -> None: + """Put popped items back at the FRONT of ``_queued_messages``. + + The undo half of :meth:`_pop_queued_messages`, for a dispatcher + whose delivery failed before any turn was appended. Restores + unconditionally — the items were already admitted, so + ``_QUEUE_MAX`` (an admission gate, not a storage invariant) does + not re-apply — and ahead of anything queued meanwhile, keeping + arrival order. + """ + with self._queued_lock: + merged = { + mid: row for mid, row in items.items() if mid not in self._retracted_while_popped + } + merged.update(self._queued_messages) + self._queued_messages.clear() + self._queued_messages.update(merged) + self._retracted_while_popped.clear() + + @staticmethod + def _render_queued_messages(items: dict[str, tuple[str, str]]) -> str: + """The one rendering of popped interjection items as USER-turn + content — ``[IMPORTANT]``-prefixed per item priority, items + joined by blank lines — shared by :meth:`_flush_queued_messages` + (the in-send flush seams) and the wake path's interjection + handoff in :meth:`deliver_wake_nudge_from_queue`, so the two + dispatch shapes cannot drift. + + Content-free items are SKIPPED, mirroring ``_collect_advisories``: + a priority prefix with nothing behind it (a bare ``!!!``) must + not become a turn — rendered alone it would read as a truthy + ``"[IMPORTANT] "`` and buy a content-free user turn. Returns + ``""`` when nothing renderable was queued. """ from turnstone.core.tool_advisory import PRIORITY_IMPORTANT - with self._queued_lock: - items = list(self._queued_messages.values()) - self._queued_messages.clear() return "\n\n".join( - f"[IMPORTANT] {msg}" if pri == PRIORITY_IMPORTANT else msg for msg, pri in items + f"[IMPORTANT] {msg}" if pri == PRIORITY_IMPORTANT else msg + for msg, pri in items.values() + if msg.strip() ) + def _pop_queued_messages_text(self) -> str: + """Pop-and-render in one step, for the flush seams whose delivery + cannot fail between pop and append (:meth:`_flush_queued_messages` + appends the turn immediately).""" + return self._render_queued_messages(self._pop_queued_messages()) + def _collect_advisories( self, assessment: OutputAssessment | None, @@ -12097,11 +12158,11 @@ class ChatSession: intervention is required for the underlying failure anyway. A queued user interjection OWNS the idle seam: when - ``_queued_messages`` is non-empty at wake time, - the synthetic wake turn yields to it — the wake-only idle nudges - are dropped and the interjection runs as a genuine user send - instead. See the branch below for the mechanics and for why the - check lives HERE rather than at the watcher's gate. + ``_queued_messages`` is non-empty at wake time, the synthetic + wake turn yields to it — the wake-only idle nudges are dropped + and the interjection runs as a genuine user send instead. See + the branch below for the mechanics and for why the check lives + HERE rather than at the watcher's gate. """ # INTERJECTION-OWNS-THE-SEAM. A non-empty ``_queued_messages`` # means a user message is waiting for this exact slot: the model @@ -12126,13 +12187,13 @@ class ChatSession: # after the user turn — better placed than on a synthetic # empty one. # * The dispatch is ``send(text)``, never ``send("")`` + flush: - # measured at this HEAD, ``send("")`` with a queued - # interjection appends an EMPTY untagged user turn, answers - # it, and only then flushes the queued text as a second user - # turn — one assistant turn late, with a bogus empty user row - # in history. Popping first and sending the popped text runs - # the same worker-thread path a real send takes and yields - # exactly one user turn carrying the interjection. + # ``send("")`` with a queued interjection appends an EMPTY + # untagged user turn, answers it, and only then flushes the + # queued text as a second user turn — one assistant turn late, + # with a bogus empty user row in history. Popping first and + # sending the popped text runs the same worker-thread path a + # real send takes and yields exactly one user turn carrying + # the interjection. # # The check lives here — not at ``wake_workstream_if_pending`` — # because this method runs ON the worker thread that owns the @@ -12143,22 +12204,75 @@ class ChatSession: # idle). A message queued AFTER this pop lands mid-send and # rides the send's ordinary flush seams — the pre-existing # mid-turn interjection behaviour of every send. - interjection = self._pop_queued_messages_text() - if interjection: - dropped = self._nudge_queue.clear_channels({WAKE_CHANNEL}) - log.info( - "wake_nudge.interjection_owns_seam ws=%s dropped_wake_nudges=%d", - self._ws_id[:8], - dropped, - ) - try: - self.send(interjection) - except GenerationCancelled: - # Same containment as the wake send below: this method IS - # the wake worker's run() closure and ``session_worker`` - # catches only ``Exception``. - log.info("wake_nudge.interjection_cancelled ws=%s", self._ws_id[:8]) - return + # + # ``_budget_exhausted`` is checked BEFORE the pop: on that latch + # ``send`` refuses without appending a turn unless a human + # approves the override, and a wake is unattended — popping + # first would destroy the message on a refusal that raises + # nothing. Skipping the branch (no pop) leaves the interjection + # queued for the user's next real send, where the approval + # prompt has someone in front of it, and falls through to the + # wake drain so this worker's exit keeps its convergence. + if not self._budget_exhausted: + popped = self._pop_queued_messages() + interjection = self._render_queued_messages(popped) + if interjection: + dropped = self._nudge_queue.clear_channels({WAKE_CHANNEL}) + log.info( + "wake_nudge.interjection_owns_seam ws=%s dropped_wake_nudges=%d", + self._ws_id[:8], + dropped, + ) + appended_before = len(self.messages) + try: + self.send(interjection) + except GenerationCancelled: + # Same containment as the wake send below: this + # method IS the wake worker's run() closure and + # ``session_worker`` catches only ``Exception``. + # Deliberately NO restore on this arm: a cancel here + # is the operator's Stop, and re-arming their + # pre-Stop words for a later seam would deliver a + # message the Stop may have been meant to overtake. + log.info("wake_nudge.interjection_cancelled ws=%s", self._ws_id[:8]) + except BaseException: + # A non-cancel escape can happen on either side of + # the user-turn append: send's PREAMBLE can raise + # before anything reached history (the popped items + # would be destroyed with only a log line), but + # send's LATE handlers re-raise after the turn was + # appended AND persisted — restoring there would + # deliver the user's words twice at the next flush + # seam. The length snapshot is the discriminator: + # restore only when nothing was appended. (A + # mid-send compaction rewrites the list and could + # coincidentally match the old length; compaction + # implies the turn appended, so that corner + # restores a duplicate — accepted as vanishingly + # rare against the common preamble-loss case.) + # + # The wake entries stay dropped on this arm — caps + # charged, nothing delivered, the accepted + # fail-closed drop — and re-queueing them would + # re-arm the worker-exit wake retry into a hot + # loop against a persistently failing send. The + # restored interjection's re-arm is the user's next + # send (its flush seams deliver the restored + # items); a worker-exit re-check of the interjection + # queue would close that window structurally. + if len(self.messages) == appended_before: + self._restore_queued_messages(popped) + raise + return + if popped: + # Everything queued was content-free (a bare priority + # marker) — nothing to dispatch, nothing worth a turn. + # Fall through to the normal wake drain below. + log.info( + "wake_nudge.interjection_empty ws=%s discarded=%d", + self._ws_id[:8], + len(popped), + ) # Two-pass drain: wake-eligible channels first. ``"quiet"`` entries # (external events demoted by a user cancel) ride a wake earned by diff --git a/turnstone/eval/nudges.py b/turnstone/eval/nudges.py index 31c38f12..bff61d69 100644 --- a/turnstone/eval/nudges.py +++ b/turnstone/eval/nudges.py @@ -58,7 +58,7 @@ import hashlib import json import os import time -from typing import TYPE_CHECKING, Any, cast +from typing import TYPE_CHECKING, Any, ClassVar, cast from openai import OpenAI @@ -347,6 +347,15 @@ class _StubCoordinatorClient(CoordinatorClient): fixture's own, never invented. """ + # A seeded node's registry heartbeat is stamped once, at world-seed + # time, while the real client's liveness window assumes an active + # cluster re-stamping every few seconds. A multi-turn run against a + # slow model can outlive the real 120s window, at which point + # ``list_nodes`` reports an empty cluster and the seeded world turns + # hollow mid-run, timing-dependently. The eval world is static by + # construction, so its nodes are live for the whole run. + _NODES_HEARTBEAT_WINDOW_S: ClassVar[int] = 10**6 + def __init__( self, storage: Any, @@ -825,7 +834,12 @@ def _seed_world(storage: Any, case: dict[str, Any]) -> None: for node in world.get("nodes", ()): node_id = node["node_id"] storage.register_service("server", node_id, node.get("url", f"http://{node_id}:8080")) - meta = [(k, str(v), "auto") for k, v in node.get("metadata", {}).items()] + # JSON-encoded exactly as every production writer stores these + # values — ``list_nodes`` re-encodes its filter values the same + # way before comparing, so a raw string here would never match a + # filtered lookup and the seeded world would stay hollow for a + # model that filters. + meta = [(k, json.dumps(v), "auto") for k, v in node.get("metadata", {}).items()] if meta: storage.set_node_metadata_bulk(node_id, meta)