diff --git a/README.md b/README.md index 1edff73..1e7349a 100644 --- a/README.md +++ b/README.md @@ -162,7 +162,8 @@ session exposes the typed JSON turn as `session.last_turn` or Usage totals sum reported nonnegative integers from recorded turns. A field is `nil` if it was never reported; the whole total is `nil` if no recorded turn included usage. Nonzero exits leave the session unchanged and do not -contribute to usage totals. +contribute to usage totals. Each recorded send updates a running total; +reading it takes constant time and does not parse the retained stdout again. Session options are the exec options in the table under [Exec options](#exec-options) and apply to every turn. Resumed turns run @@ -200,6 +201,12 @@ event. `CodexWrapper.exec/2` and `CodexWrapper.review/1` still return `{:ok, %Result{success: false}}` for a nonzero exit; runner errors return `{:error, reason}`. +Status, error, and terminal evidence cover the entire event stream. A new +`turn.started` resets only the final message and usage; it does not erase an +earlier reported outcome. For example, a completed turn followed by a new +start still has `status: :completed`, but its message and usage are `nil` +until the new turn supplies them. See `CodexWrapper.Turn` for each field's scope. + Usage has the keys `:input_tokens`, `:cached_input_tokens`, `:cache_write_input_tokens`, `:output_tokens` and `:reasoning_output_tokens`. Only nonnegative integers are retained; missing or invalid counters are diff --git a/lib/codex_wrapper/session.ex b/lib/codex_wrapper/session.ex index d2f9021..8382eb5 100644 --- a/lib/codex_wrapper/session.ex +++ b/lib/codex_wrapper/session.ex @@ -37,6 +37,7 @@ defmodule CodexWrapper.Session do session_id: String.t() | nil, history: [Result.t()], last_turn: Turn.t() | nil, + usage_total: Turn.usage() | nil, exec_opts: keyword() } @@ -44,6 +45,7 @@ defmodule CodexWrapper.Session do :config, :session_id, :last_turn, + :usage_total, history: [], exec_opts: [] ] @@ -101,6 +103,8 @@ defmodule CodexWrapper.Session do and through `last_turn/1`. The returned result and history remain raw `%Result{}` values. Inspect the projection's `status` for the JSON turn outcome, which may be `:failed` even when codex exited 0. + Each recorded turn also updates the running token totals returned by + `total_usage/1`, using the same projection. Returns: @@ -126,7 +130,8 @@ defmodule CodexWrapper.Session do session | session_id: session_id, history: session.history ++ [result], - last_turn: turn + last_turn: turn, + usage_total: sum_usage(turn.usage, session.usage_total) } {:ok, updated, result} @@ -189,7 +194,12 @@ defmodule CodexWrapper.Session do def total_cost(%__MODULE__{}), do: nil @doc """ - Sum the reported token usage across recorded turns. + Get the reported token usage accumulated by `send/3`. + + Reads the running totals in constant time without parsing raw result + history. `send/3` updates the totals when it records a turn; the raw + `%Result{}` history remains unchanged. A new or resumed session has no + totals until a recorded turn reports usage. Returns `nil` when no turn reported a usage map. Otherwise returns the five usage fields from `CodexWrapper.Turn`, summing only reported @@ -201,11 +211,7 @@ defmodule CodexWrapper.Session do and do not contribute to these totals. """ @spec total_usage(t()) :: Turn.usage() | nil - def total_usage(%__MODULE__{history: history}) do - Enum.reduce(history, nil, fn result, total -> - result |> Turn.from_result() |> Map.fetch!(:usage) |> sum_usage(total) - end) - end + def total_usage(%__MODULE__{usage_total: total}), do: total @doc """ Get the typed projection of the last recorded turn, if any. diff --git a/lib/codex_wrapper/turn.ex b/lib/codex_wrapper/turn.ex index 4eed61e..b79ce86 100644 --- a/lib/codex_wrapper/turn.ex +++ b/lib/codex_wrapper/turn.ex @@ -1,6 +1,6 @@ defmodule CodexWrapper.Turn do @moduledoc """ - A pure projection of a command's `--json` output. + A pure projection of a command's `--json` output, which can contain several turns. `from_result/1` starts no process and leaves the raw `CodexWrapper.Result` unchanged. Ordinary text messages do not need to contain JSON: @@ -9,23 +9,36 @@ defmodule CodexWrapper.Turn do turn = CodexWrapper.Turn.from_result(result) turn.final_message - The last `turn.completed` or `turn.failed` determines the native outcome. - A top-level `error` fails the turn until a later terminal event supersedes - it. A nonzero process exit always makes `status` `:failed`, including after - `turn.completed`. Without a terminal event or error, status is `:no_terminal` - when the process exited zero or its exit code is unknown. - - `final_message` is the last completed `agent_message` text after the last - `turn.started`. `thread_id` is the first nonblank string ID announced by - `thread.started`. Usage comes from the last terminal event of the last - turn. Only the five documented token fields are projected, each as a - nonnegative integer or `nil`; missing usage is `nil`, not zero. - - `error` retains native error data, or is `nil` when no native error was - reported. A process failure alone does not fabricate an error object. - `events` keeps all parsed events, including unknown fields and invalid - usage values. `terminal_event` retains the last native terminal event, - independently of process exit and subsequent top-level errors. + ## Field scopes + + `status`, `error`, and `terminal_event` summarize the **entire event stream**. + The last `turn.completed` or `turn.failed` determines the native outcome; + a top-level `error` fails it until a later terminal event supersedes that + error. A new `turn.started` does not clear these three fields. Therefore + `:completed` means the stream's last reported terminal outcome completed, + not that its last started turn necessarily finished. `:no_terminal` means + no terminal event or error was reported anywhere in the stream, provided + the process did not exit nonzero. + + `exit_code` describes the whole process. A nonzero exit always makes + `status` `:failed`, including after `turn.completed`. `error` retains native + error data or `nil`; process failure alone does not fabricate an error + object. `terminal_event` remains the last native terminal event regardless + of process exit or subsequent top-level errors. `thread_id` is the first + nonblank string ID announced by `thread.started` anywhere in the stream. + `events` keeps every parsed event, including unknown fields and invalid + usage values. + + `final_message` and `usage` describe only the **last started turn**. Both + reset to `nil` at each `turn.started`. The message is its last completed + `agent_message` text, and usage comes from its last terminal event. If no + `turn.started` appears, these fields use the whole stream. Only the five + documented token fields are projected, each as a nonnegative integer or + `nil`; missing usage is `nil`, not zero. + + For example, `turn.started`, `turn.completed`, `turn.started` produces + `status: :completed` and retains the earlier `terminal_event`, while + `final_message` and `usage` are `nil`. The new turn has no reported outcome. """ alias CodexWrapper.{JsonLineEvent, Result} diff --git a/test/codex_wrapper/session_test.exs b/test/codex_wrapper/session_test.exs index 39398ee..ed1ab18 100644 --- a/test/codex_wrapper/session_test.exs +++ b/test/codex_wrapper/session_test.exs @@ -3,8 +3,28 @@ defmodule CodexWrapper.SessionTest do alias CodexWrapper.{Config, JsonLineEvent, Result, Session} + # Synthetic NDJSON supplied as the prompt makes each test's turn output + # explicit while exercising the normal Session.send/3 recording path. + defmodule UsageRunner do + @behaviour CodexWrapper.Runner + + @impl true + def run(_binary, _args, opts, _timeout), do: {:ok, {Keyword.fetch!(opts, :stdin), 0}} + end + defp event(type, data), do: %JsonLineEvent{event_type: type, data: data, raw: ""} + defp usage_session do + Session.new(Config.new(binary: "synthetic-codex", runner: UsageRunner)) + end + + defp record_usage(session, usage) do + event = %{"type" => "turn.completed"} + event = if usage == :missing, do: event, else: Map.put(event, "usage", usage) + {:ok, session, _result} = Session.send(session, Jason.encode!(event)) + session + end + describe "new/2" do test "creates session with config" do config = Config.new() @@ -13,6 +33,7 @@ defmodule CodexWrapper.SessionTest do assert session.session_id == nil assert session.history == [] assert session.last_turn == nil + assert session.usage_total == nil assert session.exec_opts == [] end @@ -34,6 +55,7 @@ defmodule CodexWrapper.SessionTest do config = Config.new() session = Session.resume(config, "abc-123") assert Session.session_id(session) == "abc-123" + assert Session.total_usage(session) == nil end test "with exec opts" do @@ -90,26 +112,25 @@ defmodule CodexWrapper.SessionTest do describe "total_usage/1" do test "returns nil when there is no reported usage" do - session = Session.new(Config.new()) + session = usage_session() assert Session.total_usage(session) == nil - result = Result.from_cmd({~s({"type":"turn.completed"}), 0}) - assert Session.total_usage(%{session | history: [result]}) == nil + session = record_usage(session, :missing) + assert Session.turn_count(session) == 1 + assert Session.total_usage(session) == nil end test "sums known counters while preserving unknown values and zero" do usages = [ %{"input_tokens" => 12, "cached_input_tokens" => 0, "output_tokens" => 3}, + :missing, %{"input_tokens" => 5, "cache_write_input_tokens" => 2, "output_tokens" => -1}, - %{"input_tokens" => "9", "cached_input_tokens" => nil, "output_tokens" => 1.5} + %{"input_tokens" => "9", "cached_input_tokens" => nil, "output_tokens" => 1.5}, + %{"reasoning_output_tokens" => false} ] - results = - Enum.map(usages, fn usage -> - Result.from_cmd({Jason.encode!(%{"type" => "turn.completed", "usage" => usage}), 0}) - end) - - session = %{Session.new(Config.new()) | history: results} + session = Enum.reduce(usages, usage_session(), &record_usage(&2, &1)) + assert Session.turn_count(session) == length(usages) assert Session.total_usage(session) == %{ input_tokens: 17, @@ -121,8 +142,7 @@ defmodule CodexWrapper.SessionTest do end test "a reported empty usage map retains unknown counters" do - result = Result.from_cmd({~s({"type":"turn.completed","usage":{}}), 0}) - session = %{Session.new(Config.new()) | history: [result]} + session = record_usage(usage_session(), %{}) assert Session.total_usage(session) == %{ input_tokens: nil, @@ -132,6 +152,37 @@ defmodule CodexWrapper.SessionTest do reasoning_output_tokens: nil } end + + test "repeated reads use recorded totals independently of raw stdout history" do + session = + usage_session() + |> record_usage(%{"input_tokens" => 12, "output_tokens" => 0}) + |> record_usage(%{"input_tokens" => 5}) + + expected = Session.total_usage(session) + assert expected.input_tokens == 17 + assert expected.output_tokens == 0 + + # A consumer may redact retained raw output after recording a turn. + # Replaying these Results would lose the recorded usage entirely. + history = + Enum.map(session.history, fn %Result{} = result -> %{result | stdout: "redacted"} end) + + session = %{session | history: history} + + for _ <- 1..3, do: assert(Session.total_usage(session) == expected) + assert Session.history(session) == history + + session = record_usage(session, %{"input_tokens" => 1, "cache_write_input_tokens" => 0}) + + assert Session.total_usage(session) == %{ + expected + | input_tokens: 18, + cache_write_input_tokens: 0 + } + + assert Enum.take(Session.history(session), 2) == history + end end describe "turns/1" do diff --git a/test/codex_wrapper/session_turn_test.exs b/test/codex_wrapper/session_turn_test.exs index b178b99..7241782 100644 --- a/test/codex_wrapper/session_turn_test.exs +++ b/test/codex_wrapper/session_turn_test.exs @@ -157,7 +157,9 @@ defmodule CodexWrapper.SessionTurnTest do end describe "typed session server access" do - test "reports usage and preserves the previous turn when a later call fails", %{tmp_dir: dir} do + test "preserves usage and the previous turn on exit and thread mismatch failures", %{ + tmp_dir: dir + } do lines = [ started(@uuid), ~s({"type":"turn.completed","usage":{"input_tokens":8,"output_tokens":0}}) @@ -177,6 +179,48 @@ defmodule CodexWrapper.SessionTurnTest do assert SessionServer.total_usage(pid) == turn.usage assert SessionServer.history(pid) == [result] assert SessionServer.turn_count(pid) == 1 + + mismatched_lines = [ + started(@other), + ~s({"type":"turn.completed","usage":{"input_tokens":100,"output_tokens":9}}) + ] + + fake(dir, mismatched_lines, nil, 0) + + assert {:error, {:thread_mismatch, @uuid, @other, %Result{}}} = + SessionServer.send_message(pid, "third") + + assert SessionServer.last_turn(pid) == turn + assert SessionServer.total_usage(pid) == turn.usage + assert SessionServer.history(pid) == [result] + assert SessionServer.turn_count(pid) == 1 + end + + test "repeated usage reads do not replay retained raw output", %{tmp_dir: dir} do + lines = [ + started(@uuid), + ~s({"type":"turn.completed","usage":{"input_tokens":8,"output_tokens":0}}) + ] + + config = fake(dir, lines, nil, 0) + {:ok, pid} = SessionServer.start_link(config: config) + assert {:ok, %Result{}} = SessionServer.send_message(pid, "first") + assert {:ok, %Result{}} = SessionServer.send_message(pid, "second") + expected = SessionServer.total_usage(pid) + assert expected.input_tokens == 16 + assert expected.output_tokens == 0 + + # Preserve the recorded totals while redacting raw payloads. Re-parsing + # these Results instead of reading the aggregate would return nil. + :sys.replace_state(pid, fn session -> + history = + Enum.map(session.history, fn %Result{} = result -> %{result | stdout: "redacted"} end) + + %{session | history: history} + end) + + for _ <- 1..3, do: assert(SessionServer.total_usage(pid) == expected) + assert Enum.all?(SessionServer.history(pid), &(&1.stdout == "redacted")) end end diff --git a/test/codex_wrapper/turn_test.exs b/test/codex_wrapper/turn_test.exs index 0e80cdf..8f28bcd 100644 --- a/test/codex_wrapper/turn_test.exs +++ b/test/codex_wrapper/turn_test.exs @@ -109,6 +109,55 @@ defmodule CodexWrapper.TurnTest do assert Turn.from_result(result(events ++ [event("turn.started")])).final_message == nil end + test "a new turn clears its message and usage but retains the stream's reported outcome" do + usage = %{"input_tokens" => 12, "output_tokens" => 3} + error = %{"message" => "earlier failure"} + + for {outcome, status, native_error} <- [ + {event("turn.completed", %{"usage" => usage}), :completed, nil}, + {event("turn.failed", %{"error" => error, "usage" => usage}), :failed, error} + ] do + events = [ + event("thread.started", %{"thread_id" => "native-thread"}), + event("turn.started"), + message("earlier answer"), + outcome, + event("turn.started") + ] + + turn = Turn.from_result(result(events)) + assert turn.status == status + assert turn.error == native_error + assert turn.terminal_event.data == outcome + assert turn.final_message == nil + assert turn.usage == nil + assert turn.thread_id == "native-thread" + assert turn.events == JsonLineEvent.parse_lines(result(events).stdout) + + failed_process = Turn.from_result(result(events, 7)) + assert failed_process.status == :failed + assert failed_process.exit_code == 7 + assert failed_process.terminal_event == turn.terminal_event + assert failed_process.error == native_error + end + end + + test "a new turn retains a preceding top-level error until a terminal supersedes it" do + error = event("error", %{"message" => "earlier stream failure"}) + events = [event("turn.completed"), error, event("turn.started")] + turn = Turn.from_result(result(events)) + + assert turn.status == :failed + assert turn.error == error + assert turn.terminal_event.event_type == "turn.completed" + assert turn.final_message == nil + assert turn.usage == nil + + recovered = Turn.from_result(result(events ++ [event("turn.completed")])) + assert recovered.status == :completed + assert recovered.error == nil + end + test "missing turn.started is allowed, and malformed final message does not reuse earlier text" do assert Turn.from_result(result([message("hello")])).final_message == "hello"