From 1aa1a7ddf25a2ecedb5672d85722839bcd015d60 Mon Sep 17 00:00:00 2001 From: deepfates Date: Mon, 6 Jul 2026 19:03:35 -0700 Subject: [PATCH 01/10] fix: handle port child epipe shutdown --- REPORT.md | 64 +++++++++++++++++++++ lib/cantrip/medium/code/port_child.ex | 40 ++++++++++--- test/port_code_medium_test.exs | 83 +++++++++++++++++++++++++++ 3 files changed, 179 insertions(+), 8 deletions(-) create mode 100644 REPORT.md diff --git a/REPORT.md b/REPORT.md new file mode 100644 index 0000000..3ecf757 --- /dev/null +++ b/REPORT.md @@ -0,0 +1,64 @@ +# can-ix5r Report + +Branch: `epipe-graceful` + +## Reproduction + +Pre-fix command, run from the repository root: + +```sh +mix run -e 'port = Port.open({:spawn_executable, System.find_executable("elixir")}, [:binary, :exit_status, {:packet, 4}, args: Enum.flat_map(:code.get_path(), &["-pa", List.to_string(&1)]) ++ ["-e", "Cantrip.Medium.Code.PortChild.main()"]]); send_frame = fn term -> Port.command(port, :erlang.term_to_binary(term)) end; send_frame.({:init, []}); receive do {^port, {:data, payload}} -> IO.inspect(:erlang.binary_to_term(payload), label: "init") after 5000 -> IO.puts("init timeout") end; ref = System.unique_integer([:positive, :monotonic]); send_frame.({:eval, ref, "Process.sleep(200); done.(\"late\")", %{gate_names: ["done"], evaluator: :raw}}); Port.close(port); Process.sleep(1000); IO.puts("parent survived")' +``` + +Pre-fix output: + +```text +init: :ready +** (EXIT from #PID<0.94.0>) an exception was raised: + ** (ErlangError) Erlang error: :epipe + (elixir 1.19.5) lib/io.ex:308: IO.binwrite/2 + lib/cantrip/medium/code/port_child.ex:856: Cantrip.Medium.Code.PortChild.do_write_frame/2 + lib/cantrip/medium/code/port_child.ex:165: Cantrip.Medium.Code.PortChild.protocol_loop/2 + +parent survived +``` + +Sequence: parent starts `Cantrip.Medium.Code.PortChild`, receives `:ready`, sends an eval that sleeps, closes the Erlang port before the child writes its reply, then the child attempts to write a frame to closed stdout. `IO.binwrite/2` raises `:epipe` inside the linked protocol process. + +## Fix + +`Cantrip.Medium.Code.PortChild.do_write_frame/2` now catches write failures, including `:epipe`, and returns `{:error, reason}` instead of letting the linked protocol process crash. The top-level child loop checks reply-write results; a failed write logs a stderr Goodbye line and exits normally. + +Post-fix reproduction output: + +```text +init: :ready +cantrip port child: parent port closed during write (:epipe); Goodbye. +parent survived +``` + +The regression test `port child treats epipe during reply as graceful shutdown` uses the real child process and packet protocol, closes the parent port before the delayed child reply, asserts the Goodbye log, and asserts no new `erl_crash.dump*` appears in the repo root. + +## Verification + +- `mix test test/port_code_medium_test.exs` passed: 21 tests, 0 failures. +- `mix test --only mnesia --max-cases 1` passed: 56 tests, 0 failures. +- Isolated reruns of suite-timeout locations passed: + - `mix test test/familiar_behavior_test.exs:108` + - `mix test test/familiar_behavior_test.exs:159` + - `mix test test/familiar_behavior_test.exs:403` + +Full-suite status: + +- `mix test` failed once with a Mnesia backend timeout in `test/loom_backend_symmetry_test.exs:43`. +- `mix test --max-cases 1` later cascaded into Familiar behavior timeouts and was stopped after the third timeout. +- Those failures were not fixed here because the affected tests pass in isolation and are outside the port-child `:epipe` hardening scope. + +## Discovered Tickets + +- `can-nlga`: Mnesia backend symmetry test can time out in full suite. +- `can-petw`: Familiar behavior tests can cascade time out in full-suite runs. + +## Uncertainty + +I did not find original incident logs from 2026-07-05 in this workspace, so the exact production trigger remains inferred. The reproduced failure matches the requested root sequence: parent port closes, child writes to it, `IO.binwrite/2` raises `:epipe`, and the linked protocol process can take down the child BEAM. The fix hardens that sequence directly. diff --git a/lib/cantrip/medium/code/port_child.ex b/lib/cantrip/medium/code/port_child.ex index 77dae44..e5a4164 100644 --- a/lib/cantrip/medium/code/port_child.ex +++ b/lib/cantrip/medium/code/port_child.ex @@ -171,27 +171,30 @@ defmodule Cantrip.Medium.Code.PortChild do defp loop(state) do case read_frame() do {:ok, {:init, binding}} -> - write_frame(:ready) - loop(%{state | binding: persist_binding(binding)}) + continue_after_write(write_frame(:ready), %{state | binding: persist_binding(binding)}) {:ok, {:eval, ref, code, env}} when is_binary(code) and is_map(env) -> {next_state, response} = eval(code, state, env, ref) - write_frame(response) - loop(next_state) + continue_after_write(write_frame(response), next_state) {:ok, _other} -> - write_frame({:error, :unexpected_frame}) - loop(state) + continue_after_write(write_frame({:error, :unexpected_frame}), state) :eof -> :ok {:error, reason} -> - write_frame({:error, reason}) - loop(state) + continue_after_write(write_frame({:error, reason}), state) end end + defp continue_after_write(:ok, state), do: loop(state) + + defp continue_after_write({:error, reason}, _state) do + log_graceful_shutdown(reason) + :ok + end + defp eval(code, state, env, ref) do with_child_telemetry_context(env, fn -> do_eval(code, state, env, ref) @@ -855,6 +858,27 @@ defmodule Cantrip.Medium.Code.PortChild do payload = :erlang.term_to_binary(term) IO.binwrite(output, <>) :ok + rescue + e in ErlangError -> + case e.original do + :epipe -> {:error, :epipe} + reason -> {:error, {:erlang_error, reason}} + end + + e -> + {:error, Cantrip.SafeFormat.exception(e)} + catch + :exit, reason -> + {:error, {:exit, reason}} + end + + defp log_graceful_shutdown(reason) do + IO.puts( + :stderr, + "cantrip port child: parent port closed during write (#{inspect(reason)}); Goodbye." + ) + rescue + _ -> :ok end defp protocol do diff --git a/test/port_code_medium_test.exs b/test/port_code_medium_test.exs index 7fb6872..7acf295 100644 --- a/test/port_code_medium_test.exs +++ b/test/port_code_medium_test.exs @@ -157,6 +157,45 @@ defmodule PortCodeMediumTest do if tmp = Process.get(:cantrip_port_runner_tmp), do: File.rm_rf!(tmp) end + test "port child treats epipe during reply as graceful shutdown" do + tmp = + Path.join(System.tmp_dir!(), "cantrip_port_epipe_#{System.unique_integer([:positive])}") + + Process.put(:cantrip_port_epipe_tmp, tmp) + File.mkdir_p!(tmp) + + log_path = Path.join(tmp, "child-stderr.log") + runner_path = Path.join(tmp, "runner.sh") + + File.write!(runner_path, """ + #!/bin/sh + exec "$@" 2> #{log_path} + """) + + File.chmod!(runner_path, 0o755) + + before_dumps = repo_crash_dumps() + port = open_port_child(runner_path) + send_port_frame(port, {:init, []}) + + assert_receive {^port, {:data, payload}}, 5_000 + assert :erlang.binary_to_term(payload) == :ready + + ref = System.unique_integer([:positive, :monotonic]) + + send_port_frame( + port, + {:eval, ref, ~S[Process.sleep(200); done.("late")], + %{gate_names: ["done"], evaluator: :raw}} + ) + + Port.close(port) + assert eventually_contains?(log_path, "Goodbye.", 2_000) + assert repo_crash_dumps() == before_dumps + after + if tmp = Process.get(:cantrip_port_epipe_tmp), do: File.rm_rf!(tmp) + end + test "child BEAM global state does not mutate the host BEAM" do key = {__MODULE__, :persistent_term_isolation} :persistent_term.erase(key) @@ -619,4 +658,48 @@ defmodule PortCodeMediumTest do :code.purge(module) :code.delete(module) end + + defp open_port_child(runner_path) do + elixir = System.find_executable("elixir") + + child_args = + Enum.flat_map(:code.get_path(), &["-pa", List.to_string(&1)]) ++ + ["-e", "Cantrip.Medium.Code.PortChild.main()"] + + Port.open({:spawn_executable, runner_path}, [ + :binary, + :exit_status, + {:packet, 4}, + args: [elixir | child_args] + ]) + end + + defp send_port_frame(port, term) do + Port.command(port, :erlang.term_to_binary(term)) + end + + defp eventually_contains?(path, expected, timeout_ms) do + deadline = System.monotonic_time(:millisecond) + timeout_ms + eventually_contains_until?(path, expected, deadline) + end + + defp eventually_contains_until?(path, expected, deadline) do + if File.exists?(path) and String.contains?(File.read!(path), expected) do + true + else + if System.monotonic_time(:millisecond) >= deadline do + false + else + Process.sleep(25) + eventually_contains_until?(path, expected, deadline) + end + end + end + + defp repo_crash_dumps do + File.cwd!() + |> Path.join("erl_crash.dump*") + |> Path.wildcard() + |> MapSet.new() + end end From ea9d4a962c0d0f0c1533958c47085616ad10d888 Mon Sep 17 00:00:00 2001 From: deepfates Date: Mon, 6 Jul 2026 19:22:11 -0700 Subject: [PATCH 02/10] fix: narrow graceful path to :epipe, exit {:shutdown,...}, add loom flush/close lifecycle (round 2 per adversarial review) --- REPORT.md | 14 ++++- lib/cantrip/loom.ex | 29 +++++++++ lib/cantrip/loom/storage.ex | 4 +- lib/cantrip/loom/storage/jsonl.ex | 6 ++ lib/cantrip/loom/storage/mnesia.ex | 16 +++++ lib/cantrip/medium/code/port_child.ex | 44 ++++++++++--- test/port_code_medium_test.exs | 38 ++++++++++- test/support/lifecycle_loom_storage.ex | 87 ++++++++++++++++++++++++++ 8 files changed, 224 insertions(+), 14 deletions(-) create mode 100644 test/support/lifecycle_loom_storage.ex diff --git a/REPORT.md b/REPORT.md index 3ecf757..4db98f7 100644 --- a/REPORT.md +++ b/REPORT.md @@ -27,7 +27,15 @@ Sequence: parent starts `Cantrip.Medium.Code.PortChild`, receives `:ready`, send ## Fix -`Cantrip.Medium.Code.PortChild.do_write_frame/2` now catches write failures, including `:epipe`, and returns `{:error, reason}` instead of letting the linked protocol process crash. The top-level child loop checks reply-write results; a failed write logs a stderr Goodbye line and exits normally. +`Cantrip.Medium.Code.PortChild.do_write_frame/2` now handles only the closed-pipe write failure. `:epipe` is returned to the main child loop as `{:error, :epipe}`; any other write-side exception is re-raised so unexpected write failures still fail loudly. + +The top-level child loop treats `:epipe` as an expected shutdown class, not a successful normal exit. Before exiting it: + +- flushes and closes the current transient loom via `Cantrip.Loom.close/1`; +- logs the stderr Goodbye line; +- exits with `{:shutdown, {:port_write_failed, :epipe}}`. + +`Cantrip.Loom.Storage` now exposes optional `flush/1` and `close/1` lifecycle callbacks. JSONL implements them as synchronous/no-op lifecycle hooks because every append is already a complete `File.write!/3`; Mnesia implements `flush/1` with `:mnesia.sync_log/0` and leaves `close/1` as `:ok` because Mnesia is a shared VM application, not a per-loom handle. Post-fix reproduction output: @@ -37,12 +45,12 @@ cantrip port child: parent port closed during write (:epipe); Goodbye. parent survived ``` -The regression test `port child treats epipe during reply as graceful shutdown` uses the real child process and packet protocol, closes the parent port before the delayed child reply, asserts the Goodbye log, and asserts no new `erl_crash.dump*` appears in the repo root. +The regression test `port child treats epipe during reply as graceful shutdown` uses the real child process and packet protocol. It sends a loom into the child, appends a sentinel final turn before the delayed reply, closes the parent port to force `:epipe`, asserts `flush` then `close` were called by the storage backend, reloads the same loom storage, and asserts the sentinel last turn is present after re-summon. It also asserts the Goodbye log and no new `erl_crash.dump*` in the repo root. ## Verification - `mix test test/port_code_medium_test.exs` passed: 21 tests, 0 failures. -- `mix test --only mnesia --max-cases 1` passed: 56 tests, 0 failures. +- `mix test --only mnesia` passed: 56 tests, 0 failures. - Isolated reruns of suite-timeout locations passed: - `mix test test/familiar_behavior_test.exs:108` - `mix test test/familiar_behavior_test.exs:159` diff --git a/lib/cantrip/loom.ex b/lib/cantrip/loom.ex index a817dbe..e7a5cf7 100644 --- a/lib/cantrip/loom.ex +++ b/lib/cantrip/loom.ex @@ -415,6 +415,19 @@ defmodule Cantrip.Loom do end end + @doc """ + Flush and close the loom's storage backend when it exposes lifecycle hooks. + + Most built-in writes are synchronous, but this gives shutdown paths one + explicit place to force durable state before the process exits. + """ + @spec close(t()) :: :ok | {:error, term()} + def close(%__MODULE__{storage_module: module, storage_state: storage_state}) do + with {:ok, storage_state} <- flush_storage(module, storage_state) do + close_storage(module, storage_state) + end + end + @doc """ Branches `cantrip` from a prefix of `loom`. @@ -508,6 +521,22 @@ defmodule Cantrip.Loom do end end + defp flush_storage(module, storage_state) do + if function_exported?(module, :flush, 1) do + module.flush(storage_state) + else + {:ok, storage_state} + end + end + + defp close_storage(module, storage_state) do + if function_exported?(module, :close, 1) do + module.close(storage_state) + else + :ok + end + end + defp emit_persist_error(module, event, reason) do metadata = %{ diff --git a/lib/cantrip/loom/storage.ex b/lib/cantrip/loom/storage.ex index c4db8f8..267957c 100644 --- a/lib/cantrip/loom/storage.ex +++ b/lib/cantrip/loom/storage.ex @@ -14,6 +14,8 @@ defmodule Cantrip.Loom.Storage do @callback append_turn(storage_state(), map()) :: {:ok, storage_state()} | {:error, term()} @callback annotate_reward(storage_state(), non_neg_integer(), term()) :: {:ok, storage_state()} | {:error, term()} + @callback flush(storage_state()) :: {:ok, storage_state()} | {:error, term()} + @callback close(storage_state()) :: :ok | {:error, term()} @doc """ Load prior persisted state into a freshly-initialized backend. @@ -29,5 +31,5 @@ defmodule Cantrip.Loom.Storage do @callback load(storage_state()) :: {:ok, %{events: [map()], turns: [map()]}} | {:error, term()} - @optional_callbacks append_event: 2, load: 1 + @optional_callbacks append_event: 2, load: 1, flush: 1, close: 1 end diff --git a/lib/cantrip/loom/storage/jsonl.ex b/lib/cantrip/loom/storage/jsonl.ex index 9cab679..f74e06c 100644 --- a/lib/cantrip/loom/storage/jsonl.ex +++ b/lib/cantrip/loom/storage/jsonl.ex @@ -17,6 +17,12 @@ defmodule Cantrip.Loom.Storage.Jsonl do def init(_), do: {:error, "jsonl storage requires a file path"} + @impl true + def flush(state), do: {:ok, state} + + @impl true + def close(_state), do: :ok + @impl true def append_turn(%{path: path} = state, turn) do append_jsonl(path, storage_event(%{type: :turn, turn: turn})) diff --git a/lib/cantrip/loom/storage/mnesia.ex b/lib/cantrip/loom/storage/mnesia.ex index dae787d..a64fcea 100644 --- a/lib/cantrip/loom/storage/mnesia.ex +++ b/lib/cantrip/loom/storage/mnesia.ex @@ -79,6 +79,22 @@ defmodule Cantrip.Loom.Storage.Mnesia do end end + @impl true + def flush(%{mnesia: mnesia} = state) do + case call(mnesia, :sync_log, []) do + :ok -> {:ok, state} + {:error, reason} -> {:error, reason} + other -> {:error, other} + end + rescue + e -> {:error, Cantrip.SafeFormat.exception(e)} + catch + :exit, reason -> {:error, {:exit, reason}} + end + + @impl true + def close(_state), do: :ok + defp classify_native(events) do {evts, trns} = Enum.reduce(events, {[], []}, fn stored_event, {evts_acc, trns_acc} -> diff --git a/lib/cantrip/medium/code/port_child.ex b/lib/cantrip/medium/code/port_child.ex index e5a4164..aef0031 100644 --- a/lib/cantrip/medium/code/port_child.ex +++ b/lib/cantrip/medium/code/port_child.ex @@ -190,9 +190,11 @@ defmodule Cantrip.Medium.Code.PortChild do defp continue_after_write(:ok, state), do: loop(state) - defp continue_after_write({:error, reason}, _state) do - log_graceful_shutdown(reason) - :ok + defp continue_after_write({:error, :epipe}, state) do + reason = {:shutdown, {:port_write_failed, :epipe}} + finalize_loom_for_shutdown(state) + log_graceful_shutdown(:epipe) + exit(reason) end defp eval(code, state, env, ref) do @@ -275,6 +277,7 @@ defmodule Cantrip.Medium.Code.PortChild do next_state = state |> Map.put(:binding, persist_binding(binding)) + |> put_current_loom(binding) |> Map.delete(:dune_session) {next_state, {:eval_result, ref, next_state.binding, value, terminated?}} @@ -297,6 +300,7 @@ defmodule Cantrip.Medium.Code.PortChild do next_state = state |> Map.put(:binding, clean_bindings) + |> put_current_loom(next_session.bindings) |> Map.put(:dune_session, %{next_session | bindings: clean_bindings}) {next_state, {:eval_result, ref, clean_bindings, value, terminated?}} @@ -307,6 +311,7 @@ defmodule Cantrip.Medium.Code.PortChild do next_state = state |> Map.put(:binding, clean_bindings) + |> put_current_loom(session.bindings) |> Map.put(:dune_session, %{session | bindings: clean_bindings}) {next_state, {:eval_error, ref, clean_bindings, reason}} @@ -691,6 +696,13 @@ defmodule Cantrip.Medium.Code.PortChild do |> Enum.reject(fn {_k, v} -> transient_value?(v) end) end + defp put_current_loom(state, binding) do + case Keyword.get(normalize_binding(binding), :loom) do + %Cantrip.Loom{} = loom -> Map.put(state, :current_loom, loom) + _ -> state + end + end + defp externalize_binding(binding) do Enum.map(binding, fn {key, value} -> {to_string(key), externalize_term(value)} end) end @@ -862,16 +874,23 @@ defmodule Cantrip.Medium.Code.PortChild do e in ErlangError -> case e.original do :epipe -> {:error, :epipe} - reason -> {:error, {:erlang_error, reason}} + _reason -> reraise(e, __STACKTRACE__) end + end - e -> - {:error, Cantrip.SafeFormat.exception(e)} + defp finalize_loom_for_shutdown(%{current_loom: %Cantrip.Loom{} = loom}) do + case Cantrip.Loom.close(loom) do + :ok -> :ok + {:error, reason} -> log_loom_close_error(reason) + end + rescue + e -> log_loom_close_error(Cantrip.SafeFormat.exception(e)) catch - :exit, reason -> - {:error, {:exit, reason}} + :exit, reason -> log_loom_close_error({:exit, reason}) end + defp finalize_loom_for_shutdown(_state), do: :ok + defp log_graceful_shutdown(reason) do IO.puts( :stderr, @@ -881,6 +900,15 @@ defmodule Cantrip.Medium.Code.PortChild do _ -> :ok end + defp log_loom_close_error(reason) do + IO.puts( + :stderr, + "cantrip port child: loom close failed during epipe shutdown: #{inspect(reason)}" + ) + rescue + _ -> :ok + end + defp protocol do Process.get(:cantrip_port_protocol) || :persistent_term.get({__MODULE__, :protocol}) diff --git a/test/port_code_medium_test.exs b/test/port_code_medium_test.exs index 7acf295..d1b5a49 100644 --- a/test/port_code_medium_test.exs +++ b/test/port_code_medium_test.exs @@ -174,6 +174,15 @@ defmodule PortCodeMediumTest do File.chmod!(runner_path, 0o755) + loom_path = Path.join(tmp, "loom.terms") + lifecycle_path = Path.join(tmp, "loom-lifecycle.log") + + storage = + {Cantrip.TestLifecycleLoomStorage, path: loom_path, lifecycle_path: lifecycle_path} + + Code.ensure_loaded!(Cantrip.TestLifecycleLoomStorage) + loom = Cantrip.Loom.new(%{identity: "epipe-test"}, storage: storage) + sentinel = "turn-before-epipe-#{System.unique_integer([:positive])}" before_dumps = repo_crash_dumps() port = open_port_child(runner_path) send_port_frame(port, {:init, []}) @@ -185,12 +194,17 @@ defmodule PortCodeMediumTest do send_port_frame( port, - {:eval, ref, ~S[Process.sleep(200); done.("late")], - %{gate_names: ["done"], evaluator: :raw}} + {:eval, ref, epipe_loom_code(sentinel), + %{gate_names: ["done"], evaluator: :raw, loom: loom}} ) Port.close(port) assert eventually_contains?(log_path, "Goodbye.", 2_000) + assert eventually_contains?(lifecycle_path, "close", 2_000) + assert File.read!(lifecycle_path) == "flush\nclose\n" + + rehydrated = Cantrip.Loom.new(%{identity: "epipe-test"}, storage: storage) + assert List.last(rehydrated.turns).utterance.content == sentinel assert repo_crash_dumps() == before_dumps after if tmp = Process.get(:cantrip_port_epipe_tmp), do: File.rm_rf!(tmp) @@ -678,6 +692,26 @@ defmodule PortCodeMediumTest do Port.command(port, :erlang.term_to_binary(term)) end + defp epipe_loom_code(sentinel) do + """ + Code.ensure_loaded!(Cantrip.TestLifecycleLoomStorage) + + loom = + Cantrip.Loom.append_turn(loom, %{ + cantrip_id: "epipe-cantrip", + entity_id: "epipe-entity", + role: "turn", + utterance: %{content: #{inspect(sentinel)}}, + observation: [%{gate: "done", result: "late", is_error: false}], + gate_calls: [], + metadata: %{source: "epipe-regression"} + }) + + Process.sleep(200) + done.("late") + """ + end + defp eventually_contains?(path, expected, timeout_ms) do deadline = System.monotonic_time(:millisecond) + timeout_ms eventually_contains_until?(path, expected, deadline) diff --git a/test/support/lifecycle_loom_storage.ex b/test/support/lifecycle_loom_storage.ex new file mode 100644 index 0000000..366abdf --- /dev/null +++ b/test/support/lifecycle_loom_storage.ex @@ -0,0 +1,87 @@ +defmodule Cantrip.TestLifecycleLoomStorage do + @moduledoc false + + @behaviour Cantrip.Loom.Storage + + @impl true + def init(opts) do + opts = Map.new(opts) + path = Map.fetch!(opts, :path) + lifecycle_path = Map.fetch!(opts, :lifecycle_path) + + File.mkdir_p!(Path.dirname(path)) + File.write!(path, "", [:append]) + File.write!(lifecycle_path, "", [:append]) + + {:ok, %{path: path, lifecycle_path: lifecycle_path}} + rescue + e -> {:error, Cantrip.SafeFormat.exception(e)} + end + + @impl true + def append_event(%{path: path} = state, event) do + append_term(path, event) + {:ok, state} + rescue + e -> {:error, Cantrip.SafeFormat.exception(e)} + end + + @impl true + def append_turn(%{path: path} = state, turn) do + append_term(path, %{type: :turn, turn: turn}) + {:ok, state} + rescue + e -> {:error, Cantrip.SafeFormat.exception(e)} + end + + @impl true + def annotate_reward(%{path: path} = state, index, reward) do + append_term(path, %{type: :reward, index: index, reward: reward}) + {:ok, state} + rescue + e -> {:error, Cantrip.SafeFormat.exception(e)} + end + + @impl true + def load(%{path: path}) do + events = read_terms(path) + + turns = + Enum.flat_map(events, fn + %{type: type, turn: turn} when type in [:turn, "turn"] -> [turn] + _ -> [] + end) + + {:ok, %{events: events, turns: turns}} + rescue + e -> {:error, Cantrip.SafeFormat.exception(e)} + end + + @impl true + def flush(%{lifecycle_path: lifecycle_path} = state) do + File.write!(lifecycle_path, "flush\n", [:append]) + {:ok, state} + rescue + e -> {:error, Cantrip.SafeFormat.exception(e)} + end + + @impl true + def close(%{lifecycle_path: lifecycle_path}) do + File.write!(lifecycle_path, "close\n", [:append]) + :ok + rescue + e -> {:error, Cantrip.SafeFormat.exception(e)} + end + + defp append_term(path, term) do + encoded = term |> :erlang.term_to_binary() |> Base.encode64() + File.write!(path, encoded <> "\n", [:append]) + end + + defp read_terms(path) do + path + |> File.read!() + |> String.split("\n", trim: true) + |> Enum.map(fn line -> line |> Base.decode64!() |> :erlang.binary_to_term() end) + end +end From ad7b6f1fdd1eb5f0f6731fee21259a7350b5dc55 Mon Sep 17 00:00:00 2001 From: deepfates Date: Mon, 6 Jul 2026 20:25:15 -0700 Subject: [PATCH 03/10] fix: teach loom transcript and bound turn inspection --- lib/cantrip/familiar.ex | 32 ++++++++++++++--- lib/cantrip/loom.ex | 74 ++++++++++++++++++++++++++++++++++++++ lib/cantrip/medium/code.ex | 24 +++++++++++-- test/familiar_test.exs | 23 ++++++++++++ test/loom_api_test.exs | 42 ++++++++++++++++++++++ 5 files changed, 189 insertions(+), 6 deletions(-) diff --git a/lib/cantrip/familiar.ex b/lib/cantrip/familiar.ex index e56c2f0..d3734cc 100644 --- a/lib/cantrip/familiar.ex +++ b/lib/cantrip/familiar.ex @@ -43,10 +43,33 @@ defmodule Cantrip.Familiar do You inhabit the System persistently. Variables you bind persist across turns and across sends within a single summoning. The loom persists across summonings — when you're summoned again against the - same loom, prior turns are available as `loom.turns`, and the + same loom, the conversation is available through the loom, and the bindings you left set are still set. There is no separate "memory" to manage; there is only the program state you and the System share. + The loom has two primary projections. Human or parent prompts are + `loom.intents`: maps with `role: "intent"` and + `utterance: %{content: text}`. Entity actions are `loom.turns`: maps + with `role: "turn"`, utterance, observation, metadata, and for + code/bash turns the full medium `code_state` needed for replay. To + read the conversation back chronologically, use + `Cantrip.Loom.transcript(loom)`: + + loom + |> Cantrip.Loom.transcript() + |> Enum.map(fn + %{role: "intent", utterance: %{content: text}} -> "you: " <> text + %{role: "turn", utterance: %{content: c}} -> "me: " <> (c || "") + end) + + Raw turns can be large because `:code_state` contains the full + binding snapshot. For IEx-readable inspection, use bounded + projections that keep the durable record untouched: + + Cantrip.Loom.bounded_turn(List.last(loom.turns)) + Cantrip.Loom.bounded_turns(loom) + Cantrip.Loom.bounded_transcript(loom) + ## Spawning other entities Your default workspace gates are read-only observation functions: @@ -186,9 +209,10 @@ defmodule Cantrip.Familiar do The workspace visible through `read_file`, `list_dir`, and `search` is the human's project; your own source normally lives in the Cantrip dependency outside that workspace. The loom persists across - summonings at this workspace, with prior turns visible as - `loom.turns`. If you want the spellbook's intellectual lineage, it - starts at https://deepfates.com/cantrip-bibliography. + summonings at this workspace: prompts are in `loom.intents`, entity + turns are in `loom.turns`, and `Cantrip.Loom.transcript(loom)` is + the conversation view. If you want the spellbook's intellectual + lineage, it starts at https://deepfates.com/cantrip-bibliography. You operate as an active inference loop. Take the step you predict will reduce your uncertainty. Observe what comes back. Update. diff --git a/lib/cantrip/loom.ex b/lib/cantrip/loom.ex index e7a5cf7..1c46767 100644 --- a/lib/cantrip/loom.ex +++ b/lib/cantrip/loom.ex @@ -55,6 +55,8 @@ defmodule Cantrip.Loom do storage_module: Memory, storage_state: %{} + @default_binding_preview_limit 20 + @type t :: %__MODULE__{ identity: term(), schema_version: pos_integer(), @@ -305,6 +307,78 @@ defmodule Cantrip.Loom do end) end + @doc """ + Inspection-safe projection of one turn. + + Code and bash turns carry a full `:code_state` snapshot so replay and fork + can restore the medium. That snapshot can be large enough that inspecting a + raw turn is a poor IEx experience. This helper keeps the durable turn intact + and returns a bounded view with `:code_state` removed and + `:code_state_summary` added. + + ## Options + + * `:binding_preview_limit` - number of binding names to include in the + summary. Defaults to #{@default_binding_preview_limit}. + """ + @spec bounded_turn(map(), keyword()) :: map() + def bounded_turn(%{} = turn, opts \\ []) when is_list(opts) do + case Map.fetch(turn, :code_state) do + {:ok, code_state} -> + turn + |> Map.delete(:code_state) + |> Map.put(:code_state_summary, summarize_code_state(code_state, opts)) + + :error -> + turn + end + end + + @doc """ + Inspection-safe projection of `loom.turns`. + + Returns the same turn records as `loom.turns`, except each turn is passed + through `bounded_turn/2` so embedded medium state is summarized rather than + expanded into the inspected value. + """ + @spec bounded_turns(t(), keyword()) :: [map()] + def bounded_turns(%__MODULE__{turns: turns}, opts \\ []) when is_list(opts) do + Enum.map(turns, &bounded_turn(&1, opts)) + end + + @doc """ + Inspection-safe projection of `transcript/1`. + + Intents are returned unchanged. Entity turns are passed through + `bounded_turn/2`, so this is the preferred IEx-readable conversation view + when turns may include large code-medium bindings. + """ + @spec bounded_transcript(t(), keyword()) :: [map()] + def bounded_transcript(%__MODULE__{} = loom, opts \\ []) when is_list(opts) do + loom + |> transcript() + |> Enum.map(fn + %{role: "turn"} = turn -> bounded_turn(turn, opts) + record -> record + end) + end + + defp summarize_code_state(code_state, opts) do + bindings = code_state_bindings(code_state) + limit = Keyword.get(opts, :binding_preview_limit, @default_binding_preview_limit) + names = Enum.map(bindings, fn {name, _value} -> name end) + + %{ + binding_count: length(bindings), + binding_preview: Enum.take(names, limit), + omitted_binding_count: max(length(bindings) - limit, 0) + } + end + + defp code_state_bindings(%{binding: binding}) when is_list(binding), do: binding + defp code_state_bindings(%{"binding" => binding}) when is_list(binding), do: binding + defp code_state_bindings(_), do: [] + def append_executed_turn(%__MODULE__{} = loom, turn_attrs, observations, opts \\ []) do initial_turn_count = length(loom.turns) diff --git a/lib/cantrip/medium/code.ex b/lib/cantrip/medium/code.ex index bfd025d..e21ee0a 100644 --- a/lib/cantrip/medium/code.ex +++ b/lib/cantrip/medium/code.ex @@ -590,8 +590,28 @@ defmodule Cantrip.Medium.Code do keys = binding() |> Keyword.keys() - The durable path you took is in `loom.turns`. Each turn is a map with - utterance, observation, and metadata; compose with `Enum.*` to query it. + The durable conversation is in `loom`. Human or parent prompts are + `loom.intents`: maps with `role: "intent"` and + `utterance: %{content: text}`. Entity actions are `loom.turns`: maps with + `role: "turn"`, utterance, observation, metadata, and for code/bash turns + the full medium `code_state` needed for replay. + + To read the conversation back chronologically, use + `Cantrip.Loom.transcript(loom)`: + + loom + |> Cantrip.Loom.transcript() + |> Enum.map(fn + %{role: "intent", utterance: %{content: text}} -> "you: " <> text + %{role: "turn", utterance: %{content: c}} -> "me: " <> (c || "") + end) + + Raw turns can be large because `:code_state` contains the full binding + snapshot. For IEx-readable inspection, use bounded projections: + + Cantrip.Loom.bounded_turn(List.last(loom.turns)) + Cantrip.Loom.bounded_turns(loom) + Cantrip.Loom.bounded_transcript(loom) """ end diff --git a/test/familiar_test.exs b/test/familiar_test.exs index f593c06..0d764f5 100644 --- a/test/familiar_test.exs +++ b/test/familiar_test.exs @@ -143,13 +143,36 @@ defmodule Cantrip.FamiliarTest do assert prompt =~ "choose the answer shape" assert prompt =~ "speech-shaped task" assert prompt =~ "Code.fetch_docs" + assert prompt =~ "loom.intents" assert prompt =~ "loom.turns" + assert prompt =~ "Cantrip.Loom.transcript(loom)" + assert prompt =~ "Cantrip.Loom.bounded_turn" + assert prompt =~ "Cantrip.Loom.bounded_transcript" + assert prompt =~ ~s|role: "intent"| + assert prompt =~ ~s|role: "turn"| + assert prompt =~ "you: " + assert prompt =~ "me: " assert prompt =~ "human's project" assert prompt =~ "conversation child" assert prompt =~ "raw file" assert prompt =~ "specific child, medium, or batch" end + test "code medium capability text teaches loom transcript and bounded inspection" do + llm = {FakeLLM, FakeLLM.new([])} + {:ok, cantrip} = Familiar.new(llm: llm) + + capability_text = Cantrip.Medium.Registry.present(cantrip.circle).capability_text + + assert capability_text =~ "loom.intents" + assert capability_text =~ "loom.turns" + assert capability_text =~ "Cantrip.Loom.transcript(loom)" + assert capability_text =~ "Cantrip.Loom.bounded_turn" + assert capability_text =~ "Cantrip.Loom.bounded_transcript" + assert capability_text =~ ~s|role: "intent"| + assert capability_text =~ ~s|role: "turn"| + end + test "respects custom max_turns" do llm = {FakeLLM, FakeLLM.new([])} {:ok, cantrip} = Familiar.new(llm: llm, max_turns: 10) diff --git a/test/loom_api_test.exs b/test/loom_api_test.exs index 3475b31..9b6769f 100644 --- a/test/loom_api_test.exs +++ b/test/loom_api_test.exs @@ -169,4 +169,46 @@ defmodule Cantrip.LoomAPITest do refute Map.has_key?(hd(parent_event.turn.observation), :child_turns) assert child_event.turn.utterance == child_turn.utterance end + + test "bounded projections summarize code_state without changing durable turns" do + loom = + %{system_prompt: nil} + |> Cantrip.Loom.new() + |> Cantrip.Loom.append_intent("what happened?") + |> Cantrip.Loom.append_turn(%{ + role: "turn", + utterance: %{content: "looked"}, + observation: [], + code_state: %{ + binding: [ + huge: String.duplicate("x", 10_000), + kept: {:ok, :value} + ] + } + }) + + [raw_turn] = loom.turns + assert Map.has_key?(raw_turn, :code_state) + + bounded_turn = Cantrip.Loom.bounded_turn(raw_turn, binding_preview_limit: 1) + refute Map.has_key?(bounded_turn, :code_state) + + assert bounded_turn.code_state_summary == %{ + binding_count: 2, + binding_preview: [:huge], + omitted_binding_count: 1 + } + + [bounded_from_loom] = Cantrip.Loom.bounded_turns(loom, binding_preview_limit: 1) + refute Map.has_key?(bounded_from_loom, :code_state) + + transcript = Cantrip.Loom.bounded_transcript(loom, binding_preview_limit: 1) + assert Enum.map(transcript, & &1.role) == ["intent", "turn"] + turn_projection = Enum.find(transcript, &(&1.role == "turn")) + refute Map.has_key?(turn_projection, :code_state) + assert turn_projection.code_state_summary.binding_preview == [:huge] + + [still_raw] = loom.turns + assert Map.has_key?(still_raw, :code_state) + end end From f1c6d3833909112f71fe0b4f8f628fa36787923d Mon Sep 17 00:00:00 2001 From: deepfates Date: Mon, 6 Jul 2026 20:25:46 -0700 Subject: [PATCH 04/10] docs: report dee-perc legibility pass --- REPORT.md | 77 +++++++++++++++++-------------------------------------- 1 file changed, 23 insertions(+), 54 deletions(-) diff --git a/REPORT.md b/REPORT.md index 4db98f7..cbc5896 100644 --- a/REPORT.md +++ b/REPORT.md @@ -1,72 +1,41 @@ -# can-ix5r Report +# dee-perc Report -Branch: `epipe-graceful` +Branch: `codex/dee-perc-iex-legibility` -## Reproduction +Code commit: `ad7b6f1 fix: teach loom transcript and bound turn inspection` -Pre-fix command, run from the repository root: +## Reproduction Commands -```sh -mix run -e 'port = Port.open({:spawn_executable, System.find_executable("elixir")}, [:binary, :exit_status, {:packet, 4}, args: Enum.flat_map(:code.get_path(), &["-pa", List.to_string(&1)]) ++ ["-e", "Cantrip.Medium.Code.PortChild.main()"]]); send_frame = fn term -> Port.command(port, :erlang.term_to_binary(term)) end; send_frame.({:init, []}); receive do {^port, {:data, payload}} -> IO.inspect(:erlang.binary_to_term(payload), label: "init") after 5000 -> IO.puts("init timeout") end; ref = System.unique_integer([:positive, :monotonic]); send_frame.({:eval, ref, "Process.sleep(200); done.(\"late\")", %{gate_names: ["done"], evaluator: :raw}}); Port.close(port); Process.sleep(1000); IO.puts("parent survived")' -``` - -Pre-fix output: - -```text -init: :ready -** (EXIT from #PID<0.94.0>) an exception was raised: - ** (ErlangError) Erlang error: :epipe - (elixir 1.19.5) lib/io.ex:308: IO.binwrite/2 - lib/cantrip/medium/code/port_child.ex:856: Cantrip.Medium.Code.PortChild.do_write_frame/2 - lib/cantrip/medium/code/port_child.ex:165: Cantrip.Medium.Code.PortChild.protocol_loop/2 +From the repository root: -parent survived +```sh +mix test test/loom_api_test.exs test/loom_intent_persistence_test.exs test/familiar_test.exs test/code_medium_ergonomics_test.exs ``` -Sequence: parent starts `Cantrip.Medium.Code.PortChild`, receives `:ready`, sends an eval that sleeps, closes the Erlang port before the child writes its reply, then the child attempts to write a frame to closed stdout. `IO.binwrite/2` raises `:epipe` inside the linked protocol process. - -## Fix - -`Cantrip.Medium.Code.PortChild.do_write_frame/2` now handles only the closed-pipe write failure. `:epipe` is returned to the main child loop as `{:error, :epipe}`; any other write-side exception is re-raised so unexpected write failures still fail loudly. +Result: 71 tests, 0 failures. -The top-level child loop treats `:epipe` as an expected shutdown class, not a successful normal exit. Before exiting it: - -- flushes and closes the current transient loom via `Cantrip.Loom.close/1`; -- logs the stderr Goodbye line; -- exits with `{:shutdown, {:port_write_failed, :epipe}}`. - -`Cantrip.Loom.Storage` now exposes optional `flush/1` and `close/1` lifecycle callbacks. JSONL implements them as synchronous/no-op lifecycle hooks because every append is already a complete `File.write!/3`; Mnesia implements `flush/1` with `:mnesia.sync_log/0` and leaves `close/1` as `:ok` because Mnesia is a shared VM application, not a per-loom handle. - -Post-fix reproduction output: - -```text -init: :ready -cantrip port child: parent port closed during write (:epipe); Goodbye. -parent survived +```sh +mix test ``` -The regression test `port child treats epipe during reply as graceful shutdown` uses the real child process and packet protocol. It sends a loom into the child, appends a sentinel final turn before the delayed reply, closes the parent port to force `:epipe`, asserts `flush` then `close` were called by the storage backend, reloads the same loom storage, and asserts the sentinel last turn is present after re-summon. It also asserts the Goodbye log and no new `erl_crash.dump*` in the repo root. - -## Verification - -- `mix test test/port_code_medium_test.exs` passed: 21 tests, 0 failures. -- `mix test --only mnesia` passed: 56 tests, 0 failures. -- Isolated reruns of suite-timeout locations passed: - - `mix test test/familiar_behavior_test.exs:108` - - `mix test test/familiar_behavior_test.exs:159` - - `mix test test/familiar_behavior_test.exs:403` +Result: 3 properties, 653 tests, 0 failures. -Full-suite status: +## Evidence -- `mix test` failed once with a Mnesia backend timeout in `test/loom_backend_symmetry_test.exs:43`. -- `mix test --max-cases 1` later cascaded into Familiar behavior timeouts and was stopped after the third timeout. -- Those failures were not fixed here because the affected tests pass in isolation and are outside the port-child `:epipe` hardening scope. +- Familiar prompt no longer says memory is only prior turns. It teaches the real split: prompts in `loom.intents`, entity actions in `loom.turns`, and `Cantrip.Loom.transcript(loom)` as the chronological conversation view with a `you:`/`me:` example: `lib/cantrip/familiar.ex:43`. +- The prompt teaches bounded inspection helpers for raw code/bash turns whose `:code_state` carries a full binding snapshot: `lib/cantrip/familiar.ex:65`. +- The later Familiar environment paragraph repeats the same concrete affordance shape instead of the old bare `loom.turns` promise: `lib/cantrip/familiar.ex:211`. +- Code-medium capability text now teaches `loom.intents`, `loom.turns`, `Cantrip.Loom.transcript(loom)`, and the bounded projection helpers: `lib/cantrip/medium/code.ex:593`. +- `Cantrip.Loom.bounded_turn/2` is a pure view: it removes `:code_state` from the returned map and adds `:code_state_summary`; `bounded_turns/2` and `bounded_transcript/2` apply that projection without mutating the durable loom: `lib/cantrip/loom.ex:310`. +- The bounded projection test asserts the raw durable turn still has `:code_state`, while `bounded_turn/2`, `bounded_turns/2`, and `bounded_transcript/2` do not embed `:code_state`: `test/loom_api_test.exs:173`. +- Prompt/capability tests pin the taught affordances so they do not regress back to only `loom.turns`: `test/familiar_test.exs:146`. +- Existing ground truth was verified and left intact: `append_intent/3` stores intents as `role: "intent"` with `utterance: %{content: text}` (`lib/cantrip/loom.ex:255`), and `transcript/1` iterates the event log in append order (`lib/cantrip/loom.ex:273`). +- The stale Mnesia ordering bug was not re-fixed. Current storage writes monotonic integer keys (`lib/cantrip/loom/storage/mnesia.ex:34`) and reads them sorted by numeric key (`lib/cantrip/loom/storage/mnesia.ex:130`). ## Discovered Tickets -- `can-nlga`: Mnesia backend symmetry test can time out in full suite. -- `can-petw`: Familiar behavior tests can cascade time out in full-suite runs. +None. ## Uncertainty -I did not find original incident logs from 2026-07-05 in this workspace, so the exact production trigger remains inferred. The reproduced failure matches the requested root sequence: parent port closes, child writes to it, `IO.binwrite/2` raises `:epipe`, and the linked protocol process can take down the child BEAM. The fix hardens that sequence directly. +None. From a09c76e04b3be636f7642be2f374d9c6dd4655f6 Mon Sep 17 00:00:00 2001 From: deepfates Date: Mon, 6 Jul 2026 20:54:04 -0700 Subject: [PATCH 05/10] fix: surface child cast results in renderer --- REPORT.md | 40 +++++++++------- lib/cantrip.ex | 72 ++++++++++++++++++++++++---- lib/cantrip/cli/renderer.ex | 95 +++++++++++++++++++++++++++++++++++-- lib/cantrip/event.ex | 7 ++- test/cli/renderer_test.exs | 82 +++++++++++++++++++++++++++++++- 5 files changed, 262 insertions(+), 34 deletions(-) diff --git a/REPORT.md b/REPORT.md index cbc5896..d64c217 100644 --- a/REPORT.md +++ b/REPORT.md @@ -1,36 +1,42 @@ -# dee-perc Report +# dee-kids Report -Branch: `codex/dee-perc-iex-legibility` - -Code commit: `ad7b6f1 fix: teach loom transcript and bound turn inspection` +Branch: `child-cast-legibility` ## Reproduction Commands From the repository root: ```sh -mix test test/loom_api_test.exs test/loom_intent_persistence_test.exs test/familiar_test.exs test/code_medium_ergonomics_test.exs +mix test test/cli/renderer_test.exs +``` + +Result: 20 tests, 0 failures. + +```sh +mix test test/cli/renderer_test.exs test/entity_server_stream_test.exs test/composition_test.exs +``` + +Result: 41 tests, 0 failures. + +```sh +mix test test/acp_event_bridge_test.exs test/streaming_test.exs ``` -Result: 71 tests, 0 failures. +Result: 36 tests, 0 failures. ```sh -mix test +mix test test/familiar_test.exs ``` -Result: 3 properties, 653 tests, 0 failures. +Result: 33 tests, 0 failures. ## Evidence -- Familiar prompt no longer says memory is only prior turns. It teaches the real split: prompts in `loom.intents`, entity actions in `loom.turns`, and `Cantrip.Loom.transcript(loom)` as the chronological conversation view with a `you:`/`me:` example: `lib/cantrip/familiar.ex:43`. -- The prompt teaches bounded inspection helpers for raw code/bash turns whose `:code_state` carries a full binding snapshot: `lib/cantrip/familiar.ex:65`. -- The later Familiar environment paragraph repeats the same concrete affordance shape instead of the old bare `loom.turns` promise: `lib/cantrip/familiar.ex:211`. -- Code-medium capability text now teaches `loom.intents`, `loom.turns`, `Cantrip.Loom.transcript(loom)`, and the bounded projection helpers: `lib/cantrip/medium/code.ex:593`. -- `Cantrip.Loom.bounded_turn/2` is a pure view: it removes `:code_state` from the returned map and adds `:code_state_summary`; `bounded_turns/2` and `bounded_transcript/2` apply that projection without mutating the durable loom: `lib/cantrip/loom.ex:310`. -- The bounded projection test asserts the raw durable turn still has `:code_state`, while `bounded_turn/2`, `bounded_turns/2`, and `bounded_transcript/2` do not embed `:code_state`: `test/loom_api_test.exs:173`. -- Prompt/capability tests pin the taught affordances so they do not regress back to only `loom.turns`: `test/familiar_test.exs:146`. -- Existing ground truth was verified and left intact: `append_intent/3` stores intents as `role: "intent"` with `utterance: %{content: text}` (`lib/cantrip/loom.ex:255`), and `transcript/1` iterates the event log in append order (`lib/cantrip/loom.ex:273`). -- The stale Mnesia ordering bug was not re-fixed. Current storage writes monotonic integer keys (`lib/cantrip/loom/storage/mnesia.ex:34`) and reads them sorted by numeric key (`lib/cantrip/loom/storage/mnesia.ex:130`). +- Child `final_response` events at non-root depth now render as `child done: ` on stderr instead of being suppressed: `lib/cantrip/cli/renderer.ex`. +- Child cast start/end lines now label the child cantrip id, circle type, and batch index when available; child errors render as `child error` with a red failure marker: `lib/cantrip/cli/renderer.ex`. +- `cast` and `cast_batch` tool results get child-specific rendering and show child-turn counts from observations: `lib/cantrip/cli/renderer.ex`, `lib/cantrip/event.ex`. +- Child cast stream metadata now includes `child_id`, `circle`, `node`, and `batch_index` where the parent already knows them; public cast return shapes and loom grafting are unchanged: `lib/cantrip.ex`. +- Renderer regression tests cover visible child `done(value)`, labeled child starts, loud child errors, child-turn counts, and cast-batch error rendering: `test/cli/renderer_test.exs`. ## Discovered Tickets diff --git a/lib/cantrip.ex b/lib/cantrip.ex index 584a3eb..c2b309b 100644 --- a/lib/cantrip.ex +++ b/lib/cantrip.ex @@ -581,11 +581,14 @@ defmodule Cantrip do {:ok, normalized_items} -> payloads = normalized_items + |> Enum.with_index() |> Task.async_stream( - fn %{cantrip: cantrip, intent: intent} -> + fn {%{cantrip: cantrip, intent: intent}, index} -> cast(cantrip, intent, parent_context: parent_context, - record_parent_observation?: false + record_parent_observation?: false, + parent_gate: "cast_batch", + batch_index: index ) end, ordered: true, @@ -858,7 +861,8 @@ defmodule Cantrip do entity_state = Map.get(parent_context, :entity_state) record_observation? = Keyword.get(opts, :record_parent_observation?, true) parent_gate = Keyword.get(opts, :parent_gate, "cast") - opts = Keyword.drop(opts, [:record_parent_observation?, :parent_gate]) + batch_index = Keyword.get(opts, :batch_index) + opts = Keyword.drop(opts, [:record_parent_observation?, :parent_gate, :batch_index]) case prepare_child_cast(cantrip, parent_context) do {:ok, transient_cantrip, depth} -> @@ -870,7 +874,11 @@ defmodule Cantrip do emit_parent_event( entity_state, - {:child_start, %{depth: depth, intent: intent, node: node}} + {:child_start, + child_event_meta(transient_cantrip, depth, intent, + node: node, + batch_index: batch_index + )} ) emit_child_start_telemetry(parent_context, depth) @@ -881,7 +889,12 @@ defmodule Cantrip do emit_parent_event( entity_state, - {:child_end, %{depth: depth, result: value, node: node}} + {:child_end, + child_event_meta(transient_cantrip, depth, nil, + result: value, + node: node, + batch_index: batch_index + )} ) emit_child_stop_telemetry(parent_context, depth, :ok) @@ -903,7 +916,12 @@ defmodule Cantrip do emit_parent_event( entity_state, - {:child_end, %{depth: depth, error: Cantrip.SafeFormat.inspect(reason), node: node}} + {:child_end, + child_event_meta(transient_cantrip, depth, nil, + error: Cantrip.SafeFormat.inspect(reason), + node: node, + batch_index: batch_index + )} ) emit_child_stop_telemetry(parent_context, depth, :error) @@ -941,7 +959,8 @@ defmodule Cantrip do entity_state = Map.get(parent_context, :entity_state) record_observation? = Keyword.get(opts, :record_parent_observation?, true) parent_gate = Keyword.get(opts, :parent_gate, "cast") - opts = Keyword.drop(opts, [:record_parent_observation?, :parent_gate]) + batch_index = Keyword.get(opts, :batch_index) + opts = Keyword.drop(opts, [:record_parent_observation?, :parent_gate, :batch_index]) case prepare_child_cast(cantrip, parent_context) do {:ok, transient_cantrip, depth} -> @@ -955,14 +974,28 @@ defmodule Cantrip do |> maybe_put_new(:stream_to, Map.get(parent_context, :stream_to)) |> maybe_put_new(:stream_barrier?, Map.get(parent_context, :stream_barrier?)) - emit_parent_event(entity_state, {:child_start, %{depth: depth, intent: intent}}) + emit_parent_event( + entity_state, + {:child_start, + child_event_meta(transient_cantrip, depth, intent, batch_index: batch_index)} + ) + emit_child_start_telemetry(parent_context, depth) case run_cast(transient_cantrip, intent, cast_opts) do {:ok, value, next_cantrip, child_loom, meta} -> next_cantrip = restore_child_declared_wards(cantrip, next_cantrip) remember_parent_child_llm(parent_context, next_cantrip) - emit_parent_event(entity_state, {:child_end, %{depth: depth, result: value}}) + + emit_parent_event( + entity_state, + {:child_end, + child_event_meta(transient_cantrip, depth, nil, + result: value, + batch_index: batch_index + )} + ) + emit_child_stop_telemetry(parent_context, depth, :ok) if record_observation?, @@ -983,7 +1016,11 @@ defmodule Cantrip do emit_parent_event( entity_state, - {:child_end, %{depth: depth, error: Cantrip.SafeFormat.inspect(reason)}} + {:child_end, + child_event_meta(transient_cantrip, depth, nil, + error: Cantrip.SafeFormat.inspect(reason), + batch_index: batch_index + )} ) emit_child_stop_telemetry(parent_context, depth, :error) @@ -1257,6 +1294,21 @@ defmodule Cantrip do defp maybe_put_new(opts, _key, nil), do: opts defp maybe_put_new(opts, key, value), do: Keyword.put_new(opts, key, value) + defp child_event_meta(%__MODULE__{} = cantrip, depth, intent, extras) do + %{ + depth: depth, + intent: intent, + child_id: cantrip.id, + circle: cantrip.circle.type + } + |> drop_nil_values() + |> Map.merge(extras |> Map.new() |> drop_nil_values()) + end + + defp drop_nil_values(map) do + Map.reject(map, fn {_key, value} -> is_nil(value) end) + end + defp normalize_parent_context(%{} = context) do Map.new(context, fn {k, v} -> key = diff --git a/lib/cantrip/cli/renderer.ex b/lib/cantrip/cli/renderer.ex index 5947ace..777d8dd 100644 --- a/lib/cantrip/cli/renderer.ex +++ b/lib/cantrip/cli/renderer.ex @@ -77,6 +77,13 @@ defmodule Cantrip.CLI.Renderer do {[indent(d, line), "\n"], :stderr, state} end + def render_event(state, {%{depth: d}, {:tool_call, %{gate: gate} = meta}}) + when gate in ["cast", "cast_batch"] do + label = child_tool_label(gate, meta) + line = [" ", Owl.Data.tag("▸ ", :magenta) |> Owl.Data.to_chardata(), label] + {[indent(d, line), "\n"], :stderr, state} + end + def render_event(state, {%{depth: d}, {:tool_call, %{gate: gate} = meta}}) do label = case meta[:args_summary] do @@ -88,6 +95,35 @@ defmodule Cantrip.CLI.Renderer do {[indent(d, line), "\n"], :stderr, state} end + def render_event( + state, + {%{depth: d}, {:tool_result, %{gate: gate, result: result, is_error: true} = meta}} + ) + when gate in ["cast", "cast_batch"] do + text = summarize(result) + + line = + Owl.Data.tag([" ✗ ", child_tool_label(gate, meta), ": ", text], :red) + |> Owl.Data.to_chardata() + + {[indent(d, line), "\n"], :stderr, state} + end + + def render_event( + state, + {%{depth: d}, {:tool_result, %{gate: gate, result: result, is_error: false} = meta}} + ) + when gate in ["cast", "cast_batch"] do + text = summarize(result) + child_turns = child_turn_count(meta) + + line = + Owl.Data.tag([" ✓ ", child_tool_label(gate, meta), ": ", text, child_turns], :green) + |> Owl.Data.to_chardata() + + {[indent(d, line), "\n"], :stderr, state} + end + def render_event( state, {%{depth: d}, {:tool_result, %{gate: gate, result: result, is_error: true}}} @@ -125,15 +161,24 @@ defmodule Cantrip.CLI.Renderer do {[result_str, "\n"], :stdout, state} end + def render_event(state, {%{depth: d}, {:final_response, %{result: result}}}) when d > 0 do + line = + Owl.Data.tag([" ✓ child done: ", summarize(result)], :green) + |> Owl.Data.to_chardata() + + {[indent(d - 1, line), "\n"], :stderr, state} + end + def render_event(state, {_, {:final_response, _}}), do: {"", :stderr, state} # -- Child delegation -- - def render_event(state, {%{depth: d}, {:child_start, %{intent: intent}}}) do + def render_event(state, {%{depth: d}, {:child_start, %{intent: intent} = meta}}) do line = [ " ", Owl.Data.tag("▸ ", :magenta) |> Owl.Data.to_chardata(), - "cast: \"", + child_start_label(meta), + ": \"", to_string(intent), "\"" ] @@ -147,12 +192,14 @@ defmodule Cantrip.CLI.Renderer do end def render_event(state, {%{depth: d}, {:child_end, %{error: err}}}) do - line = Owl.Data.tag([" ✗ cast: ", to_string(err)], :red) |> Owl.Data.to_chardata() + line = Owl.Data.tag([" ✗ child error: ", to_string(err)], :red) |> Owl.Data.to_chardata() {[indent(d, line), "\n"], :stderr, state} end def render_event(state, {%{depth: d}, {:child_end, %{result: result}}}) do - line = Owl.Data.tag([" ✓ cast: ", summarize(result)], :green) |> Owl.Data.to_chardata() + line = + Owl.Data.tag([" ✓ child result: ", summarize(result)], :green) |> Owl.Data.to_chardata() + {[indent(d, line), "\n"], :stderr, state} end @@ -175,6 +222,46 @@ defmodule Cantrip.CLI.Renderer do defp prefix(depth), do: String.duplicate(" ", depth) + defp child_start_label(meta) do + ["cast", child_index(meta), child_subject(meta), child_circle(meta)] + |> Enum.reject(&(&1 in [nil, ""])) + |> Enum.join(" ") + end + + defp child_tool_label(gate, meta) do + [gate, child_index(meta), child_subject(meta), child_circle(meta)] + |> Enum.reject(&(&1 in [nil, ""])) + |> Enum.join(" ") + end + + defp child_index(%{index: index}) when is_integer(index), do: "##{index + 1}" + defp child_index(%{batch_index: index}) when is_integer(index), do: "##{index + 1}" + defp child_index(_meta), do: nil + + defp child_subject(%{child_id: id}) when is_binary(id), do: id + defp child_subject(%{cantrip_id: id}) when is_binary(id), do: id + defp child_subject(_meta), do: nil + + defp child_circle(%{circle: circle}) when is_atom(circle), do: "(#{circle})" + defp child_circle(%{circle: circle}) when is_binary(circle), do: "(#{circle})" + defp child_circle(%{medium: medium}) when is_atom(medium), do: "(#{medium})" + defp child_circle(%{medium: medium}) when is_binary(medium), do: "(#{medium})" + defp child_circle(_meta), do: nil + + defp child_turn_count(%{child_turn_count: count}) when is_integer(count) do + " (#{count} child turn#{plural(count)})" + end + + defp child_turn_count(%{child_turns: turns}) when is_list(turns) do + count = length(turns) + " (#{count} child turn#{plural(count)})" + end + + defp child_turn_count(_meta), do: "" + + defp plural(1), do: "" + defp plural(_), do: "s" + # ── Result summarization ───────────────────────────────────────────── @max_display 300 diff --git a/lib/cantrip/event.ex b/lib/cantrip/event.ex index 55b77f8..9f5103f 100644 --- a/lib/cantrip/event.ex +++ b/lib/cantrip/event.ex @@ -72,7 +72,8 @@ defmodule Cantrip.Event do gate: obs.gate, result: obs.result, is_error: obs.is_error, - tool_call_id: tool_call_id + tool_call_id: tool_call_id, + child_turn_count: child_turn_count(obs) }} ] end) @@ -184,4 +185,8 @@ defmodule Cantrip.Event do defp args_summary("search", %{} = a), do: Map.get(a, "pattern", Map.get(a, :pattern)) defp args_summary("mix", %{} = a), do: Map.get(a, "task", Map.get(a, :task)) defp args_summary(_, _), do: nil + + defp child_turn_count(%{child_turns: turns}) when is_list(turns), do: length(turns) + defp child_turn_count(%{"child_turns" => turns}) when is_list(turns), do: length(turns) + defp child_turn_count(_obs), do: 0 end diff --git a/test/cli/renderer_test.exs b/test/cli/renderer_test.exs index c821f95..f06175d 100644 --- a/test/cli/renderer_test.exs +++ b/test/cli/renderer_test.exs @@ -116,14 +116,16 @@ defmodule Cantrip.CLI.RendererTest do assert IO.iodata_to_binary(output) =~ "The answer is 42" end - test "final_response at depth > 0 is suppressed" do + test "final_response at depth > 0 renders child done result on stderr" do state = Renderer.new() {output, device, _} = Renderer.render_event(state, {env(1), {:final_response, %{result: "child result"}}}) assert device == :stderr - assert IO.iodata_to_binary(output) == "" + text = IO.iodata_to_binary(output) + assert text =~ "child done" + assert text =~ "child result" end test "final_response inspects non-string results" do @@ -150,6 +152,82 @@ defmodule Cantrip.CLI.RendererTest do {output, _, _} = Renderer.render_event(state, {:unknown_event, %{}}) assert IO.iodata_to_binary(output) == "" end + + test "child_start identifies child, circle, batch index, and intent" do + state = Renderer.new() + + event = + {env(), + {:child_start, + %{ + depth: 1, + intent: "read notes.md", + child_id: "cantrip_child", + circle: :conversation, + batch_index: 0 + }}} + + {output, device, _} = Renderer.render_event(state, event) + + assert device == :stderr + text = IO.iodata_to_binary(output) + assert text =~ "cast #1 cantrip_child (conversation)" + assert text =~ ~s|"read notes.md"| + end + + test "child_end error is loud" do + state = Renderer.new() + + {output, device, _} = + Renderer.render_event(state, {env(1), {:child_end, %{error: "boom"}}}) + + assert device == :stderr + text = IO.iodata_to_binary(output) + assert text =~ "✗" + assert text =~ "child error" + assert text =~ "boom" + end + + test "cast tool_result shows child turn count" do + state = Renderer.new() + + event = + {env(), + {:tool_result, + %{ + gate: "cast", + result: "child answer", + is_error: false, + child_turn_count: 2, + child_id: "cantrip_child", + circle: :code + }}} + + {output, device, _} = Renderer.render_event(state, event) + + assert device == :stderr + text = IO.iodata_to_binary(output) + assert text =~ "cast cantrip_child (code)" + assert text =~ "child answer" + assert text =~ "2 child turns" + end + + test "cast_batch tool_result errors render as child failures" do + state = Renderer.new() + + event = + {env(), + {:tool_result, + %{gate: "cast_batch", result: "max_depth exceeded", is_error: true, batch_index: 1}}} + + {output, device, _} = Renderer.render_event(state, event) + + assert device == :stderr + text = IO.iodata_to_binary(output) + assert text =~ "✗" + assert text =~ "cast_batch #2" + assert text =~ "max_depth exceeded" + end end describe "depth indentation from envelope" do From 08754a1090e48319a6bd36938463e34d8faa16ed Mon Sep 17 00:00:00 2001 From: deepfates Date: Mon, 6 Jul 2026 21:06:01 -0700 Subject: [PATCH 06/10] fix: attribute child cast tool results --- REPORT.md | 30 ++++---- lib/cantrip.ex | 90 +++++++++++++++++++++--- lib/cantrip/cli/renderer.ex | 22 +++++- lib/cantrip/event.ex | 14 +++- lib/cantrip/medium/code/port.ex | 114 ++++++++++++++++++++++-------- test/cli/renderer_test.exs | 118 ++++++++++++++++++++++++++++++++ 6 files changed, 326 insertions(+), 62 deletions(-) diff --git a/REPORT.md b/REPORT.md index d64c217..0dda3df 100644 --- a/REPORT.md +++ b/REPORT.md @@ -1,4 +1,4 @@ -# dee-kids Report +# dee-kids Bounce Report Branch: `child-cast-legibility` @@ -10,33 +10,27 @@ From the repository root: mix test test/cli/renderer_test.exs ``` -Result: 20 tests, 0 failures. +Result: 22 tests, 0 failures. ```sh -mix test test/cli/renderer_test.exs test/entity_server_stream_test.exs test/composition_test.exs -``` - -Result: 41 tests, 0 failures. - -```sh -mix test test/acp_event_bridge_test.exs test/streaming_test.exs +mix test test/familiar_test.exs ``` -Result: 36 tests, 0 failures. +Result: 33 tests, 0 failures. ```sh -mix test test/familiar_test.exs +mix test test/composition_test.exs ``` -Result: 33 tests, 0 failures. +Result: 13 tests, 0 failures. ## Evidence -- Child `final_response` events at non-root depth now render as `child done: ` on stderr instead of being suppressed: `lib/cantrip/cli/renderer.ex`. -- Child cast start/end lines now label the child cantrip id, circle type, and batch index when available; child errors render as `child error` with a red failure marker: `lib/cantrip/cli/renderer.ex`. -- `cast` and `cast_batch` tool results get child-specific rendering and show child-turn counts from observations: `lib/cantrip/cli/renderer.ex`, `lib/cantrip/event.ex`. -- Child cast stream metadata now includes `child_id`, `circle`, `node`, and `batch_index` where the parent already knows them; public cast return shapes and loom grafting are unchanged: `lib/cantrip.ex`. -- Renderer regression tests cover visible child `done(value)`, labeled child starts, loud child errors, child-turn counts, and cast-batch error rendering: `test/cli/renderer_test.exs`. +- Real `cast` parent observations now include `child_id`, `circle`, `node`, and `batch_index` when the parent knows them, and `Cantrip.Event.tool_events/1` propagates those fields into real `:tool_call` and `:tool_result` events: `lib/cantrip.ex`, `lib/cantrip/event.ex`. +- Real `cast_batch` parent observations now include a `children` attribution list. Single-child batches also expose top-level `child_id`/`circle`; multi-child batches render the honest aggregate label from the child list: `lib/cantrip.ex`, `lib/cantrip/cli/renderer.ex`. +- The code-medium port proxy was also patched because it synthesizes parent observations for proxied `Cantrip.cast/2` and `Cantrip.cast_batch/1` calls: `lib/cantrip/medium/code/port.ex`. +- Renderer regressions now include live runtime probes that call `Cantrip.cast(..., stream_to: self())`, assert the emitted real `:tool_result` metadata, then render that exact event: `test/cli/renderer_test.exs`. +- `can-sgch` was addressed with a note and remains open for coordinator handling. ## Discovered Tickets @@ -44,4 +38,4 @@ None. ## Uncertainty -None. +I did not run a provider-backed live LLM. The new probes exercise the real Cantrip runtime and streaming event path with deterministic `Cantrip.FakeLLM`. diff --git a/lib/cantrip.ex b/lib/cantrip.ex index c2b309b..d8c1894 100644 --- a/lib/cantrip.ex +++ b/lib/cantrip.ex @@ -611,7 +611,8 @@ defmodule Cantrip do "cast_batch", Cantrip.SafeFormat.inspect(reason), true, - [] + [], + cast_batch_observation_meta(normalized_items, payloads) ) {:error, reason} @@ -620,7 +621,16 @@ defmodule Cantrip do next_cantrips = Enum.map(payloads, fn {:ok, _value, next, _loom, _meta} -> next end) looms = Enum.map(payloads, fn {:ok, _value, _next, loom, _meta} -> loom end) child_turns = Enum.flat_map(looms, & &1.turns) - push_parent_cast_observation(parent_context, "cast_batch", values, false, child_turns) + + push_parent_cast_observation( + parent_context, + "cast_batch", + values, + false, + child_turns, + cast_batch_observation_meta(normalized_items, payloads) + ) + {:ok, values, next_cantrips, looms, %{count: length(values)}} end @@ -630,7 +640,8 @@ defmodule Cantrip do "cast_batch", Cantrip.SafeFormat.inspect(reason), true, - [] + [], + cast_batch_observation_meta(items, nil) ) {:error, reason} @@ -906,7 +917,11 @@ defmodule Cantrip do parent_gate, value, false, - child_loom.turns + child_loom.turns, + child_observation_meta(transient_cantrip, + node: node, + batch_index: batch_index + ) ) {:ok, value, next_cantrip, child_loom, meta} @@ -933,7 +948,11 @@ defmodule Cantrip do parent_gate, Cantrip.SafeFormat.inspect(reason), true, - [] + [], + child_observation_meta(transient_cantrip, + node: node, + batch_index: batch_index + ) ) {:error, reason, %{next_cantrip | node: node}} @@ -947,7 +966,8 @@ defmodule Cantrip do parent_gate, Cantrip.SafeFormat.inspect(reason), true, - [] + [], + child_observation_meta(cantrip, node: node, batch_index: batch_index) ) {:error, reason, next_cantrip} @@ -1005,7 +1025,8 @@ defmodule Cantrip do parent_gate, value, false, - child_loom.turns + child_loom.turns, + child_observation_meta(transient_cantrip, batch_index: batch_index) ) {:ok, value, next_cantrip, child_loom, meta} @@ -1032,7 +1053,8 @@ defmodule Cantrip do parent_gate, Cantrip.SafeFormat.inspect(reason), true, - [] + [], + child_observation_meta(transient_cantrip, batch_index: batch_index) ) {:error, reason, next_cantrip} @@ -1046,7 +1068,8 @@ defmodule Cantrip do parent_gate, Cantrip.SafeFormat.inspect(reason), true, - [] + [], + child_observation_meta(cantrip, batch_index: batch_index) ) error @@ -1404,10 +1427,20 @@ defmodule Cantrip do end end - defp push_parent_cast_observation(parent_context, gate, result, is_error, child_turns) do + defp push_parent_cast_observation( + parent_context, + gate, + result, + is_error, + child_turns, + metadata + ) do case parent_context && Map.get(parent_context, :observation_collector) do collector when is_pid(collector) -> - observation = %{gate: gate, result: result, is_error: is_error, child_turns: child_turns} + observation = + %{gate: gate, result: result, is_error: is_error, child_turns: child_turns} + |> Map.merge(metadata |> Map.new() |> drop_nil_values()) + Agent.update(collector, &(&1 ++ [observation])) _ -> @@ -1415,6 +1448,41 @@ defmodule Cantrip do end end + defp child_observation_meta(%__MODULE__{} = cantrip, extras) do + %{ + child_id: cantrip.id, + circle: cantrip.circle.type + } + |> Map.merge(extras |> Map.new() |> drop_nil_values()) + end + + defp cast_batch_observation_meta(items, payloads) when is_list(items) do + children = + items + |> Enum.with_index() + |> Enum.flat_map(fn + {%{cantrip: %__MODULE__{} = cantrip}, index} -> + [child_observation_meta(cantrip, batch_index: index)] + + {_item, _index} -> + [] + end) + + failed_batch_index = + if is_list(payloads) do + Enum.find_index(payloads, &match?({:error, _, _}, &1)) + end + + base = + %{children: children, failed_batch_index: failed_batch_index} + |> drop_nil_values() + + case children do + [single] -> Map.merge(single, base) + _ -> base + end + end + defp messages_from_turns(turns, call) do prefix = if is_nil(call.system_prompt), diff --git a/lib/cantrip/cli/renderer.ex b/lib/cantrip/cli/renderer.ex index 777d8dd..6b2a962 100644 --- a/lib/cantrip/cli/renderer.ex +++ b/lib/cantrip/cli/renderer.ex @@ -229,7 +229,7 @@ defmodule Cantrip.CLI.Renderer do end defp child_tool_label(gate, meta) do - [gate, child_index(meta), child_subject(meta), child_circle(meta)] + [gate, child_index(meta), child_subject(meta) || child_children(meta), child_circle(meta)] |> Enum.reject(&(&1 in [nil, ""])) |> Enum.join(" ") end @@ -242,6 +242,26 @@ defmodule Cantrip.CLI.Renderer do defp child_subject(%{cantrip_id: id}) when is_binary(id), do: id defp child_subject(_meta), do: nil + defp child_children(%{children: children}) when is_list(children) do + children + |> Enum.map(&child_descriptor/1) + |> Enum.reject(&(&1 in [nil, ""])) + |> case do + [] -> nil + labels -> Enum.join(labels, ", ") + end + end + + defp child_children(_meta), do: nil + + defp child_descriptor(%{} = child) do + [child_subject(child), child_circle(child)] + |> Enum.reject(&(&1 in [nil, ""])) + |> Enum.join(" ") + end + + defp child_descriptor(_child), do: nil + defp child_circle(%{circle: circle}) when is_atom(circle), do: "(#{circle})" defp child_circle(%{circle: circle}) when is_binary(circle), do: "(#{circle})" defp child_circle(%{medium: medium}) when is_atom(medium), do: "(#{medium})" diff --git a/lib/cantrip/event.ex b/lib/cantrip/event.ex index 9f5103f..72c8830 100644 --- a/lib/cantrip/event.ex +++ b/lib/cantrip/event.ex @@ -59,6 +59,8 @@ defmodule Cantrip.Event do Enum.flat_map(observations, fn obs -> tool_call_id = obs[:tool_call_id] || mint_tool_call_id() + attribution = child_attribution(obs) + [ {:tool_call, %{ @@ -66,7 +68,8 @@ defmodule Cantrip.Event do tool_call_id: tool_call_id, kind: gate_kind(obs.gate), args_summary: args_summary(obs.gate, obs[:args]) - }}, + } + |> Map.merge(attribution)}, {:tool_result, %{ gate: obs.gate, @@ -74,7 +77,8 @@ defmodule Cantrip.Event do is_error: obs.is_error, tool_call_id: tool_call_id, child_turn_count: child_turn_count(obs) - }} + } + |> Map.merge(attribution)} ] end) end @@ -186,6 +190,12 @@ defmodule Cantrip.Event do defp args_summary("mix", %{} = a), do: Map.get(a, "task", Map.get(a, :task)) defp args_summary(_, _), do: nil + defp child_attribution(obs) do + obs + |> Map.take([:child_id, :circle, :node, :batch_index, :children, :failed_batch_index]) + |> Map.reject(fn {_key, value} -> is_nil(value) or value == [] end) + end + defp child_turn_count(%{child_turns: turns}) when is_list(turns), do: length(turns) defp child_turn_count(%{"child_turns" => turns}) when is_list(turns), do: length(turns) defp child_turn_count(_obs), do: 0 diff --git a/lib/cantrip/medium/code/port.ex b/lib/cantrip/medium/code/port.ex index 5987cc3..4f068eb 100644 --- a/lib/cantrip/medium/code/port.ex +++ b/lib/cantrip/medium/code/port.ex @@ -325,18 +325,24 @@ defmodule Cantrip.Medium.Code.Port do |> Keyword.put(:record_parent_observation?, false), {:ok, value, next_cantrip, loom, meta} <- Cantrip.cast(cantrip, intent, cast_opts) do {next_handle, state} = put_child_handle(state, next_cantrip, handle) - observation = %{gate: "cast", result: value, is_error: false, child_turns: loom.turns} + + observation = + %{gate: "cast", result: value, is_error: false, child_turns: loom.turns} + |> Map.merge(child_observation_meta(cantrip, [])) + {{:ok, value, next_handle, loom, meta}, state, [observation]} else {:error, reason, next_cantrip} -> {next_handle, state} = put_child_handle(state, next_cantrip, handle) - observation = %{ - gate: "cast", - result: Cantrip.SafeFormat.inspect(reason), - is_error: true, - child_turns: [] - } + observation = + %{ + gate: "cast", + result: Cantrip.SafeFormat.inspect(reason), + is_error: true, + child_turns: [] + } + |> Map.merge(child_observation_meta(next_cantrip, [])) {{:error, reason, next_handle}, state, [observation]} @@ -353,31 +359,48 @@ defmodule Cantrip.Medium.Code.Port do with {:ok, normalized_items} <- resolve_batch_items(state, items), opts <- normalize_opts(opts), parent_context <- Map.get(runtime, :parent_context), - batch_opts = Keyword.put(opts, :parent_context, parent_context), - {:ok, values, next_cantrips, looms, meta} <- - Cantrip.cast_batch(normalized_items, batch_opts) do - {handles, state} = - Enum.zip(normalized_items, next_cantrips) - |> Enum.map_reduce(state, fn {%{handle: old_handle}, next_cantrip}, acc -> - put_child_handle(acc, next_cantrip, old_handle) - end) - - observation = %{ - gate: "cast_batch", - result: values, - is_error: false, - child_turns: Enum.flat_map(looms, & &1.turns) - } - - {{:ok, values, handles, looms, meta}, state, [observation]} + batch_opts = Keyword.put(opts, :parent_context, parent_context) do + case Cantrip.cast_batch(normalized_items, batch_opts) do + {:ok, values, next_cantrips, looms, meta} -> + {handles, state} = + Enum.zip(normalized_items, next_cantrips) + |> Enum.map_reduce(state, fn {%{handle: old_handle}, next_cantrip}, acc -> + put_child_handle(acc, next_cantrip, old_handle) + end) + + observation = + %{ + gate: "cast_batch", + result: values, + is_error: false, + child_turns: Enum.flat_map(looms, & &1.turns) + } + |> Map.merge(cast_batch_observation_meta(normalized_items)) + + {{:ok, values, handles, looms, meta}, state, [observation]} + + {:error, reason} -> + observation = + %{ + gate: "cast_batch", + result: Cantrip.SafeFormat.inspect(reason), + is_error: true, + child_turns: [] + } + |> Map.merge(cast_batch_observation_meta(normalized_items)) + + {{:error, reason}, state, [observation]} + end else {:error, reason} -> - observation = %{ - gate: "cast_batch", - result: Cantrip.SafeFormat.inspect(reason), - is_error: true, - child_turns: [] - } + observation = + %{ + gate: "cast_batch", + result: Cantrip.SafeFormat.inspect(reason), + is_error: true, + child_turns: [] + } + |> Map.merge(cast_batch_observation_meta([])) {{:error, reason}, state, [observation]} end @@ -474,6 +497,37 @@ defmodule Cantrip.Medium.Code.Port do ArgumentError -> value end + defp child_observation_meta(%Cantrip{} = cantrip, extras) do + %{ + child_id: cantrip.id, + circle: cantrip.circle.type + } + |> Map.merge(extras |> Map.new() |> drop_nil_values()) + end + + defp cast_batch_observation_meta(items) when is_list(items) do + children = + items + |> Enum.with_index() + |> Enum.flat_map(fn + {%{cantrip: %Cantrip{} = cantrip}, index} -> + [child_observation_meta(cantrip, batch_index: index)] + + {_item, _index} -> + [] + end) + + case children do + [] -> %{} + [single] -> Map.merge(single, %{children: children}) + _ -> %{children: children} + end + end + + defp drop_nil_values(map) do + Map.reject(map, fn {_key, value} -> is_nil(value) end) + end + defp sanitize_observation(observation) when is_map(observation) do observation |> redact_observation_field(:args) diff --git a/test/cli/renderer_test.exs b/test/cli/renderer_test.exs index f06175d..cdffa3e 100644 --- a/test/cli/renderer_test.exs +++ b/test/cli/renderer_test.exs @@ -1,6 +1,7 @@ defmodule Cantrip.CLI.RendererTest do use ExUnit.Case, async: true + alias Cantrip.FakeLLM alias Cantrip.CLI.Renderer # Helper to wrap events in an envelope @@ -228,6 +229,123 @@ defmodule Cantrip.CLI.RendererTest do assert text =~ "cast_batch #2" assert text =~ "max_depth exceeded" end + + test "real cast tool_result carries child attribution and renders it" do + child_llm = {FakeLLM, FakeLLM.new([%{code: ~s[done.("child answer")]}])} + + parent_llm = + {FakeLLM, + FakeLLM.new([ + %{ + code: """ + {:ok, child} = Cantrip.new(circle: %{type: :code, gates: [:done]}) + {:ok, result, _child, _loom, _meta} = Cantrip.cast(child, "work") + done.(result) + """ + } + ])} + + {:ok, parent} = + Cantrip.new( + llm: parent_llm, + child_llm: child_llm, + circle: %{ + type: :code, + gates: [:done], + wards: [%{max_turns: 5}, %{max_depth: 1}, %{sandbox: :unrestricted}] + } + ) + + assert {:ok, "child answer", _parent, _loom, _meta} = + Cantrip.cast(parent, "delegate", stream_to: self()) + + events = collect_stream_events() + + assert {env, {:tool_result, meta}} = + Enum.find(events, fn + {_env, {:tool_result, %{gate: "cast"}}} -> true + _event -> false + end) + + assert env.depth == 0 + assert is_binary(meta.child_id) + assert meta.circle == :code + assert meta.child_turn_count == 1 + + {output, :stderr, _state} = + Renderer.render_event(Renderer.new(), {env, {:tool_result, meta}}) + + text = IO.iodata_to_binary(output) + assert text =~ "cast #{meta.child_id} (code)" + assert text =~ "child answer" + assert text =~ "1 child turn" + end + + test "real cast_batch tool_result carries child attribution list and renders it" do + parent_llm = + {FakeLLM, + FakeLLM.new([ + %{ + code: """ + children = + for label <- ["a", "b"] do + child_llm = {Cantrip.FakeLLM, Cantrip.FakeLLM.new([%{code: ~s[done.("\#{label}")]}])} + {:ok, child} = Cantrip.new(llm: child_llm, circle: %{type: :code, gates: [:done]}) + %{cantrip: child, intent: label} + end + + {:ok, values, _children, _looms, _meta} = Cantrip.cast_batch(children) + done.(values) + """ + } + ])} + + {:ok, parent} = + Cantrip.new( + llm: parent_llm, + circle: %{ + type: :code, + gates: [:done], + wards: [%{max_turns: 5}, %{max_depth: 1}, %{max_concurrent_children: 2}] + } + ) + + assert {:ok, ["a", "b"], _parent, _loom, _meta} = + Cantrip.cast(parent, "batch", stream_to: self()) + + events = collect_stream_events() + + assert {env, {:tool_result, meta}} = + Enum.find(events, fn + {_env, {:tool_result, %{gate: "cast_batch"}}} -> true + _event -> false + end) + + assert env.depth == 0 + + assert [%{child_id: first_id, circle: :code}, %{child_id: second_id, circle: :code}] = + meta.children + + assert is_binary(first_id) + assert is_binary(second_id) + assert meta.child_turn_count == 2 + + {output, :stderr, _state} = + Renderer.render_event(Renderer.new(), {env, {:tool_result, meta}}) + + text = IO.iodata_to_binary(output) + assert text =~ "cast_batch #{first_id} (code), #{second_id} (code)" + assert text =~ ~s|["a", "b"]| + assert text =~ "2 child turns" + end + end + + defp collect_stream_events(events \\ []) do + receive do + {:cantrip_event, event} -> collect_stream_events([event | events]) + after + 50 -> Enum.reverse(events) + end end describe "depth indentation from envelope" do From 75ab6d9ff3e9c7b72605c50847dbab3b76f5a2d1 Mon Sep 17 00:00:00 2001 From: deepfates Date: Tue, 7 Jul 2026 01:22:24 -0700 Subject: [PATCH 07/10] chore: remove internal process files from the repo --- REPORT.md | 41 ----------------------------------------- 1 file changed, 41 deletions(-) delete mode 100644 REPORT.md diff --git a/REPORT.md b/REPORT.md deleted file mode 100644 index 0dda3df..0000000 --- a/REPORT.md +++ /dev/null @@ -1,41 +0,0 @@ -# dee-kids Bounce Report - -Branch: `child-cast-legibility` - -## Reproduction Commands - -From the repository root: - -```sh -mix test test/cli/renderer_test.exs -``` - -Result: 22 tests, 0 failures. - -```sh -mix test test/familiar_test.exs -``` - -Result: 33 tests, 0 failures. - -```sh -mix test test/composition_test.exs -``` - -Result: 13 tests, 0 failures. - -## Evidence - -- Real `cast` parent observations now include `child_id`, `circle`, `node`, and `batch_index` when the parent knows them, and `Cantrip.Event.tool_events/1` propagates those fields into real `:tool_call` and `:tool_result` events: `lib/cantrip.ex`, `lib/cantrip/event.ex`. -- Real `cast_batch` parent observations now include a `children` attribution list. Single-child batches also expose top-level `child_id`/`circle`; multi-child batches render the honest aggregate label from the child list: `lib/cantrip.ex`, `lib/cantrip/cli/renderer.ex`. -- The code-medium port proxy was also patched because it synthesizes parent observations for proxied `Cantrip.cast/2` and `Cantrip.cast_batch/1` calls: `lib/cantrip/medium/code/port.ex`. -- Renderer regressions now include live runtime probes that call `Cantrip.cast(..., stream_to: self())`, assert the emitted real `:tool_result` metadata, then render that exact event: `test/cli/renderer_test.exs`. -- `can-sgch` was addressed with a note and remains open for coordinator handling. - -## Discovered Tickets - -None. - -## Uncertainty - -I did not run a provider-backed live LLM. The new probes exercise the real Cantrip runtime and streaming event path with deterministic `Cantrip.FakeLLM`. From d5c9398bed1ccbaac7314b8fb49f721f4ed69538 Mon Sep 17 00:00:00 2001 From: deepfates Date: Tue, 7 Jul 2026 01:28:06 -0700 Subject: [PATCH 08/10] chore: ignore fleet process artifacts --- .gitignore | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/.gitignore b/.gitignore index 4f5da7e..f1d5ca3 100644 --- a/.gitignore +++ b/.gitignore @@ -25,3 +25,11 @@ Mnesia.*/ *~ erl_crash.dump scratch/ + +# fleet process artifacts, never committed (tracked in the fleet ledger instead) +/REPORT.md +/REPORT-*.md +/VERDICT.md +/VERDICT-*.md +/.tickets/ +/FRESHREAD*.md From 9144605cb2d0c308558aae2023f7bb9b56e8c0f6 Mon Sep 17 00:00:00 2001 From: deepfates Date: Tue, 7 Jul 2026 01:41:33 -0700 Subject: [PATCH 09/10] feat: add familiar interrupt steering controls --- REPORT.md | 44 +++++ lib/cantrip.ex | 42 +++++ lib/cantrip/acp/agent_handler.ex | 57 +++++- lib/cantrip/acp/runtime.ex | 8 +- lib/cantrip/acp/runtime/familiar.ex | 33 +++- lib/cantrip/entity_server.ex | 278 +++++++++++++++++++++++++--- test/interrupt_steer_test.exs | 268 +++++++++++++++++++++++++++ test/summon_test.exs | 14 +- 8 files changed, 707 insertions(+), 37 deletions(-) create mode 100644 REPORT.md create mode 100644 test/interrupt_steer_test.exs diff --git a/REPORT.md b/REPORT.md new file mode 100644 index 0000000..a513a87 --- /dev/null +++ b/REPORT.md @@ -0,0 +1,44 @@ +# dee-intr report + +## Summary + +Implemented interrupt, steer, queued sends, hard stop, and ACP cancel wiring for persistent Familiar entities. + +- `Cantrip.interrupt/2` requests cancellation of the active episode at the next turn boundary and records loom events. +- `Cantrip.steer/2` injects a user message at the next boundary without ending the episode; it records control events and an intent. +- `Cantrip.send/3` now queues mid-turn sends and replies to each caller when its queued episode completes. +- `Cantrip.hard_stop/2` replies to active/queued callers, records `:hard_stopped`, kills the runner, and flushes/closes the loom. +- ACP `session/cancel` now calls runtime cancellation; Familiar maps interrupted prompts to `PromptResponse.stop_reason == :cancelled`. +- ACP prompt preparation stores a first-prompt entity pid before the blocking prompt call so first-turn cancel can reach the entity while preserving prompt trace ids. + +## Reproduce commands + +```bash +git checkout feat/interrupt-steer +mix test test/interrupt_steer_test.exs test/summon_test.exs test/acp_agent_test.exs +mix test test/familiar_behavior_test.exs +mix test +``` + +Evidence from this workspace: + +```text +mix test test/interrupt_steer_test.exs test/summon_test.exs test/acp_agent_test.exs +28 tests, 0 failures + +mix test test/familiar_behavior_test.exs +21 tests, 0 failures + +mix test +3 properties, 664 tests, 0 failures +``` + +Note: in this managed sandbox, `mix test` needed escalation because Mix PubSub attempted a local TCP socket and the sandbox returned `:eperm`. + +## Uncertainty + +Interrupt is a safe-boundary cancel. It does not abort an in-flight provider call mid-request; the request is observed at the next runtime boundary, then the loom records the interruption and the active episode returns cancelled/interrupted. + +## Filed tickets + +None. diff --git a/lib/cantrip.ex b/lib/cantrip.ex index d8c1894..90aff5b 100644 --- a/lib/cantrip.ex +++ b/lib/cantrip.ex @@ -490,6 +490,18 @@ defmodule Cantrip do DynamicSupervisor.start_child(Cantrip.EntitySupervisor, spec) end + @doc """ + Creates a persistent entity without running an intent, with entity options. + + This is primarily useful for hosts that need to attach runtime metadata such + as `:trace_id` before the first intent is sent. + """ + @spec summon(t(), keyword()) :: {:ok, pid()} | {:error, term()} + def summon(%__MODULE__{} = cantrip, opts) when is_list(opts) do + spec = {EntityServer, [cantrip: cantrip, lazy: true] ++ opts} + DynamicSupervisor.start_child(Cantrip.EntitySupervisor, spec) + end + @doc """ Creates a persistent entity and immediately runs the first intent. @@ -532,6 +544,36 @@ defmodule Cantrip do EntityServer.send_intent(pid, intent, opts) end + @doc """ + Interrupts a persistent entity's active episode at the next safe turn boundary. + + The entity process stays alive, its loom records the interruption, and queued + sends continue after the interrupted episode returns. + """ + @spec interrupt(pid(), term()) :: :ok + def interrupt(pid, reason \\ :interrupt) when is_pid(pid) do + EntityServer.interrupt(pid, reason) + end + + @doc """ + Queues a steering message for delivery at the next turn boundary. + + Steering does not end the active episode. The message is appended to the + entity's message history and loom as an intent before the next provider turn. + """ + @spec steer(pid(), String.t()) :: :ok + def steer(pid, message) when is_pid(pid) and is_binary(message) do + EntityServer.steer(pid, message) + end + + @doc """ + Immediately terminates a persistent entity after recording and closing its loom. + """ + @spec hard_stop(pid(), term()) :: :ok + def hard_stop(pid, reason \\ :hard_stop) when is_pid(pid) do + EntityServer.hard_stop(pid, reason) + end + @doc """ Runs one entity episode for `intent`. diff --git a/lib/cantrip/acp/agent_handler.ex b/lib/cantrip/acp/agent_handler.ex index d58e007..66977fc 100644 --- a/lib/cantrip/acp/agent_handler.ex +++ b/lib/cantrip/acp/agent_handler.ex @@ -121,8 +121,31 @@ defmodule Cantrip.ACP.AgentHandler do end end - defp dispatch({:cancel, _notif}, _table) do - :ok + defp dispatch({:cancel, %ACP.CancelNotification{} = notif}, table) do + runtime = :ets.lookup_element(table, :runtime, 2) + + case :ets.lookup(table, {:session, notif.session_id}) do + [{{:session, session_id}, session}] -> + if function_exported?(runtime, :cancel, 1) do + case runtime.cancel(session) do + {:ok, next_session} -> + :ets.insert(table, {{:session, session_id}, next_session}) + :ok + + {:error, reason, next_session} -> + :ets.insert(table, {{:session, session_id}, next_session}) + {:error, %ACP.Error{code: -32_002, message: Cantrip.SafeFormat.inspect(reason)}} + + {:error, reason} -> + {:error, %ACP.Error{code: -32_002, message: Cantrip.SafeFormat.inspect(reason)}} + end + else + :ok + end + + [] -> + :ok + end end defp dispatch(_request, _table) do @@ -143,10 +166,40 @@ defmodule Cantrip.ACP.AgentHandler do runtime = :ets.lookup_element(table, :runtime, 2) bridge = lookup_bridge(table, session_id) + with {:ok, session} <- prepare_prompt(runtime, table, session_id, session) do + prompt_prepared_runtime(runtime, table, session_id, bridge, session, text) + else + {:error, reason, next_session} -> + :ets.insert(table, {{:session, session_id}, next_session}) + {:error, %ACP.Error{code: -32_002, message: Cantrip.SafeFormat.inspect(reason)}} + end + end + + defp prepare_prompt(runtime, table, session_id, session) do + if function_exported?(runtime, :prepare_prompt, 1) do + case runtime.prepare_prompt(session) do + {:ok, next_session} -> + :ets.insert(table, {{:session, session_id}, next_session}) + {:ok, next_session} + + {:error, _reason, _next_session} = error -> + error + end + else + {:ok, session} + end + end + + defp prompt_prepared_runtime(runtime, table, session_id, bridge, session, text) do case runtime.prompt(session, text) do {:ok, answer, next_session} -> handle_prompt_answer(table, session_id, bridge, answer, next_session) + {:cancelled, next_session} -> + if bridge, do: Cantrip.ACP.EventBridge.flush(bridge) + :ets.insert(table, {{:session, session_id}, next_session}) + {:ok, %ACP.PromptResponse{stop_reason: :cancelled}} + {:error, reason, next_session} -> if bridge, do: Cantrip.ACP.EventBridge.flush(bridge) :ets.insert(table, {{:session, session_id}, next_session}) diff --git a/lib/cantrip/acp/runtime.ex b/lib/cantrip/acp/runtime.ex index 4ea4fe6..a2223fd 100644 --- a/lib/cantrip/acp/runtime.ex +++ b/lib/cantrip/acp/runtime.ex @@ -2,5 +2,11 @@ defmodule Cantrip.ACP.Runtime do @moduledoc false @callback new_session(map()) :: {:ok, term()} | {:error, String.t()} - @callback prompt(term(), String.t()) :: {:ok, String.t(), term()} | {:error, String.t(), term()} + @callback prepare_prompt(term()) :: {:ok, term()} | {:error, String.t(), term()} + + @callback prompt(term(), String.t()) :: + {:ok, String.t(), term()} | {:cancelled, term()} | {:error, String.t(), term()} + @callback cancel(term()) :: {:ok, term()} | {:error, String.t(), term()} + + @optional_callbacks prepare_prompt: 1, cancel: 1 end diff --git a/lib/cantrip/acp/runtime/familiar.ex b/lib/cantrip/acp/runtime/familiar.ex index c306045..8926b9a 100644 --- a/lib/cantrip/acp/runtime/familiar.ex +++ b/lib/cantrip/acp/runtime/familiar.ex @@ -53,6 +53,22 @@ defmodule Cantrip.ACP.Runtime.Familiar do end end + @impl true + def prepare_prompt(%{cantrip: cantrip, entity_pid: nil} = session) do + opts = + case Map.get(session, :trace_id) do + trace_id when is_binary(trace_id) and trace_id != "" -> [trace_id: trace_id] + _ -> [] + end + + case Cantrip.summon(cantrip, opts) do + {:ok, pid} -> {:ok, %{session | entity_pid: pid}} + {:error, reason} -> {:error, Cantrip.SafeFormat.inspect(reason), session} + end + end + + def prepare_prompt(session), do: {:ok, session} + @impl true def prompt(%{cantrip: cantrip, entity_pid: nil} = session, text) when is_binary(text) do opts = stream_opts(session) @@ -69,7 +85,11 @@ defmodule Cantrip.ACP.Runtime.Familiar do end {:error, reason, next_cantrip} -> - {:error, Cantrip.SafeFormat.inspect(reason), %{session | cantrip: next_cantrip}} + if reason == :interrupted do + {:cancelled, %{session | cantrip: next_cantrip}} + else + {:error, Cantrip.SafeFormat.inspect(reason), %{session | cantrip: next_cantrip}} + end end end @@ -87,9 +107,20 @@ defmodule Cantrip.ACP.Runtime.Familiar do {:error, reason} -> {:error, Cantrip.SafeFormat.inspect(reason), session} + + {:error, reason, next_cantrip} when reason == :interrupted -> + {:cancelled, %{session | cantrip: next_cantrip}} end end + @impl true + def cancel(%{entity_pid: pid} = session) when is_pid(pid) do + :ok = Cantrip.interrupt(pid, :acp_cancel) + {:ok, session} + end + + def cancel(session), do: {:ok, session} + defp normalize_answer(nil), do: "" defp normalize_answer(answer) when is_binary(answer), diff --git a/lib/cantrip/entity_server.ex b/lib/cantrip/entity_server.ex index bbbc6fe..8907758 100644 --- a/lib/cantrip/entity_server.ex +++ b/lib/cantrip/entity_server.ex @@ -23,6 +23,9 @@ defmodule Cantrip.EntityServer do stream_barrier?: false, runner: nil, running: nil, + pending: :queue.new(), + controls: [], + stop_requested?: false, entity_started_at: nil, # The summary text from this turn's fold (if folding fired # in `prepare_request`). Threaded into the medium's runtime @@ -49,6 +52,21 @@ defmodule Cantrip.EntityServer do GenServer.call(pid, {:send_intent, intent, opts}, :infinity) end + @doc "Interrupt the active episode at the next safe turn boundary." + def interrupt(pid, reason \\ :interrupt) when is_pid(pid) do + GenServer.call(pid, {:interrupt, reason}, :infinity) + end + + @doc "Steer the entity with a message delivered at the next turn boundary." + def steer(pid, message) when is_pid(pid) and is_binary(message) do + GenServer.call(pid, {:steer, message}, :infinity) + end + + @doc "Terminate the entity immediately after appending a stop event and closing its loom." + def hard_stop(pid, reason \\ :hard_stop) when is_pid(pid) do + GenServer.call(pid, {:hard_stop, reason}, :infinity) + end + @impl true def init(opts) do cantrip = Keyword.fetch!(opts, :cantrip) @@ -119,25 +137,75 @@ defmodule Cantrip.EntityServer do @impl true def handle_call({:send_intent, intent, opts}, from, state) do if state.running do - {:reply, busy_reply(state), state} + {:noreply, enqueue_pending(state, from, intent, opts)} else start_send_intent_episode(state, from, intent, opts) end end + @impl true + def handle_call({:interrupt, reason}, _from, %{running: nil} = state) do + event = control_event(:interrupted, reason, :idle) + {:reply, :ok, %{state | loom: Loom.append_event(state.loom, event)}} + end + + def handle_call({:interrupt, reason}, _from, state) do + send_runner_control(state.runner, {:interrupt, reason}) + {:reply, :ok, state} + end + + @impl true + def handle_call({:steer, message}, _from, %{running: nil} = state) do + {:reply, :ok, append_boundary_steer(state, message)} + end + + def handle_call({:steer, message}, _from, state) do + send_runner_control(state.runner, {:steer, message}) + {:reply, :ok, state} + end + + @impl true + def handle_call({:hard_stop, reason}, _from, state) do + state = + state + |> append_control_event(:hard_stopped, reason, running_kind(state)) + |> close_loom() + + stop_runner(state.runner, :kill) + reply_running(state, {:error, :hard_stopped, state.cantrip}) + reply_pending(state.pending, {:error, :hard_stopped, state.cantrip}) + + {:stop, {:shutdown, reason}, :ok, %{state | running: nil, pending: :queue.new(), runner: nil}} + end + @impl true def handle_info( {:entity_episode_result, ref, {reply, final_state, stop?}}, - %{running: %{ref: ref, from: from}, runner: runner} + %{running: %{ref: ref, from: from}, runner: runner} = state ) do GenServer.reply(from, reply) - final_state = %{final_state | running: nil, runner: runner} + pending = state.pending - if stop? do - {:stop, :normal, final_state} - else - {:noreply, final_state} + final_state = %{ + final_state + | running: nil, + runner: runner, + pending: pending, + controls: [], + stop_requested?: false + } + + cond do + stop? -> + final_state = close_loom(final_state) + {:stop, :normal, final_state} + + pending_empty?(pending) -> + {:noreply, final_state} + + true -> + start_next_pending(final_state) end end @@ -148,6 +216,7 @@ defmodule Cantrip.EntityServer do state = state |> maybe_reply_runner_down(reason) + |> reply_all_pending_runner_down(reason) |> snapshot_runner_owned_state() case start_runner() do @@ -214,6 +283,45 @@ defmodule Cantrip.EntityServer do ) end + defp enqueue_pending(state, from, intent, opts) do + pending = :queue.in(%{from: from, intent: intent, opts: opts, queued?: true}, state.pending) + %{state | pending: pending} + end + + defp start_next_pending(state) do + case :queue.out(state.pending) do + {{:value, %{from: from, intent: intent, opts: opts} = item}, pending} -> + state = %{state | pending: pending} + + state = + if Map.get(item, :queued?), + do: append_control_event(state, :intent_queued, intent, :send_intent), + else: state + + start_send_intent_episode(state, from, intent, opts) + + {:empty, _pending} -> + {:noreply, state} + end + end + + defp pending_empty?(pending), do: :queue.is_empty(pending) + + defp reply_running(%{running: %{from: from}}, reply), do: GenServer.reply(from, reply) + defp reply_running(_state, _reply), do: :ok + + defp reply_pending(pending, reply) do + pending + |> :queue.to_list() + |> Enum.each(fn %{from: from} -> GenServer.reply(from, reply) end) + end + + defp reply_all_pending_runner_down(state, reason) do + reply = {:error, "entity run failed: #{Cantrip.SafeFormat.inspect(reason)}", state.cantrip} + reply_pending(state.pending, reply) + %{state | pending: :queue.new()} + end + defp start_episode(%{running: nil, runner: %{pid: pid}} = state, from, kind, opts) do ref = make_ref() @@ -259,7 +367,7 @@ defmodule Cantrip.EntityServer do case Cantrip.Telemetry.with_context(state.entity_id, state.trace_id, fn -> run_loop(state) end) do {:error, reason, final_state} -> - emit_entity_stop(final_state, :error) + emit_entity_stop(final_state, entity_error_stop_reason(reason)) await_stream_barrier(final_state) final_state = @@ -346,6 +454,16 @@ defmodule Cantrip.EntityServer do defp stop_runner(_runner), do: :ok + defp stop_runner(%{pid: pid, monitor_ref: monitor_ref}, reason) when is_pid(pid) do + Process.demonitor(monitor_ref, [:flush]) + + if Process.alive?(pid) do + Process.exit(pid, reason) + end + end + + defp stop_runner(_runner, _reason), do: :ok + defp build_initial_messages(cantrip, intent, lazy) do cond do is_binary(intent) -> @@ -535,27 +653,101 @@ defmodule Cantrip.EntityServer do next_messages = Cantrip.Turn.next_messages(state.messages, state.cantrip.circle.type, executed) - next_state = %{next_state | messages: next_messages} - - if terminated do - case Cantrip.Turn.final_response( - classified, - executed, - %{entity_id: state.entity_id, turns: next_state.turns}, - usage - ) do - {:error, msg} -> - {:error, msg, next_state} - - {:ok, value, meta} -> - emit_event(state, {:final_response, %{result: value}}) - {value, next_state, meta} - end - else - run_loop(next_state) + next_state = + next_state + |> Map.put(:messages, next_messages) + |> apply_boundary_controls() + |> emit_boundary_heartbeat() + + cond do + next_state.stop_requested? -> + interrupted(next_state) + + terminated -> + case Cantrip.Turn.final_response( + classified, + executed, + %{entity_id: state.entity_id, turns: next_state.turns}, + usage + ) do + {:error, msg} -> + {:error, msg, next_state} + + {:ok, value, meta} -> + emit_event(state, {:final_response, %{result: value}}) + {value, next_state, meta} + end + + true -> + run_loop(next_state) end end + defp interrupted(state) do + state = append_control_event(state, :interrupted, :interrupt, :turn_boundary) + emit_event(state, {:final_response, %{result: nil, stop_reason: :cancelled}}) + + {:error, :interrupted, state} + end + + defp apply_boundary_controls(state) do + state + |> collect_runner_controls() + |> deliver_steers() + end + + defp collect_runner_controls(state) do + receive do + {:control, {:interrupt, reason}} -> + %{state | stop_requested?: true, controls: state.controls ++ [{:interrupt, reason}]} + |> collect_runner_controls() + + {:control, {:steer, message}} -> + %{state | controls: state.controls ++ [{:steer, message}]} + |> collect_runner_controls() + after + 0 -> state + end + end + + defp deliver_steers(%{controls: []} = state), do: state + + defp deliver_steers(state) do + Enum.reduce(state.controls, %{state | controls: []}, fn + {:interrupt, reason}, acc -> + append_control_event(acc, :interrupt_requested, reason, :turn_boundary) + + {:steer, message}, acc -> + append_boundary_steer(acc, message) + end) + end + + defp append_boundary_steer(state, message) do + messages = + if state.lazy do + initial_messages(state.cantrip.identity, state.cantrip.circle, message) + else + state.messages ++ [%{role: :user, content: message}] + end + + loom = + state.loom + |> Loom.append_event(control_event(:steer_queued, message, :turn_boundary)) + |> Loom.append_event(control_event(:steer_delivered, message, :turn_boundary)) + |> Loom.append_intent(message, cantrip_id: state.cantrip.id, entity_id: state.entity_id) + + %{state | messages: messages, loom: loom, lazy: false} + end + + defp emit_boundary_heartbeat(state) do + emit_event( + state, + {:turn_boundary, %{turns: state.turns, queued_control_count: length(state.controls)}} + ) + + state + end + defp initial_messages(identity, circle, intent) do capability_text = MediumRegistry.present(circle).capability_text @@ -745,6 +937,40 @@ defmodule Cantrip.EntityServer do %{state | stream_to: stream_to, stream_barrier?: stream_barrier?} end + defp send_runner_control(%{pid: pid}, control) when is_pid(pid) do + send(pid, {:control, control}) + :ok + end + + defp send_runner_control(_runner, _control), do: :ok + + defp append_control_event(state, type, reason, kind) do + %{state | loom: Loom.append_event(state.loom, control_event(type, reason, kind))} + end + + defp control_event(type, reason, kind) do + %{ + type: type, + reason: control_reason(reason), + episode_kind: kind + } + end + + defp control_reason(reason) when is_binary(reason), do: reason + defp control_reason(reason) when is_atom(reason), do: Atom.to_string(reason) + defp control_reason(reason), do: Cantrip.SafeFormat.inspect(reason) + + defp running_kind(%{running: %{kind: kind}}), do: kind + defp running_kind(_state), do: :idle + + defp entity_error_stop_reason(:interrupted), do: :cancelled + defp entity_error_stop_reason(_reason), do: :error + + defp close_loom(state) do + _ = Loom.close(state.loom) + state + end + defp emit_entity_stop(state, reason) do Cantrip.Telemetry.execute( [:cantrip, :entity, :stop], diff --git a/test/interrupt_steer_test.exs b/test/interrupt_steer_test.exs new file mode 100644 index 0000000..2ba8d6e --- /dev/null +++ b/test/interrupt_steer_test.exs @@ -0,0 +1,268 @@ +defmodule Cantrip.InterruptSteerTest do + use ExUnit.Case, async: true + + alias Cantrip.ACP.AgentHandler + + defmodule BlockingScriptLLM do + @behaviour Cantrip.LLM + + @impl true + def query(%{test_pid: test_pid, responses: [response | rest]} = state, request) do + content = request.messages |> List.last() |> Map.fetch!(:content) + send(test_pid, {:blocking_script_llm_started, self(), content, request.messages}) + + receive do + {:release_blocking_script_llm, ^content} -> + {:ok, response(response), %{state | responses: rest}} + after + 1_000 -> + {:error, %{message: "blocking script llm was not released"}, state} + end + end + + defp response(%Cantrip.LLM.Response{} = response), do: response + + defp response(%{code: code} = attrs) when is_binary(code) do + attrs + |> Map.delete(:code) + |> Map.put(:tool_calls, [%{id: "tc_script", gate: "elixir", args: %{"code" => code}}]) + |> response() + end + + defp response(attrs) when is_map(attrs) do + %Cantrip.LLM.Response{ + content: Map.get(attrs, :content), + tool_calls: Map.get(attrs, :tool_calls, []), + usage: %{} + } + end + end + + defmodule FamiliarRuntimeFromProcess do + @behaviour Cantrip.ACP.Runtime + + @impl true + def new_session(params) do + params = + case Process.get(:interrupt_steer_test_llm) do + nil -> params + llm -> Map.put(params, "llm", llm) + end + + Cantrip.ACP.Runtime.Familiar.new_session(params) + end + + @impl true + def prepare_prompt(session), do: Cantrip.ACP.Runtime.Familiar.prepare_prompt(session) + + @impl true + def prompt(session, text), do: Cantrip.ACP.Runtime.Familiar.prompt(session, text) + + @impl true + def cancel(session), do: Cantrip.ACP.Runtime.Familiar.cancel(session) + end + + defmodule LifecycleStorage do + @behaviour Cantrip.Loom.Storage + + @impl true + def init(opts) do + opts = Map.new(opts) + path = Map.fetch!(opts, :path) + lifecycle_path = Map.fetch!(opts, :lifecycle_path) + File.mkdir_p!(Path.dirname(path)) + File.write!(path, "", [:append]) + File.write!(lifecycle_path, "", [:append]) + {:ok, %{path: path, lifecycle_path: lifecycle_path}} + end + + @impl true + def append_event(%{path: path} = state, event) do + encoded = event |> :erlang.term_to_binary() |> Base.encode64() + File.write!(path, encoded <> "\n", [:append]) + {:ok, state} + end + + @impl true + def append_turn(state, turn), do: append_event(state, %{type: :turn, turn: turn}) + + @impl true + def annotate_reward(state, index, reward) do + append_event(state, %{type: :reward, index: index, reward: reward}) + end + + @impl true + def load(_state), do: {:ok, %{events: [], turns: []}} + + @impl true + def flush(%{lifecycle_path: lifecycle_path} = state) do + File.write!(lifecycle_path, "flush\n", [:append]) + {:ok, state} + end + + @impl true + def close(%{lifecycle_path: lifecycle_path}) do + File.write!(lifecycle_path, "close\n", [:append]) + :ok + end + end + + test "interrupt stops the active episode at the next turn boundary and records loom events" do + {:ok, cantrip} = code_cantrip([%{code: "x = 1"}]) + {:ok, pid} = Cantrip.summon(cantrip) + + task = Task.async(fn -> Cantrip.send(pid, "slow") end) + assert_receive {:blocking_script_llm_started, llm_pid, "slow", _messages}, 1_000 + + assert :ok = Cantrip.interrupt(pid) + send(llm_pid, {:release_blocking_script_llm, "slow"}) + + assert {:error, :interrupted, next_cantrip} = Task.await(task, 5_000) + state = :sys.get_state(pid) + + assert next_cantrip.id == state.cantrip.id + assert Enum.any?(state.loom.events, &match?(%{type: :interrupt_requested}, &1)) + assert Enum.any?(state.loom.events, &match?(%{type: :interrupted}, &1)) + end + + test "steer injects a message at the next turn boundary without killing the run" do + {:ok, cantrip} = + code_cantrip([ + %{code: "x = 1"}, + %{code: ~s|done.("steered")|} + ]) + + {:ok, pid} = Cantrip.summon(cantrip) + task = Task.async(fn -> Cantrip.send(pid, "first") end) + + assert_receive {:blocking_script_llm_started, llm_pid, "first", _messages}, 1_000 + assert :ok = Cantrip.steer(pid, "change course") + send(llm_pid, {:release_blocking_script_llm, "first"}) + + assert_receive {:blocking_script_llm_started, llm_pid2, "change course", messages}, 5_000 + assert List.last(messages) == %{role: :user, content: "change course"} + send(llm_pid2, {:release_blocking_script_llm, "change course"}) + + assert {:ok, "steered", _next, loom, _meta} = Task.await(task, 5_000) + assert Enum.any?(loom.events, &match?(%{type: :steer_queued}, &1)) + assert Enum.any?(loom.events, &match?(%{type: :steer_delivered}, &1)) + assert Enum.any?(loom.intents, &(&1.utterance.content == "change course")) + end + + test "mid-turn sends queue and run in order at episode boundaries" do + {:ok, cantrip} = + conversation_cantrip([ + %{tool_calls: [%{gate: "done", args: %{answer: "one"}}]}, + %{tool_calls: [%{gate: "done", args: %{answer: "two"}}]} + ]) + + {:ok, pid} = Cantrip.summon(cantrip) + first = Task.async(fn -> Cantrip.send(pid, "first") end) + + assert_receive {:blocking_script_llm_started, llm_pid, "first", _messages}, 1_000 + + second = Task.async(fn -> Cantrip.send(pid, "second") end) + refute Task.yield(second, 50) + + send(llm_pid, {:release_blocking_script_llm, "first"}) + assert {:ok, "one", _next, _loom, _meta} = Task.await(first, 1_000) + + assert_receive {:blocking_script_llm_started, llm_pid2, "second", _messages}, 1_000 + send(llm_pid2, {:release_blocking_script_llm, "second"}) + assert {:ok, "two", _next, loom, _meta} = Task.await(second, 1_000) + + assert Enum.any?(loom.events, &match?(%{type: :intent_queued}, &1)) + assert Enum.map(loom.intents, & &1.utterance.content) == ["first", "second"] + end + + test "hard stop terminates immediately and closes the loom" do + tmp = Path.join(System.tmp_dir!(), "cantrip-hard-stop-#{System.unique_integer([:positive])}") + File.mkdir_p!(tmp) + loom_path = Path.join(tmp, "loom.terms") + lifecycle_path = Path.join(tmp, "lifecycle.log") + + storage = {LifecycleStorage, [path: loom_path, lifecycle_path: lifecycle_path]} + + {:ok, cantrip} = + conversation_cantrip([%{tool_calls: [%{gate: "done", args: %{answer: "late"}}]}], storage) + + {:ok, pid} = Cantrip.summon(cantrip) + + task = Task.async(fn -> Cantrip.send(pid, "slow") end) + assert_receive {:blocking_script_llm_started, _llm_pid, "slow", _messages}, 1_000 + + assert :ok = Cantrip.hard_stop(pid) + assert {:error, :hard_stopped, _next} = Task.await(task, 1_000) + refute Process.alive?(pid) + + assert File.read!(lifecycle_path) == "flush\nclose\n" + assert Enum.any?(read_terms(loom_path), &match?(%{type: :hard_stopped}, &1)) + end + + test "ACP session/cancel interrupts a Familiar prompt and returns cancelled stop reason" do + llm = {BlockingScriptLLM, %{test_pid: self(), responses: [%{code: "x = 1"}]}} + Process.put(:interrupt_steer_test_llm, llm) + on_exit(fn -> Process.delete(:interrupt_steer_test_llm) end) + + table = AgentHandler.new(runtime: FamiliarRuntimeFromProcess) + AgentHandler.handle_request(init_request(), table) + + {:ok, %ACP.NewSessionResponse{session_id: session_id}} = + AgentHandler.handle_request({:new_session, %ACP.NewSessionRequest{cwd: "/tmp"}}, table) + + prompt = + Task.async(fn -> + AgentHandler.handle_request( + {:prompt, + %ACP.PromptRequest{ + session_id: session_id, + prompt: [{:text, %ACP.TextContent{text: "slow"}}] + }}, + table + ) + end) + + assert_receive {:blocking_script_llm_started, llm_pid, "slow", _messages}, 500 + + assert :ok = + AgentHandler.handle_request( + {:cancel, ACP.CancelNotification.new(session_id)}, + table + ) + + send(llm_pid, {:release_blocking_script_llm, "slow"}) + assert {:ok, %ACP.PromptResponse{stop_reason: :cancelled}} = Task.await(prompt, 500) + end + + defp code_cantrip(responses, storage \\ :memory) do + Cantrip.new( + llm: {BlockingScriptLLM, %{test_pid: self(), responses: responses}}, + loom_storage: storage, + circle: %{type: :code, gates: [:done], wards: [%{max_turns: 5}, %{require_done_tool: true}]} + ) + end + + defp conversation_cantrip(responses, storage \\ :memory) do + Cantrip.new( + llm: {BlockingScriptLLM, %{test_pid: self(), responses: responses}}, + loom_storage: storage, + circle: %{type: :conversation, gates: [:done], wards: [%{max_turns: 5}]} + ) + end + + defp init_request do + {:initialize, + %ACP.InitializeRequest{ + protocol_version: 1, + client_capabilities: %ACP.ClientCapabilities{}, + client_info: %{"name" => "test"} + }} + end + + defp read_terms(path) do + path + |> File.read!() + |> String.split("\n", trim: true) + |> Enum.map(fn line -> line |> Base.decode64!() |> :erlang.binary_to_term() end) + end +end diff --git a/test/summon_test.exs b/test/summon_test.exs index ddb8ce2..e4bb074 100644 --- a/test/summon_test.exs +++ b/test/summon_test.exs @@ -140,7 +140,7 @@ defmodule Cantrip.SummonTest do assert state_after_second.code_state.port_session.port == first_port end - test "persistent entity mailbox stays responsive while an episode is running" do + test "persistent entity queues sends while an episode is running" do llm = {BlockingLLM, %{test_pid: self()}} {:ok, cantrip} = @@ -155,15 +155,15 @@ defmodule Cantrip.SummonTest do assert_receive {:blocking_llm_started, query_pid, "slow"}, 200 - started_at = System.monotonic_time(:millisecond) - assert {:error, "entity is already running", _cantrip} = Cantrip.send(pid, "second") - elapsed = System.monotonic_time(:millisecond) - started_at - - assert elapsed < 200, - "second send should be rejected by the EntityServer mailbox, not wait for provider work" + queued = Task.async(fn -> Cantrip.send(pid, "second") end) + refute Task.yield(queued, 50) send(query_pid, {:release_blocking_llm, "slow"}) assert {:ok, "released:slow", _cantrip, _loom, _meta} = Task.await(running, 500) + + assert_receive {:blocking_llm_started, query_pid2, "second"}, 500 + send(query_pid2, {:release_blocking_llm, "second"}) + assert {:ok, "released:second", _cantrip, _loom, _meta} = Task.await(queued, 500) end test "send preserves the terminating turn's assistant message in state.messages" do From 78d25a5e0455d4f9ace41656ef82b4a7e6f8dfe8 Mon Sep 17 00:00:00 2001 From: deepfates Date: Tue, 7 Jul 2026 01:48:52 -0700 Subject: [PATCH 10/10] fix: force unicode stdio for ACP server --- lib/cantrip/acp/server.ex | 1 + test/acp_agent_stdio_test.exs | 36 ++++++++++++++++------------------- 2 files changed, 17 insertions(+), 20 deletions(-) diff --git a/lib/cantrip/acp/server.ex b/lib/cantrip/acp/server.ex index f9aff68..9c98e9f 100644 --- a/lib/cantrip/acp/server.ex +++ b/lib/cantrip/acp/server.ex @@ -7,6 +7,7 @@ defmodule Cantrip.ACP.Server do """ def run(opts \\ []) do + :ok = :io.setopts(:standard_io, encoding: :unicode) {:ok, _apps} = Application.ensure_all_started(:cantrip) runtime = Keyword.get(opts, :runtime, Cantrip.ACP.Runtime.Familiar) diff --git a/test/acp_agent_stdio_test.exs b/test/acp_agent_stdio_test.exs index dcce393..9bd3c08 100644 --- a/test/acp_agent_stdio_test.exs +++ b/test/acp_agent_stdio_test.exs @@ -38,14 +38,17 @@ defmodule Cantrip.ACP.AgentStdioTest do assert %{"id" => 2, "result" => %{"sessionId" => session_id}} = session_resp assert is_binary(session_id) - # Prompt + # Prompt with non-ASCII text. Erlang Port-spawned BEAMs default stdio to + # latin1, so this exercises both inbound UTF-8 decoding and outbound JSON. + prompt_text = "hello “from Port” 🚀" + send_json(port, %{ "jsonrpc" => "2.0", "id" => 3, "method" => "session/prompt", "params" => %{ "sessionId" => session_id, - "prompt" => [%{"type" => "text", "text" => "hello"}] + "prompt" => [%{"type" => "text", "text" => prompt_text}] } }) @@ -57,7 +60,11 @@ defmodule Cantrip.ACP.AgentStdioTest do "params" => %{ "sessionId" => ^session_id, "update" => %{ - "sessionUpdate" => "agent_message_chunk" + "sessionUpdate" => "agent_message_chunk", + "content" => %{ + "type" => "text", + "text" => "echo:hello “from Port” 🚀 — “ok” ✨" + } } } } = update @@ -80,24 +87,9 @@ defmodule Cantrip.ACP.AgentStdioTest do eval = """ defmodule StubRuntime do def new_session(%{"cwd" => cwd}), do: {:ok, %{cwd: cwd, n: 0}} - def prompt(session, text), do: {:ok, "echo:" <> text, %{session | n: session.n + 1}} + def prompt(session, text), do: {:ok, "echo:" <> text <> " — “ok” ✨", %{session | n: session.n + 1}} end - {:ok, _apps} = Application.ensure_all_started(:cantrip) - - table = Cantrip.ACP.AgentHandler.new(runtime: StubRuntime) - gl = Process.group_leader() - - {:ok, conn} = - ACP.AgentSideConnection.start_link( - handler: Cantrip.ACP.AgentHandler, - handler_state: table, - input: gl, - output: gl - ) - - Cantrip.ACP.AgentHandler.set_connection(table, conn) - # Watchdog: exit when the test parent dies so we never leak this BEAM. # Port.close from the test side does not deliver SIGTERM to the spawned # executable on macOS, so without this watchdog every test run leaves @@ -113,7 +105,7 @@ defmodule Cantrip.ACP.AgentStdioTest do System.halt(0) end) - Process.sleep(:infinity) + Cantrip.ACP.Server.run(runtime: StubRuntime) """ args = @@ -130,9 +122,13 @@ defmodule Cantrip.ACP.AgentStdioTest do defp recv_json(port) do receive do {^port, {:data, {:eol, line}}} -> + assert String.valid?(line) + refute String.contains?(line, "\\x{") Jason.decode!(line) {^port, {:data, {:noeol, line}}} -> + assert String.valid?(line) + refute String.contains?(line, "\\x{") Jason.decode!(line) {^port, {:exit_status, status}} ->