Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 8 additions & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
20 changes: 13 additions & 7 deletions lib/codex_wrapper/session.ex
Original file line number Diff line number Diff line change
Expand Up @@ -37,13 +37,15 @@ 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()
}

defstruct [
:config,
:session_id,
:last_turn,
:usage_total,
history: [],
exec_opts: []
]
Expand Down Expand Up @@ -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:

Expand All @@ -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}
Expand Down Expand Up @@ -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
Expand All @@ -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.
Expand Down
49 changes: 31 additions & 18 deletions lib/codex_wrapper/turn.ex
Original file line number Diff line number Diff line change
@@ -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:
Expand All @@ -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}
Expand Down
75 changes: 63 additions & 12 deletions test/codex_wrapper/session_test.exs
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand All @@ -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

Expand All @@ -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
Expand Down Expand Up @@ -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,
Expand All @@ -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,
Expand All @@ -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
Expand Down
46 changes: 45 additions & 1 deletion test/codex_wrapper/session_turn_test.exs
Original file line number Diff line number Diff line change
Expand Up @@ -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}})
Expand All @@ -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

Expand Down
49 changes: 49 additions & 0 deletions test/codex_wrapper/turn_test.exs
Original file line number Diff line number Diff line change
Expand Up @@ -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"

Expand Down
Loading