fix(s127): complete slice 127 (a test that times out says so)

collect/3 returned the same value whether it found what it was waiting for or
gave up waiting:

    after
      remaining -> Enum.reverse(acc)

So every assertion downstream reported a missing value rather than a timeout,
and a reader goes looking for a logic defect that is not there.

Opened from main's FIPS leg failing the slice 125 merge push (run 36253217084)
while the pull_request run of the same commit passed. Not R26: no Exqlite error,
no Database busy, no write path. Reproduced locally byte for byte by shortening
one deadline from 15,000 ms to 300 ms.

This trap cost a working day earlier in the same session. Slice 126 was opened
on a test ending collect(id, &match?({:state, :idle}, &1), 15_000), where an
idle broadcast from before the message under test was already in the mailbox;
collect returned at once, every assertion ran mid-turn, and the result was a
confident report of a product defect in the approval handshake that did not
exist.

await_event/3 raises, naming the deadline and every event it did see, because
"waited 15s and saw thinking then idle" and "waited 15s and saw nothing at all"
send a reader to opposite places. 20 asserting call sites migrated; the count
came from the census rather than from reading.

The census makes the old shape unwritable: the result of collect/3 may only be
discarded. It enumerates from git ls-files, skips triple-quoted blocks and
comments (without which it reported its own explanation of itself), asserts it
enumerated more than 40 files so it cannot go blind, and was demonstrated
against a planted violation.

The deadline was sized from a measurement, not a hunch. Five instrumented runs
gave 4,745 to 4,786 ms for the fork wait; the file's other two waits return from
the mailbox in 0 to 3 ms and were left alone rather than raised on a hunch.
15,000 was 3.1x local and the leg still missed it, so 60,000. A generous
deadline costs nothing on the passing path, because await_event/3 returns the
instant the condition matches; the only thing it buys is how long a genuine hang
takes to report.

Named await_event/3 rather than await/3: Trinity.MCP.ServersUnderTest already
exports await/2, and a default argument makes await/3 export an await/2 too, so
every MCP test importing both failed to compile.

Nothing under lib/ changed.

Signed-off-by: Ayla Croft <aylacroft@proton.me>

Author: HackTuah

test/session_case_test.exs

@@ -13,7 +13,7 @@ defmodule SessionCaseTest do
   """
   use Trinity.DataCase, async: false
 
-  import Trinity.SessionCase, only: [collect: 3, await: 3]
+  import Trinity.SessionCase, only: [collect: 3, await_event: 3]
 
   alias Trinity.Factory
 
@@ -29,7 +29,7 @@ defmodule SessionCaseTest do
       send(self(), {:session, id, {:state, :idle}})
 
       message =
-        assert_raise(RuntimeError, fn -> await(id, &match?({:forked, _}, &1), 200) end)
+        assert_raise(RuntimeError, fn -> await_event(id, &match?({:forked, _}, &1), 200) end)
         |> Map.fetch!(:message)
 
       assert message =~ "200ms", "the failure does not say how long it waited: #{message}"
@@ -43,7 +43,7 @@ defmodule SessionCaseTest do
       send(self(), {:session, id, {:state, :thinking}})
       send(self(), {:session, id, {:forked, "child-1"}})
 
-      events = await(id, &match?({:forked, _}, &1), 1_000)
+      events = await_event(id, &match?({:forked, _}, &1), 1_000)
 
       assert {:forked, "child-1"} = Enum.find(events, &match?({:forked, _}, &1))
       assert {:state, :thinking} in events
@@ -61,6 +61,67 @@ defmodule SessionCaseTest do
     end
   end
 
+  describe "AC3: the shape that caused this cannot be written again" do
+    test "every call to collect/3 discards its result; anything asserting on one uses await/3" do
+      {out, 0} = System.cmd("git", ["ls-files", "test/**/*.exs", "test/*.exs"])
+      files = String.split(out, "\n", trim: true)
+
+      assert length(files) > 40,
+             "only #{length(files)} test files enumerated; the census is blind"
+
+      # Only files that take the helper from Trinity.SessionCase. `llm_test.exs` defines a
+      # `collect/1` of its own that has nothing to do with sessions.
+      users =
+        for f <- files,
+            src = File.read!(f),
+            src =~ "use Trinity.SessionCase" or src =~ "import Trinity.SessionCase",
+            not (src =~ ~r/defp? collect\(/),
+            do: {f, src}
+
+      assert length(users) > 15, "only #{length(users)} files use the helper; the census is blind"
+
+      violations =
+        for {f, src} <- users,
+            f != "test/session_case_test.exs",
+            {line, n} <- code_lines(src),
+            String.contains?(line, "collect("),
+            not String.contains?(line, "_ = collect("),
+            not String.contains?(line, "do_collect("),
+            do: "#{f}:#{n}: #{String.trim(line)}"
+
+      assert violations == [],
+             """
+             these call sites use the result of collect/3, which cannot tell a match from a \
+             timeout, so a slow runner reports a missing value instead of a deadline. Use await_event/3, \
+             which raises and names what it waited for:
+
+             #{Enum.join(violations, "\n")}
+             """
+    end
+  end
+
+  # Lines that are actually code: a `\"\"\"` block is documentation and a `#` line is a comment, and
+  # naming the helper in prose is not calling it. Without this the census reports its own
+  # explanation of itself.
+  defp code_lines(src) do
+    src
+    |> String.split("\n")
+    |> Enum.with_index(1)
+    |> Enum.reduce({false, []}, fn {line, n}, {in_doc?, acc} ->
+      fences =
+        line
+        |> String.graphemes()
+        |> Enum.chunk_every(3, 1)
+        |> Enum.count(&(&1 == ["\"", "\"", "\""]))
+
+      now_in_doc? = if rem(fences, 2) == 1, do: not in_doc?, else: in_doc?
+      keep? = not in_doc? and not now_in_doc? and not String.starts_with?(String.trim(line), "#")
+      {now_in_doc?, if(keep?, do: [{line, n} | acc], else: acc)}
+    end)
+    |> elem(1)
+    |> Enum.reverse()
+  end
+
   describe "AC4: collect/3 still settles for the callers that do not care" do
     test "it returns what it saw and does not raise", %{id: id} do
       send(self(), {:session, id, {:state, :idle}})

test/support/session_case.ex

@@ -58,7 +58,43 @@ defmodule Trinity.SessionCase do
     {:ok, pid}
   end
 
-  @doc "Collects session events for `session_id` until `until` matches or the timeout passes."
+  @doc """
+  Waits for a session event matching `until`, and **raises** if the deadline passes first.
+
+  Reach for this whenever the test asserts on what came back. `collect/3` below cannot tell a
+  caller whether it matched or gave up, so an assertion after it reports a missing value rather
+  than a timeout, and a reader goes looking for a logic defect that is not there (slice 127).
+
+  The raise names the deadline and every event that did arrive, because "waited 15s and saw
+  thinking then idle" and "waited 15s and saw nothing at all" send a reader to opposite places.
+  """
+  def await_event(session_id, until, timeout \\ 5_000) do
+    deadline = System.monotonic_time(:millisecond) + timeout
+    events = do_collect(session_id, until, deadline, [])
+
+    if Enum.any?(events, until) do
+      events
+    else
+      raise """
+      waited #{timeout}ms for a #{inspect(session_id)} event matching the given condition and it \
+      never arrived.
+
+      Events seen, in order: #{inspect(events, limit: 30)}
+
+      An empty list means the session emitted nothing at all, which usually means it was never \
+      started, the test never subscribed, or the thing that should have triggered it did not run. \
+      A non-empty list means the session was working and did not reach this condition in time.
+      """
+    end
+  end
+
+  @doc """
+  Collects session events until `until` matches **or the timeout passes**, returning what it saw.
+
+  A timeout and a match return the same shape, so this cannot be asserted on: use `await_event/3` for
+  that. This is for settling, where the caller genuinely does not care whether the condition was
+  reached, and a census in `test/session_case_test.exs` enforces that its result is discarded.
+  """
   def collect(session_id, until, timeout \\ 5_000) do
     deadline = System.monotonic_time(:millisecond) + timeout
     do_collect(session_id, until, deadline, [])

test/trinity/mcp/mrtr_test.exs

@@ -68,7 +68,7 @@ defmodule Trinity.MCP.MrtrTest do
 
     # The owner answers; the Session re-runs the call; the retry echoes the state and completes.
     {:ok, %Approval{answer: @answer}} = Permissions.decide_request(a2, :once, answer: @answer)
-    events = collect(id, &match?({:state, :idle}, &1))
+    events = await_event(id, &match?({:state, :idle}, &1))
 
     assert {:assistant_message, %{content: "done "}} =
              Enum.find(events, &match?({:assistant_message, _}, &1))

test/trinity/memory/compaction_test.exs

@@ -150,7 +150,7 @@ defmodule Trinity.Memory.CompactionTest do
       repeats = div((soft + 600 - base) * 3, byte_size(unit))
       {:ok, _} = Session.send_user_message(pid, String.duplicate(unit, repeats))
 
-      events = collect(id, &match?({:state, :compacting}, &1), 5_000)
+      events = await_event(id, &match?({:state, :compacting}, &1), 5_000)
 
       assert :compacting in for({:state, s} <- events, do: s),
              "the kill needs the compacting state"
@@ -168,7 +168,7 @@ defmodule Trinity.Memory.CompactionTest do
       # The retry: the next message compacts once, cleanly.
       Fake.object_delay(0)
       {:ok, _} = Session.send_user_message(new_pid, "after the kill")
-      events = collect(id, &match?({:state, :idle}, &1), 15_000)
+      events = await_event(id, &match?({:state, :idle}, &1), 15_000)
       assert :compacting in for({:state, s} <- events, do: s)
       compactions = Enum.filter(Sessions.history(id, limit: 500), &Compactor.compaction?/1)
       assert length(compactions) == before + 1
@@ -193,7 +193,16 @@ defmodule Trinity.Memory.CompactionTest do
       Fake.script(script_deltas(1, "ok"))
       huge = String.duplicate("a message that no compaction can shrink ", 600)
       {:ok, _} = Session.send_user_message(pid, huge)
-      events = collect(id, &match?({:forked, _}, &1), 15_000)
+      # 60s, and the number is measured rather than picked. This wait is the only real one in this
+      # file: locally it takes 4,745 to 4,786 ms over five runs, where the two waits above return
+      # from the mailbox in 0 to 3 ms. The old 15,000 was 3.1x the local figure and the FIPS leg
+      # still missed it (run 36253217084), which is what opened slice 127.
+      #
+      # A generous deadline costs nothing when the event arrives, because `await_event/3` returns
+      # the instant the condition matches; the only thing it buys is how long a genuine hang takes
+      # to report. So the trade is "long enough that a slow runner never trips it", and 12.6x the
+      # local median is that.
+      events = await_event(id, &match?({:forked, _}, &1), 60_000)
       assert {:forked, child_id} = Enum.find(events, &match?({:forked, _}, &1))
       assert :compacting in for({:state, s} <- events, do: s)
 

test/trinity/permissions/session_flow_test.exs

@@ -39,7 +39,7 @@ defmodule Trinity.Permissions.SessionFlowTest do
     tool_turn([{"c1", "echo", %{"text" => "plain"}}])
     {:ok, pid} = start_drained(id)
     {:ok, _} = Session.send_user_message(pid, "go")
-    events = collect(id, &match?({:state, :idle}, &1))
+    events = await_event(id, &match?({:state, :idle}, &1))
     refute :approval_wait in for({:state, s} <- events, do: s)
     refute_received {:approval, _, _}
     assert Permissions.list_approvals(session_id: id) == []
@@ -61,7 +61,7 @@ defmodule Trinity.Permissions.SessionFlowTest do
     assert [%Approval{id: ^aid}] = Permissions.pending(id)
 
     {:ok, _} = Permissions.decide_request(aid, :once)
-    events = collect(id, &match?({:state, :idle}, &1))
+    events = await_event(id, &match?({:state, :idle}, &1))
 
     assert {:assistant_message, %{content: "done done "}} =
              Enum.find(events, &match?({:assistant_message, _}, &1))
@@ -83,7 +83,7 @@ defmodule Trinity.Permissions.SessionFlowTest do
     # The same call again: no request.
     tool_turn([{"c1", "write_note", @args}])
     {:ok, _} = Session.send_user_message(pid, "two")
-    events = collect(id, &match?({:state, :idle}, &1))
+    events = await_event(id, &match?({:state, :idle}, &1))
     refute :approval_wait in for({:state, s} <- events, do: s)
     refute_received {:approval, :requested, _}
     assert length(tool_rows(id)) == 2
@@ -112,7 +112,7 @@ defmodule Trinity.Permissions.SessionFlowTest do
     {:ok, _} = Session.send_user_message(pid, "write")
     assert_receive {:approval, :requested, %Approval{id: aid}}, 2_000
     {:ok, _} = Permissions.decide_request(aid, :deny)
-    events = collect(id, &match?({:state, :idle}, &1))
+    events = await_event(id, &match?({:state, :idle}, &1))
 
     assert Enum.any?(
              events,

test/trinity/sessions/crash_test.exs

@@ -53,7 +53,7 @@ defmodule Trinity.Sessions.CrashTest do
     assert {:turn_interrupted,
             %Message{id: ^draft_id, parts: %{"interrupted" => true, "draft" => false}}} =
              Enum.find(
-               collect(a.id, &match?({:turn_interrupted, _}, &1), 2_000),
+               await_event(a.id, &match?({:turn_interrupted, _}, &1), 2_000),
                &match?({:turn_interrupted, _}, &1)
              )
 

test/trinity/sessions/many_sessions_test.exs

@@ -21,7 +21,7 @@ defmodule Trinity.Sessions.ManySessionsTest do
         fn r ->
           :ok = Sessions.subscribe(r.id)
           {:ok, _} = Sessions.send_user_message(r.id, "hi")
-          events = collect(r.id, &match?({:assistant_message, _}, &1), 30_000)
+          events = await_event(r.id, &match?({:assistant_message, _}, &1), 30_000)
           {r.id, match?({:assistant_message, _}, List.last(events))}
         end,
         max_concurrency: @n,

test/trinity/sessions/session_test.exs

@@ -36,7 +36,7 @@ defmodule Trinity.Sessions.SessionTest do
       {:ok, pid} = start_drained(id)
       assert {:ok, %Message{seq: 1, role: "user"}} = Session.send_user_message(pid, "hello")
 
-      events = collect(id, &match?({:assistant_message, _}, &1))
+      events = await_event(id, &match?({:assistant_message, _}, &1))
       assert Enum.all?(events, &Events.valid?/1)
 
       assert {:user_message, %Message{content: "hello"}} =
@@ -75,7 +75,7 @@ defmodule Trinity.Sessions.SessionTest do
       Fake.script(script_deltas(1_000, "y"))
       {:ok, pid} = start_drained(id)
       {:ok, _} = Session.send_user_message(pid, "go")
-      events = collect(id, &match?({:assistant_message, _}, &1))
+      events = await_event(id, &match?({:assistant_message, _}, &1))
       deltas = for {:assistant_delta, s} <- events, do: s
       assert length(deltas) <= 25, "got #{length(deltas)} delta broadcasts"
       assert Enum.join(deltas) == String.duplicate("y", 1_000)
@@ -91,7 +91,7 @@ defmodule Trinity.Sessions.SessionTest do
       Fake.scripts([nil, script_deltas(3, "final ")] |> Enum.map(&(&1 || default_script())))
       {:ok, pid} = start_drained(id)
       {:ok, _} = Session.send_user_message(pid, "weather?")
-      events = collect(id, fn e -> match?({:state, :idle}, e) and true end, 8_000)
+      events = await_event(id, fn e -> match?({:state, :idle}, e) and true end, 8_000)
       states = for {:state, s} <- events, do: s
       assert :tool_wait in states
       assert {:tool_call, %{id: "call_1", name: "get_weather"}} in events
@@ -137,7 +137,7 @@ defmodule Trinity.Sessions.SessionTest do
       Fake.fail(1, Trinity.LLM.Error.permanent(:boom))
       {:ok, pid} = start_drained(id)
       {:ok, _} = Session.send_user_message(pid, "hi")
-      events = collect(id, &match?({:state, :idle}, &1))
+      events = await_event(id, &match?({:state, :idle}, &1))
       states = for {:state, s} <- events, do: s
       assert :error in states
       assert {:error, _} = Enum.find(events, &match?({:error, _}, &1))
@@ -152,14 +152,14 @@ defmodule Trinity.Sessions.SessionTest do
       assert {:ok, _} = Session.send_user_message(pid, "again")
 
       assert {:assistant_message, _} =
-               List.last(collect(id, &match?({:assistant_message, _}, &1)))
+               List.last(await_event(id, &match?({:assistant_message, _}, &1)))
     end
 
     test "a Task that raises is an error turn too", %{id: id} do
       Fake.script([{:text_delta, "partial "}, :raise_now])
       {:ok, pid} = start_drained(id)
       {:ok, _} = Session.send_user_message(pid, "hi")
-      events = collect(id, &match?({:state, :idle}, &1))
+      events = await_event(id, &match?({:state, :idle}, &1))
       assert :error in for({:state, s} <- events, do: s)
 
       assert [
@@ -191,7 +191,7 @@ defmodule Trinity.Sessions.SessionTest do
 
       assert {:turn_interrupted, %Message{content: "start ", parts: %{"interrupted" => true}}} =
                Enum.find(
-                 collect(id, &match?({:turn_interrupted, _}, &1), 1_000),
+                 await_event(id, &match?({:turn_interrupted, _}, &1), 1_000),
                  &match?({:turn_interrupted, _}, &1)
                )
 

test/trinity/tools/runner_test.exs

@@ -45,7 +45,7 @@ defmodule Trinity.Tools.RunnerTest do
     {t, {:ok, _}} = :timer.tc(fn -> Session.send_user_message(pid, "go") end)
     _ = t
     started = System.monotonic_time(:millisecond)
-    events = collect(id, &match?({:state, :idle}, &1), 10_000)
+    events = await_event(id, &match?({:state, :idle}, &1), 10_000)
     elapsed = System.monotonic_time(:millisecond) - started
     {events, elapsed, Sessions.history(id)}
   end

test/trinity_web/live/approval_live_test.exs

@@ -6,7 +6,7 @@ defmodule TrinityWeb.ApprovalLiveTest do
   @moduletag :capture_log
 
   import Phoenix.LiveViewTest
-  import Trinity.SessionCase, only: [script_deltas: 2, collect: 3]
+  import Trinity.SessionCase, only: [script_deltas: 2, collect: 3, await_event: 3]
 
   alias Trinity.Factory
   alias Trinity.LLM.Providers.Fake
@@ -84,7 +84,7 @@ defmodule TrinityWeb.ApprovalLiveTest do
     # Under the rule the next call runs without a card.
     write_turn()
     send_message(view, "again")
-    events = collect(id, &match?({:state, :idle}, &1), 5_000)
+    events = await_event(id, &match?({:state, :idle}, &1), 5_000)
     refute :approval_wait in for({:state, s} <- events, do: s)
 
     # A call outside the rule asks, and Deny records the denial.