diff --git a/emrg/server/daemon.py b/emrg/server/daemon.py index 638d203c..ed168922 100644 --- a/emrg/server/daemon.py +++ b/emrg/server/daemon.py @@ -2096,7 +2096,8 @@ async def _github_connect_web_cancel(self) -> None: pass async def _task_vibe_check(self, task_name: str, session_id: str, cwd: str, - prompt: str = "", completion_summary: str = "") -> dict: + prompt: str = "", completion_summary: str = "", + cycle_started_at: str = "") -> dict: """Structured LLM ask (Ask mode, no tools) about a finished task cycle. The agent must answer in strict JSON (rant 2026-08-20T10:58:55, @@ -2138,10 +2139,20 @@ async def _task_vibe_check(self, task_name: str, session_id: str, cwd: str, # Rant 2026-08-19T10:15:43: load the task's own session history by its # fixed session_id (session files are organized by cwd). Recent N # messages only — the whole session may be very long. + # + # `cycle_started_at` is the boundary this slice was missing: the session + # is reused by every run of this task, so "the last 100 messages" is + # mostly *previous* runs whenever this one was light on narrative. The + # summariser then describes the neighbour — measured 2026-09-30 on two + # consecutive `emrg-task` records whose `work` strings name the same + # merge, the second cycle's own findings absent — and `recommend_slowdown` + # rides the same call, so a mis-read run can also mis-cadence itself. + # The instant is the `task` frame's own `timestamp`, so no clock is read + # here; an absent value keeps the old behaviour (`records_since`). if session_id and cwd: try: session = Session.load(session_id, Path(cwd)) - history = session.get_messages_for_llm() + history = session.get_messages_for_llm(since=cycle_started_at) if history: # Rant 2026-08-19T19:25:56 (root cause): slicing the # validated list can orphan a leading role:"tool" message @@ -3305,9 +3316,13 @@ async def _process_message( cwd = msg.get("cwd", "") prompt = msg.get("prompt", "") summary = msg.get("completion_summary", "") + # The instant this run was dispatched (the scheduler's `cycle_time`, + # which is also the run's id). Absent from an older scheduler, in + # which case the evidence is the whole session exactly as before. + cycle_started_at = msg.get("cycle_started_at", "") try: result = await self._task_vibe_check( - task_name, session_id, cwd, prompt, summary, + task_name, session_id, cwd, prompt, summary, cycle_started_at, ) await self._send(ws, { "type": "vibe_check_result", diff --git a/emrg/server/scheduler.py b/emrg/server/scheduler.py index 75d14902..a570593d 100644 --- a/emrg/server/scheduler.py +++ b/emrg/server/scheduler.py @@ -2327,7 +2327,8 @@ def _saturation_heartbeat_active(self) -> bool: """ return self._slowdown_active - async def _request_vibe_check(self, ws, prompt: str, completion_summary: str) -> dict | None: + async def _request_vibe_check(self, ws, prompt: str, completion_summary: str, + cycle_started_at: str = "") -> dict | None: """Ask the daemon for a structured vibe check on the SAME connection. Sends ``task_vibe_check`` and waits for ``vibe_check_result``. Rant @@ -2355,6 +2356,12 @@ async def _request_vibe_check(self, ws, prompt: str, completion_summary: str) -> "task_name": self.name, "prompt": (prompt or "")[:2000], "completion_summary": (completion_summary or "")[:3000], + # The instant this cycle was dispatched — the same value the + # `task` frame carries as its `timestamp`, so the daemon can tell + # this run's records from the previous run's inside a session + # that both share. Without it the summariser reads the shared + # session and can describe the neighbour (measured 2026-09-30). + "cycle_started_at": cycle_started_at or "", }, ensure_ascii=False)) while True: try: @@ -2668,6 +2675,7 @@ async def _run_evolution_cycle(self) -> str: vibe_result = await self._request_vibe_check( ws, prompt=prompt, completion_summary=completion_content[:3000], + cycle_started_at=cycle_time.isoformat(), ) except Exception as e: self._logger.exception("TaskHandler[%s] error", self.name) diff --git a/emrg/session.py b/emrg/session.py index c728aeb6..395b31c1 100644 --- a/emrg/session.py +++ b/emrg/session.py @@ -54,6 +54,61 @@ def generate_session_id(cwd: Path) -> str: return prefix + suffix +def _parse_instant(value: object) -> datetime | None: + """An ISO-8601 instant from a stored timestamp, or `None` if unreadable. + + `None` is "could not measure", never "epoch": the caller keeps the record + when this returns `None`, because excluding evidence is the direction that + cannot be noticed. + """ + if not isinstance(value, str) or not value: + return None + try: + return datetime.fromisoformat(value) + except ValueError: + return None + + +def records_since(records: list[dict], since: str) -> list[dict]: + """The records from `since` (an ISO-8601 instant) onward — the run's own. + + A task session is **one session for every run of that task**: `emrg-evolution-` + is reused cycle after cycle, so its history holds all of them. Anything reading + "the recent history" as evidence about *this* run therefore reads the previous + run's work as if it were this one's, unless it knows where the run began. This + function is that knowledge, applied: `since` is the instant the run was dispatched, + and it is carried in the `task` frame the scheduler already sends, so no clock is + consulted and no state is kept. + + Fail-open in both directions, deliberately: + + * an empty or unparseable `since` returns the list **unchanged** — a window whose + start is unknown is the window the session always had, and guessing a boundary + would be worse than not having one; + * a record whose own timestamp is missing or unreadable is **kept** — dropping it + would delete evidence silently, and a summary that read one record too many is + the smaller error than one that cannot see a tool call at all. + + Comparing a naive timestamp with an aware one raises, and a stored timestamp is + naive local time (`datetime.now().isoformat()`), so exactly one side being aware + is treated as unmeasurable rather than as an order. + """ + if not since: + return list(records) + cutoff = _parse_instant(since) + if cutoff is None: + return list(records) + kept: list[dict] = [] + for record in records: + at = _parse_instant(record.get("timestamp")) + if at is None or (at.tzinfo is None) != (cutoff.tzinfo is None): + kept.append(record) + continue + if at >= cutoff: + kept.append(record) + return kept + + def records_to_messages(records: list[dict]) -> list[dict]: """Convert stored history records to OpenAI-compatible messages. @@ -465,7 +520,7 @@ def _read_history(self) -> list[dict]: logger.warning("corrupt line in history.jsonl, skipping") return records - def get_messages_for_llm(self) -> list[dict]: + def get_messages_for_llm(self, since: str = "") -> list[dict]: """This session's history, converted to OpenAI-compatible messages. The conversion itself is :func:`records_to_messages`, which is a function @@ -474,8 +529,13 @@ def get_messages_for_llm(self) -> list[dict]: *candidate* record list it has not written anywhere, and asking with a different conversion than the live request uses would be asking a different question than the refusal answered. + + `since` narrows the records first, through :func:`records_since`: a task + session is reused by every run of its task, so a reader that wants *this* + run's work has to say where it began. The default is the whole history, + which is what this method always returned. """ - return records_to_messages(self._read_history()) + return records_to_messages(records_since(self._read_history(), since)) # ── Compact ─────────────────────────────────────────────── diff --git a/tests/test_run_evidence_boundary.py b/tests/test_run_evidence_boundary.py new file mode 100644 index 00000000..4b89d214 --- /dev/null +++ b/tests/test_run_evidence_boundary.py @@ -0,0 +1,391 @@ +"""A run's evidence must start where the run started, not at the session's tail. + +Why this file exists +-------------------- +A scheduled task runs in **one** session for its whole life: `emrg-evolution-` +is reused by every run. So "the recent history of that session" is not evidence about +*this* run — it is a window that mostly holds the *previous* one whenever this run was +light on narrative (a vote-only run, a verification run, a run whose work is reading). + +Measured on `emrg-task.jsonl` 2026-09-30 (issue filed 2026-10-08): two consecutive +records whose `work` strings describe the **same** merge — the second record naming the +first run's work as if it had just happened, and omitting everything the second run +found (the two differ `68` vs `54` tool calls, so the second run was real work). + +`work` is also what `recommend_slowdown` rides, so a run mis-read this way can also +mis-cadence itself. And `work` is the only per-run summary a human reads, in the GUI's +task recent-runs table — which is also what a cycle reads when it reconstructs state. + +The seam under test +------------------- +The scheduler already knows when a run was dispatched — it is `cycle_time`, the very +value the `task` frame carries as its `timestamp` and its id. Nothing had to be +invented: it had to be *carried* to the summariser. Three assertions, on three layers, +so a break anywhere is visible: + +* `records_since` — the pure rule, including the two fail-open directions; +* `Session.get_messages_for_llm(since=…)` — that the rule reaches the reader, against a + session file on disk holding two runs; +* `EmrgServer._task_vibe_check` — the outcome the defect is about: what the **judge** + is given, read off the messages the LLM call actually received (a stub `llm` records + them), because asserting on the history helper alone would be one level short of the + claim. Both directions are asserted: with the boundary the previous run's work is + absent, and without it the same session still shows it — otherwise a filter that + dropped everything would pass. +* the **hand-over** itself, at the sender: every layer above can be correct while the + fix does nothing, since the instant has to travel from the scheduler to the daemon. + That is where this bug lived (`cycle_time` existed and was simply not carried), so a + guard that stopped at the daemon would leave the defect reproducible. + +`python -m pytest tests/test_run_evidence_boundary.py -q` +""" + +from __future__ import annotations + +import asyncio +import json +import logging + +import pytest + +from emrg.server.daemon import EmrgServer +from emrg.server.scheduler import TaskHandler +from emrg.session import Session, records_since + +# The salient work of the *previous* run, and of this one. Distinct vocabularies so +# "did the judge see it" is answerable by substring, the way the defect was measured. +PREVIOUS_RUN_WORK = "merged PR #1760 and cut the v0.3.6 TUI crash" +THIS_RUN_WORK = "voted on the landing tree and wrote the cycle record" + +RUN_ONE_START = "2026-10-08T09:00:04" +RUN_TWO_START = "2026-10-08T13:08:25" + + +def _record(role: str, content: str, at: str) -> dict: + return {"type": "message", "role": role, "content": content, "timestamp": at} + + +def _two_run_session(cwd) -> Session: + """One session file holding two runs, as the real task session is.""" + session = Session.create_with_id("emrg-evolution-boundary-test", cwd) + for record in ( + _record("user", "run one prompt", "2026-10-08T09:00:05"), + _record("assistant", f"I {PREVIOUS_RUN_WORK}", "2026-10-08T09:20:00"), + _record("user", "run two prompt", "2026-10-08T13:08:26"), + _record("assistant", f"I {THIS_RUN_WORK}", "2026-10-08T13:30:00"), + ): + session.append_message(dict(record)) + return session + + +def _contents(messages: list[dict]) -> str: + return "\n".join(str(m.get("content") or "") for m in messages) + + +# --- the pure rule ----------------------------------------------------------- + + +def test_records_since_keeps_only_the_run_it_names() -> None: + records = [ + _record("assistant", "older", "2026-10-08T09:20:00"), + _record("assistant", "newer", "2026-10-08T13:30:00"), + ] + kept = records_since(records, RUN_TWO_START) + assert [r["content"] for r in kept] == ["newer"] + + +def test_a_boundary_on_the_record_is_inclusive() -> None: + """The `task` frame's own timestamp is the first record's, so >= is the rule.""" + records = [_record("user", "the run's own prompt", RUN_TWO_START)] + assert records_since(records, RUN_TWO_START) == records + + +@pytest.mark.parametrize("since", ["", "not-a-timestamp", "2026-13-45T99:99:99"]) +def test_an_unusable_boundary_leaves_the_history_alone(since: str) -> None: + """A window whose start is unknown is the window the session always had. + + Guessing a boundary here would be worse than having none: it is the direction + that silently deletes evidence, and it would revert to the old behaviour only + for the cycles nobody is looking at. + """ + records = [_record("assistant", "work", "2026-10-08T09:20:00")] + assert records_since(records, since) == records + + +@pytest.mark.parametrize( + "record", + [ + {"type": "message", "role": "assistant", "content": "no timestamp at all"}, + _record("assistant", "unreadable", "not-a-timestamp"), + _record("assistant", "aware, while the boundary is naive", "2026-10-08T13:30:00+00:00"), + ], +) +def test_a_record_whose_time_cannot_be_ordered_is_kept(record: dict) -> None: + """Unmeasurable is not old. + + Dropping it would remove a tool call from the window with nothing to show for + it; a summary that read one record too many is the smaller error, and stored + timestamps are naive local time while a boundary could be either. + """ + assert records_since([record], RUN_TWO_START) == [record] + + +# --- the reader -------------------------------------------------------------- + + +def test_get_messages_for_llm_can_start_at_a_run_boundary(tmp_path) -> None: + session = _two_run_session(tmp_path) + everything = session.get_messages_for_llm() + assert PREVIOUS_RUN_WORK in _contents(everything), ( + "the fixture must actually hold the previous run, or the assertion below " + "would pass for the wrong reason" + ) + assert THIS_RUN_WORK in _contents(everything) + + this_run = session.get_messages_for_llm(since=RUN_TWO_START) + assert THIS_RUN_WORK in _contents(this_run) + assert PREVIOUS_RUN_WORK not in _contents(this_run), ( + "the previous run's work is still in the window this run's reader is given" + ) + + +def test_without_a_boundary_the_reader_is_unchanged(tmp_path) -> None: + """The default is the whole history — what this method always returned.""" + session = _two_run_session(tmp_path) + assert session.get_messages_for_llm() == session.get_messages_for_llm(since="") + + +# --- the wire: the boundary has to be *carried*, not merely supported --------- +# +# This is the half the defect actually lives in. Every test above can pass while +# the fix does nothing at all, because they call the daemon directly: the scheduler +# is the one that knows when the run started, and the daemon can only use what the +# frame delivers. A guard for an "existed but was never handed over" bug therefore +# has to assert the hand-over, at the sender and at the receiver — one layer of a +# three-layer wire is exactly what a green suite would not notice. + + +class _RecordingWs: + """The two socket calls `_request_vibe_check` makes, and nothing else.""" + + def __init__(self, reply: dict) -> None: + self.sent: list[dict] = [] + self._reply = reply + + async def send(self, payload: str) -> None: + self.sent.append(json.loads(payload)) + + async def recv(self) -> str: + return json.dumps(self._reply) + + +def _bare_handler() -> TaskHandler: + """A handler with only the attributes the sender reaches for. + + `object.__new__` skips `__init__` on purpose: the question is what this method + puts on the frame, and building a real handler would drag the whole scheduler in. + """ + handler = object.__new__(TaskHandler) + handler.name = "emrg-task" + handler._derived = {} + handler._logger = logging.getLogger("test.run_evidence_boundary") + return handler + + +def test_the_scheduler_carries_the_boundary_on_the_frame() -> None: + ws = _RecordingWs({"type": "vibe_check_result", "ok": True, "result": { + "work": THIS_RUN_WORK, "recommend_slowdown": False, "slowdown_reason": "", + }}) + handler = _bare_handler() + + asyncio.run(handler._request_vibe_check( + ws, prompt="p", completion_summary="s", cycle_started_at=RUN_TWO_START, + )) + + assert ws.sent, "the request was never sent" + frame = ws.sent[0] + assert frame["type"] == "task_vibe_check" + assert frame.get("cycle_started_at") == RUN_TWO_START, ( + "the daemon cannot slice at a boundary the scheduler did not send — this is " + "the defect: the instant existed (`cycle_time`) but never reached the frame" + ) + + +def test_an_old_caller_still_sends_a_well_formed_frame() -> None: + """The default is present-but-empty, never absent: the field is part of the frame. + + An absent key and an empty one read the same at the daemon (`msg.get(..., "")`), + but pinning the empty string is what tells a later refactor that dropping the + field is a wire change, not a tidy-up. + """ + ws = _RecordingWs({"type": "vibe_check_result", "ok": True, "result": { + "work": THIS_RUN_WORK, "recommend_slowdown": False, "slowdown_reason": "", + }}) + + asyncio.run(_bare_handler()._request_vibe_check( + ws, prompt="p", completion_summary="s", + )) + + assert ws.sent[0]["cycle_started_at"] == "" + + +# --- the hand-over, at its *production* call site ----------------------------- +# +# `test_the_scheduler_carries_the_boundary_on_the_frame` above pins one edge of the +# hand-over: given the instant, the sender puts it on the frame. The instant it is +# *given* has an edge of its own — `_run_evolution_cycle` computes `cycle_time` and +# hands it over — and nothing pinned that one. Measured while reviewing this PR +# (2026-10-08): deleting `cycle_started_at=cycle_time.isoformat()` from that single +# call site left this file, `tests/test_scheduler.py` and `tests/test_ws_e2e.py` +# green, because every arm above supplies the argument itself or starts one layer +# below. That is the defect shape exactly — a value that exists and is not carried — +# so the arm below drives the **cycle**, not the method. + + +def test_the_cycle_hands_the_run_boundary_to_the_sender(tmp_path, monkeypatch) -> None: + """The instant the cycle hands over is the one it stamped on its own `task` frame. + + Driving `_run_evolution_cycle` is what makes this an assertion about the wire + rather than about a method: the cycle is free to compute any instant it likes, + and the property that matters is that the summariser receives *that* one. So + the expected value is not written here — it is read off the `task` frame the + same cycle sent, which is the frame the daemon records the run by. + + `_make_cycle_handler` is `tests/test_scheduler.py`'s fully-scripted handler + (imported across test modules the way `tests/test_ws_e2e.py` is); it is used + for its connection stub alone, and the connection is wrapped below so the + frames this cycle sends can be read back. + """ + from emrg.server import scheduler as sched_mod + from tests.test_scheduler import _make_cycle_handler + + handler, _ = _make_cycle_handler( + tmp_path, + frames=[ + {"request_id": "r1", "content": "Done", "done": True, + "delta": False, "session_id": "s"}, + ], + ) + + sockets: list = [] + connect = sched_mod.connect_to_server + + async def capturing_connect(): + ws = await connect() + sockets.append(ws) + return ws + + monkeypatch.setattr(sched_mod, "connect_to_server", capturing_connect) + + handed: dict = {} + + async def spy(self, ws, prompt, completion_summary, cycle_started_at=""): + handed["cycle_started_at"] = cycle_started_at + return None + + monkeypatch.setattr(TaskHandler, "_request_vibe_check", spy) + + asyncio.run(handler._run_evolution_cycle()) + + assert sockets, "the cycle never connected, so its frames cannot be read back" + frames = [] + for raw in sockets[0].sent: + try: + frames.append(json.loads(raw) if isinstance(raw, str) else raw) + except (TypeError, ValueError): # a frame this arm does not read + continue + task_frames = [f for f in frames if f.get("type") == "task"] + assert task_frames, ( + "the cycle sent no `task` frame, so this arm has no anchor — the frame is " + "where the run's own instant is recorded, and the assertion below is that " + "the summariser is given the same one" + ) + stamped = task_frames[0].get("timestamp") + assert stamped, "the `task` frame carries no timestamp, so the anchor is empty" + + assert "cycle_started_at" in handed, ( + "the cycle never asked for a vibe check, so the hand-over was not exercised " + "at all — this arm would then pass without measuring anything" + ) + assert handed["cycle_started_at"] == stamped, ( + f"the cycle stamped {stamped!r} on its own `task` frame but handed " + f"{handed['cycle_started_at']!r} to the vibe check — the summariser would " + "slice the shared session at an instant the run did not start at, which is " + "the defect (an empty or wrong boundary reads the neighbour's work)" + ) + + +# --- the outcome: what the judge is handed ----------------------------------- + + +class _RecordingLLM: + """Stands in for the daemon's LLM client and keeps what it was asked about.""" + + def __init__(self) -> None: + self.messages: list[dict] | None = None + + async def chat(self, messages, tools=None): # noqa: ANN001, ANN201 - the client's shape + self.messages = messages + return {"content": json.dumps( + {"work": THIS_RUN_WORK, "recommend_slowdown": False, "slowdown_reason": ""} + )} + + +def _bare_server(llm: _RecordingLLM) -> EmrgServer: + """An `EmrgServer` with only the attribute `_task_vibe_check` reaches for. + + `object.__new__` skips `__init__` deliberately: this test is about the evidence + the summariser is given, and standing up a real server would test the server. + """ + server = object.__new__(EmrgServer) + server.llm = llm + return server + + +def test_the_judge_is_not_shown_the_previous_runs_work(tmp_path) -> None: + """The defect itself: a `work` string about the neighbour must be unproducible. + + Measured at the seam that matters — the messages the LLM call received — rather + than at the helper that built them. Driven with `asyncio.run`, the convention + this suite uses (`pytest-asyncio` is not installed here): a bare + `@pytest.mark.asyncio` test is collected and then *not run*, which would make + this file green while asserting nothing. + """ + _two_run_session(tmp_path) + llm = _RecordingLLM() + server = _bare_server(llm) + + asyncio.run(server._task_vibe_check( + "emrg-task", "emrg-evolution-boundary-test", str(tmp_path), + prompt="run two prompt", completion_summary="done", + cycle_started_at=RUN_TWO_START, + )) + + assert llm.messages is not None + seen = _contents(llm.messages) + assert THIS_RUN_WORK in seen + assert PREVIOUS_RUN_WORK not in seen, ( + "the judge was given the previous run's work — this is the defect" + ) + + +def test_the_same_session_still_shows_both_runs_without_a_boundary(tmp_path) -> None: + """The other direction: the filter narrows, it does not empty. + + Without this arm, a `_task_vibe_check` that passed a boundary of *now* — or an + empty history — would satisfy the test above while summarising nothing. + """ + _two_run_session(tmp_path) + llm = _RecordingLLM() + server = _bare_server(llm) + + asyncio.run(server._task_vibe_check( + "emrg-task", "emrg-evolution-boundary-test", str(tmp_path), + prompt="run two prompt", completion_summary="done", + )) + + assert llm.messages is not None + seen = _contents(llm.messages) + assert PREVIOUS_RUN_WORK in seen and THIS_RUN_WORK in seen, ( + "an absent boundary must keep the behaviour this session always had" + ) diff --git a/tests/test_ws_e2e.py b/tests/test_ws_e2e.py index e883dd70..05592d62 100644 --- a/tests/test_ws_e2e.py +++ b/tests/test_ws_e2e.py @@ -332,6 +332,78 @@ async def fake_chat(messages, tools=None): await cleanup() asyncio.run(_test()) + def test_vibe_check_frame_carries_the_run_boundary_into_the_evidence(self): + """`cycle_started_at` on the frame must reach the evidence the judge sees. + + A task session is reused by every run of its task, so the daemon's + "recent history" slice is mostly the *previous* run whenever this one was + light on narrative — measured 2026-09-30 on two consecutive `emrg-task` + records whose `work` strings describe the same merge. The scheduler fixes + that by sending the instant the run was dispatched; this asserts the + receiving half at the real wire, because the two halves of a hand-over bug + fail independently — the sender can be correct while the handler drops the + field, and every unit test of the summariser still passes. + """ + async def _test(): + with tempfile.TemporaryDirectory() as tmp: + tmp = Path(tmp) + sess_dir = tmp / ".emrg" / "sessions" / "emrg-evolution-two-runs-task" + sess_dir.mkdir(parents=True, exist_ok=True) + (sess_dir / "history.jsonl").write_text( + json.dumps({"type": "message", "role": "user", + "content": "previous run prompt", + "timestamp": "2026-10-08T09:00:05"}) + "\n" + + json.dumps({"type": "message", "role": "assistant", + "content": "the previous run merged PR #1760", + "timestamp": "2026-10-08T09:20:00"}) + "\n" + + json.dumps({"type": "message", "role": "user", + "content": "this run prompt", + "timestamp": "2026-10-08T13:08:26"}) + "\n" + + json.dumps({"type": "message", "role": "assistant", + "content": "this run voted and wrote the record", + "timestamp": "2026-10-08T13:30:00"}) + "\n", + encoding="utf-8", + ) + + server, _, cleanup = await _boot_server(tmp) + try: + seen = {} + + async def fake_chat(messages, tools=None): + seen["messages"] = messages + return {"content": '{"work": "voted", ' + '"recommend_slowdown": false, "slowdown_reason": ""}'} + server.llm.chat = fake_chat + + ws = await connect_to_server() + try: + await ws.send(json.dumps({ + "type": "task_vibe_check", + "session_id": "emrg-evolution-two-runs-task", + "cwd": str(tmp), + "task_name": "two-runs-task", + "prompt": "run cycle", + "completion_summary": "auxiliary", + "cycle_started_at": "2026-10-08T13:08:25", + }, ensure_ascii=False)) + frame = await asyncio.wait_for(ws.recv(), timeout=10) + assert json.loads(frame).get("ok") is True + + contents = "\n".join( + str(m.get("content", "")) for m in seen.get("messages", []) + ) + assert "this run voted" in contents, contents + assert "the previous run merged" not in contents, ( + "the frame's cycle_started_at did not reach the evidence — " + "the handler dropped the boundary and the judge read the " + "previous run's work as this run's" + ) + finally: + await ws.close() + finally: + await cleanup() + asyncio.run(_test()) + def test_vibe_check_long_session_window_leading_tool_stripped(self): """Rant 2026-08-19T19:25:56 (root cause): slicing a validated session history to the last 100 messages can orphan a leading role:'tool'