From 5e158f1d0d7d62a20f601a9b82e7cccaca748a11 Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Thu, 12 Nov 2026 11:55:29 -0500 Subject: [PATCH 1/6] turn.tool carries the conversation's live spend so far Carries #2413 from the closed integration PR #2306 (workstreams/main); it never reached dev. Co-Authored-By: Claude Mythos 5.2 --- cmd/codeaf/app.go | 4 +- cmd/codeaf/appspendlive_test.go | 216 ++++++++++++++++++ cmd/codeaf/apptools.go | 11 +- .../unreleased/2413-turn-tool-live-spend.md | 12 + docs/design/app-v3/PROTOCOL.md | 5 + internal/appwire/turntool_spend_test.go | 64 ++++++ internal/appwire/wire.go | 5 + .../manual/chat/running-from-the-terminal.md | 3 + 8 files changed, 316 insertions(+), 4 deletions(-) create mode 100644 cmd/codeaf/appspendlive_test.go create mode 100644 docs/changes/unreleased/2413-turn-tool-live-spend.md create mode 100644 internal/appwire/turntool_spend_test.go diff --git a/cmd/codeaf/app.go b/cmd/codeaf/app.go index 910808abb0..29a2dcd79d 100644 --- a/cmd/codeaf/app.go +++ b/cmd/codeaf/app.go @@ -2994,7 +2994,7 @@ func (b *appBackend) pump(a *appAgent, id string, live *appLive, turn uint64, n b.stepMoved(a, live, title) row := b.callBegan(a, id, turn, event, title) row.Summary = event.Hint - live.box.Event(appwire.EventTurnTool, row) + live.sayTool(row) case session.EventToolOutput: b.callOutput(a, id, live, event) case session.EventToolFinished: @@ -3013,7 +3013,7 @@ func (b *appBackend) pump(a *appAgent, id string, live *appLive, turn uint64, n if title == "" { title = redact.Secrets(tui3.ToolTitle(event.Tool, event.Args, "")) } - live.box.Event(appwire.EventTurnTool, b.callEnded(a, id, turn, event, title, summary, appDetail(summary), succeeded)) + live.sayTool(b.callEnded(a, id, turn, event, title, summary, appDetail(summary), succeeded)) b.noteContext(a, live) // THE SEVENTEENTH PROCESS is refused by the engine, and the turn // says so once at its end. diff --git a/cmd/codeaf/appspendlive_test.go b/cmd/codeaf/appspendlive_test.go new file mode 100644 index 0000000000..78f98aee27 --- /dev/null +++ b/cmd/codeaf/appspendlive_test.go @@ -0,0 +1,216 @@ +package main + +import ( + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "github.com/Agent-Field/codeaf/internal/appwire" + "github.com/Agent-Field/codeaf/internal/provider" + "github.com/Agent-Field/codeaf/internal/session" +) + +// Contract (a), (c), (d): a priced provider call reaches both tool phases +// during a live turn without deadlocking, while open and list keep saved spend. +func TestTurnToolCarriesLiveSpendDuringARunningTurn(t *testing.T) { + testTurnToolSpend(t, .125, .25) +} + +// Contract (b), (d): a real turn with no spend omits the JSON key at both +// tool phases and completes under the same deadline as a priced turn. +func TestTurnToolOmitsSpendWhenNothingWasSpent(t *testing.T) { + testTurnToolSpend(t, 0, 0) +} + +// Contract (a), (d): failed calls, approval updates and output refreshes carry +// the live reading at each emission, without holding the call-log lock. +func TestTurnToolRestatementsCarryTheCurrentSpend(t *testing.T) { + hold := make(chan struct{}) + defer close(hold) + conv := &fakeConversation{cost: .375, hold: hold, script: []session.Event{ + {Kind: session.EventToolBegin, CallID: "failed", Tool: "bash"}, + {Kind: session.EventToolFailed, CallID: "failed", Tool: "bash", Text: "failed"}, + {Kind: session.EventToolBegin, CallID: "long", Tool: "bash"}, + }} + backend, rec, agent := workBackend(t, conv) + backend.toolQuiet = 0 + if _, err := backend.Send(agent.ID, "read the output", ""); err != nil { + t.Fatal(err) + } + failed := toolRow(rec, t, func(row appwire.TurnTool) bool { return row.Call == "failed" && row.Phase == appwire.PhaseEnd }) + if failed.SpentUSD == nil || *failed.SpentUSD != .375 || failed.OK == nil || *failed.OK { + t.Fatalf("failed tool row = %+v", failed) + } + toolRow(rec, t, func(row appwire.TurnTool) bool { return row.Call == "long" && row.Phase == appwire.PhaseBegin }) + conv.mu.Lock() + conv.cost = .5 + conv.mu.Unlock() + conv.raise(session.Event{Kind: session.EventQuestion, Question: &session.Question{ + ID: 7, Kind: session.QuestionConsent, Head: "Allow the command?", + Subject: session.SubjectRef{Kind: session.SubjectCall, CallID: "long", Name: "bash"}, + Options: session.AnswerOptions(session.QuestionConsent), + }}) + asked := toolRow(rec, t, func(row appwire.TurnTool) bool { return row.Call == "long" && row.Ask != "" }) + if asked.SpentUSD == nil || *asked.SpentUSD != .5 { + t.Fatalf("approval tool row = %+v", asked) + } + conv.mu.Lock() + conv.cost = .75 + conv.mu.Unlock() + a, err := backend.find(agent.ID) + if err != nil { + t.Fatal(err) + } + backend.callOutput(a, agent.ID, backend.liveOf(a), session.Event{Kind: session.EventToolOutput, CallID: "long", Text: "still running\n"}) + fresh := toolRow(rec, t, func(row appwire.TurnTool) bool { return row.Call == "long" && len(row.Tail) == 1 }) + if fresh.SpentUSD == nil || *fresh.SpentUSD != .75 { + t.Fatalf("refreshed tool row = %+v", fresh) + } +} + +func testTurnToolSpend(t *testing.T, saved, cost float64) { + t.Helper() + d := newUnreadDisk(t) + meta, err := session.LoadMeta(d.dir) + if err != nil { + t.Fatal(err) + } + meta.SpentUSD = saved + if err := session.SaveMeta(d.dir, meta); err != nil { + t.Fatal(err) + } + transcript := filepath.Join(d.dir, "transcript.jsonl") + journal := fmt.Sprintf("{\"type\":\"session\",\"version\":1,\"id\":%q}\n{\"type\":\"usage\",\"usage\":{\"costUsd\":%g}}\n", d.id, saved) + if err := os.WriteFile(transcript, []byte(journal), 0o600); err != nil { + t.Fatal(err) + } + path := filepath.Join(meta.Workspace, "receipt.txt") + if err := os.WriteFile(path, []byte("the tool ran\n"), 0o600); err != nil { + t.Fatal(err) + } + args, err := json.Marshal(map[string]string{"path": path}) + if err != nil { + t.Fatal(err) + } + finish := make(chan struct{}) + var once sync.Once + release := func() { once.Do(func() { close(finish) }) } + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var request struct { + Messages []struct { + Role string `json:"role"` + } `json:"messages"` + } + if err := json.NewDecoder(r.Body).Decode(&request); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + answered := false + for _, message := range request.Messages { + answered = answered || message.Role == "tool" + } + if answered { + select { + case <-finish: + case <-r.Context().Done(): + return + } + } + w.Header().Set("Content-Type", "text/event-stream") + if answered { + fmt.Fprint(w, "data: "+`{"choices":[{"index":0,"delta":{"role":"assistant","content":"done"},"finish_reason":"stop"}],"usage":{"prompt_tokens":10,"completion_tokens":5,"cost":0}}`+"\n\n") + } else { + fmt.Fprintf(w, "data: "+`{"choices":[{"index":0,"delta":{"role":"assistant","tool_calls":[{"index":0,"id":"live-spend","type":"function","function":{"name":"read","arguments":%q}}]},"finish_reason":"tool_calls"}],"usage":{"prompt_tokens":10,"completion_tokens":5,"cost":%g}}`+"\n\n", string(args), cost) + } + fmt.Fprint(w, "data: [DONE]\n\n") + })) + t.Cleanup(server.Close) + profile := t.TempDir() + var engine *session.Agent + d.backend.open = func(ask appOpenAsk) (*appConn, error) { + var err error + engine, err = session.New(session.Config{ + Workspace: ask.Workspace, Place: session.Place{Dir: d.dir}, SessionFile: ask.Transcript, + ProfileDir: profile, Model: "test/model", APIKey: "fixture", BaseURL: server.URL, + OneModel: true, Routing: provider.RoutingOff, DeterministicNames: true, + }) + if err != nil { + return nil, err + } + return &appConn{Conv: engine, Workspace: ask.Workspace, Transcript: ask.Transcript, Release: func() { _ = engine.Close() }}, nil + } + // Release the held response before either the server or backend is closed. + t.Cleanup(release) + d.listed() + if _, err := d.backend.OpenAgent(d.id); err != nil { + t.Fatal(err) + } + if _, err := d.backend.Send(d.id, "read receipt.txt, then finish", ""); err != nil { + t.Fatal(err) + } + for _, phase := range []string{appwire.PhaseBegin, appwire.PhaseEnd} { + row := toolRow(d.rec, t, func(row appwire.TurnTool) bool { return row.Call == "live-spend" && row.Phase == phase }) + live := engine.Usage().CostUSD + if live != saved+cost { + t.Fatalf("live meter = %v, want %v", live, saved+cost) + } + if live > 0 && (row.SpentUSD == nil || *row.SpentUSD != live) { + t.Fatalf("%s tool row = %+v, want live spend %v", phase, row, live) + } + raw, err := json.Marshal(row) + if err != nil { + t.Fatal(err) + } + var fields map[string]json.RawMessage + if err := json.Unmarshal(raw, &fields); err != nil { + t.Fatal(err) + } + if _, present := fields["spent_usd"]; present != (live > 0) { + t.Fatalf("%s tool row has wrong spend presence: %s", phase, raw) + } + if phase == appwire.PhaseEnd && (row.OK == nil || !*row.OK || !strings.Contains(strings.Join(row.Tail, "\n"), "the tool ran")) { + t.Fatalf("the tool did not run successfully: %+v", row) + } + opened, err := d.backend.OpenAgent(d.id) + if err != nil || opened.SpentUSD != saved || opened.State != appwire.StateWorking { + t.Fatalf("running agents.open = %+v, %v; want saved spend %v", opened, err, saved) + } + if listed := d.listed(); listed.SpentUSD != saved { + t.Fatalf("running agents.list spend = %v, want saved spend %v", listed.SpentUSD, saved) + } + } + release() + d.rec.waitFor(t, 5*time.Second, appTurnDone) + entries, _, err := d.backend.Entries(d.id, "", 100) + if err != nil { + t.Fatal(err) + } + found := false + for _, entry := range entries { + if entry.Kind != appwire.EntryTool { + continue + } + found = true + raw, err := json.Marshal(entry) + if err != nil { + t.Fatal(err) + } + var fields map[string]json.RawMessage + if err := json.Unmarshal(raw, &fields); err != nil { + t.Fatal(err) + } + if _, present := fields["spent_usd"]; present { + t.Fatalf("history tool row carries live spend: %s", raw) + } + } + if !found { + t.Fatal("history did not replay the tool call") + } +} diff --git a/cmd/codeaf/apptools.go b/cmd/codeaf/apptools.go index 430092b0f4..815fe469a0 100644 --- a/cmd/codeaf/apptools.go +++ b/cmd/codeaf/apptools.go @@ -295,6 +295,13 @@ func (b *appBackend) rowLocked(agent string, c *appCall) appwire.TurnTool { return row } +// sayTool reads the live meter only after the caller has released its locks, +// because a local conversation's Usage takes the session agent's lock. +func (live *appLive) sayTool(row appwire.TurnTool) { + row.SpentUSD = appFigure(live.conn.Conv.Usage().CostUSD) + live.box.Event(appwire.EventTurnTool, row) +} + // factsLocked is a call's facts with its tail as it stands: three lines while // it runs, twelve once it has ended. func (c *appCall) factsLocked() appwire.ToolFacts { @@ -468,7 +475,7 @@ func (b *appBackend) callAsked(a *appAgent, agent string, live *appLive, call, a row = b.rowLocked(agent, c) }) if known { - live.box.Event(appwire.EventTurnTool, row) + live.sayTool(row) } } @@ -517,7 +524,7 @@ func (b *appBackend) tickTools(a *appAgent, agent string, live *appLive) func() } }) for _, row := range rows { - live.box.Event(appwire.EventTurnTool, row) + live.sayTool(row) } } }) diff --git a/docs/changes/unreleased/2413-turn-tool-live-spend.md b/docs/changes/unreleased/2413-turn-tool-live-spend.md new file mode 100644 index 0000000000..84acadc799 --- /dev/null +++ b/docs/changes/unreleased/2413-turn-tool-live-spend.md @@ -0,0 +1,12 @@ +--- +kind: added +title: Tool step events carry the conversation's live spend so far +pr: 2413 +surface: [app, docs] +invalidates: + - A running turn said nothing about its conversation's spend until it ended; turn.tool now carries the live figure as spent_usd. +--- + +`spent_usd` on `turn.tool` is the same live meter `agents.budget.get` answers as +`engine.spent_usd`, so it can be ahead of `Agent.spent_usd`, which is saved at +turn end. Zero or unknown readings and history rows omit it. diff --git a/docs/design/app-v3/PROTOCOL.md b/docs/design/app-v3/PROTOCOL.md index 897d253df7..b916cde008 100644 --- a/docs/design/app-v3/PROTOCOL.md +++ b/docs/design/app-v3/PROTOCOL.md @@ -1761,6 +1761,11 @@ and the reply is empty. | `turn.thinking_done` | `TurnThinkingDone` | | `turn.tool` | `TurnTool` | +`turn.tool` optionally carries `spent_usd`, the conversation's live spend when +the step was said (the same meter as `agents.budget.get`'s `engine.spent_usd`, +which can be ahead of the saved `Agent.spent_usd`); zero or unknown readings and +history rows omit it, and absence means the engine did not say. + ## Slash catalogue commands and forms `slash.catalog` lists the shared terminal vocabulary; the table is 54 commands, 93 rows, 31 aliases. diff --git a/internal/appwire/turntool_spend_test.go b/internal/appwire/turntool_spend_test.go new file mode 100644 index 0000000000..3d9988635d --- /dev/null +++ b/internal/appwire/turntool_spend_test.go @@ -0,0 +1,64 @@ +package appwire + +import ( + "encoding/json" + "testing" +) + +// Contract (a), (b): the registered event schema exposes an optional numeric +// spend, and the JSON event carries a supplied figure while omitting absence. +func TestTurnToolSpendWireContract(t *testing.T) { + out, err := SchemaJSON() + if err != nil { + t.Fatal(err) + } + var doc struct { + Events map[string]struct { + Ref string `json:"$ref"` + } `json:"events"` + Defs map[string]struct { + Properties map[string]struct { + AnyOf []struct { + Type string `json:"type"` + } `json:"anyOf"` + } `json:"properties"` + Required []string `json:"required"` + } `json:"$defs"` + } + if err := json.Unmarshal(out, &doc); err != nil { + t.Fatal(err) + } + if doc.Events[EventTurnTool].Ref != "#/$defs/TurnTool" { + t.Fatalf("turn.tool schema = %+v", doc.Events[EventTurnTool]) + } + shape := doc.Defs["TurnTool"] + alts := shape.Properties["spent_usd"].AnyOf + if len(alts) != 2 || alts[0].Type != "number" || alts[1].Type != "null" { + t.Fatalf("TurnTool.spent_usd schema = %+v", alts) + } + for _, field := range shape.Required { + if field == "spent_usd" { + t.Fatal("TurnTool.spent_usd became required") + } + } + if _, present := doc.Defs["Entry"].Properties["spent_usd"]; present { + t.Fatal("history entries acquired live spend") + } + usd := .375 + for _, figure := range []*float64{nil, &usd} { + raw, err := json.Marshal(TurnTool{Agent: "a", Turn: 1, Call: "c", Name: "read", Phase: PhaseBegin, SpentUSD: figure}) + if err != nil { + t.Fatal(err) + } + var fields map[string]json.RawMessage + if err := json.Unmarshal(raw, &fields); err != nil { + t.Fatal(err) + } + if _, present := fields["spent_usd"]; present != (figure != nil) { + t.Fatalf("wrong spend presence: %s", raw) + } + if figure != nil && string(fields["spent_usd"]) != "0.375" { + t.Fatalf("wrong spend figure: %s", raw) + } + } +} diff --git a/internal/appwire/wire.go b/internal/appwire/wire.go index 65c67016a1..ab8c77d6e5 100644 --- a/internal/appwire/wire.go +++ b/internal/appwire/wire.go @@ -1018,6 +1018,11 @@ type ( Phase string `json:"phase"` OK *bool `json:"ok,omitempty"` Summary string `json:"summary,omitempty"` + // SpentUSD is the conversation's live spend when this step was said, + // the same meter agents.budget.get calls engine.spent_usd. It can be + // ahead of Agent.SpentUSD, saved at turn end, and is omitted when zero + // or unknown. + SpentUSD *float64 `json:"spent_usd,omitempty"` ToolFacts } TurnDone struct { diff --git a/internal/manual/chat/running-from-the-terminal.md b/internal/manual/chat/running-from-the-terminal.md index e0ab93c4f7..0d409f6e9e 100644 --- a/internal/manual/chat/running-from-the-terminal.md +++ b/internal/manual/chat/running-from-the-terminal.md @@ -460,6 +460,9 @@ and twelve once it ends. A command's whole output is kept on this Mac for a week under the app's folder in codeaf's home, so it can be opened after the reply ends, or after the app restarts. +Live step events also carry the conversation's spend so far when it is known +and positive; the saved conversation total can catch up after the reply ends. + ## The desktop app's conversation — steer or queue, the decision dock, and notes to a task In the desktop app a message typed while the agent works can steer or queue. From 4703823519f1c4d4dde94c8feb9f7a00bd53db39 Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Thu, 12 Nov 2026 11:56:54 -0500 Subject: [PATCH 2/6] A resumed conversation restores its late receipts, so it never saves less than it charged Carries #2414 from the closed integration PR #2306 (workstreams/main); it never reached dev. The one conflict, Digest's fields, keeps dev's folded-usage total beside the late-receipt sum. Co-Authored-By: Claude Mythos 5.2 --- .../unreleased/2414-replay-late-receipts.md | 12 + internal/manual/chat/sessions-and-rewind.md | 10 +- internal/session/agent.go | 9 +- internal/session/digest.go | 22 +- internal/session/loop.go | 5 +- internal/session/placemeta.go | 12 +- internal/session/sessionfile.go | 69 +++-- internal/session/spend_replay_test.go | 245 ++++++++++++++++++ 8 files changed, 338 insertions(+), 46 deletions(-) create mode 100644 docs/changes/unreleased/2414-replay-late-receipts.md create mode 100644 internal/session/spend_replay_test.go diff --git a/docs/changes/unreleased/2414-replay-late-receipts.md b/docs/changes/unreleased/2414-replay-late-receipts.md new file mode 100644 index 0000000000..4cef8d6e70 --- /dev/null +++ b/docs/changes/unreleased/2414-replay-late-receipts.md @@ -0,0 +1,12 @@ +--- +kind: fixed +title: A resumed conversation keeps its late receipts and never saves a lower spend +pr: 2414 +surface: [chat, engine, app] +invalidates: + - Replay restored a conversation's spend from usage seals only, so a restart forgot late receipts and the next turn could save a lower total. + - Journal call lines were never read back as spend; a call line marked as a late receipt now restores one call and its charge. +--- + +Ordinary call lines remain evidence only. A meta row rebuilt from the +transcript counts the same late receipts. diff --git a/internal/manual/chat/sessions-and-rewind.md b/internal/manual/chat/sessions-and-rewind.md index 41e279334c..27505cf9f3 100644 --- a/internal/manual/chat/sessions-and-rewind.md +++ b/internal/manual/chat/sessions-and-rewind.md @@ -310,7 +310,8 @@ exactly `[image /path/to/file — file changed or gone]`. Yes. Every piece of work that cost money appends a `usage` line to the transcript, carrying the model it ran on, tokens in and out, cache read and cache write, the money, how many -requests it took and how long it lasted. +requests it took and how long it lasted. A late receipt writes a marked `call` line instead +(below). There is one line per **completed turn**, and one more for each call made beside a turn — naming the conversation, the memory reflex, the check on a reply, a task's fold-up, a @@ -323,7 +324,8 @@ One exception to "one line per turn": a turn that was answered by **more than on model's own tokens and money. A single sum could only name one of them, and naming the one standing at the end would hand the whole turn's bill to the wrong model. -**On a resume these lines are added up, and that sum is the session's totals.** So `/cost`, +**On a resume these lines, and the marked late-receipt `call` lines below, are added up, +and that sum is the session's totals.** So `/cost`, `/status` and the status line show what the whole conversation has spent across every restart, not only what has happened since you reopened it. The spend ceiling reads the same sum — see the spending pages for what that means the day after you reach one. @@ -354,11 +356,11 @@ provider's own **receipt** is fetched afterwards, the request writes its line th with `arm`: `hedge` for a rescue arm, `reconciled` for any other late receipt. A receipt names no endpoint, so the line leaves that field out rather than guess. -These lines are **evidence, never spending**. Nothing adds them up, `/cost` does not read +Ordinary call lines are **evidence, never spending**. Nothing adds them up, `/cost` does not read them, and a resume ignores them entirely, because every dollar on them is already in the `usage` line that sealed the turn. They are there so a question about *shape* — what does a mostly-cached request actually cost here — can be read out of the file instead of guessed -at. +at. A line with an `arm` is the exception: its money is in no seal, so a resume adds it once. ## Picking up where you left off diff --git a/internal/session/agent.go b/internal/session/agent.go index 24bd5fe384..8265818f34 100644 --- a/internal/session/agent.go +++ b/internal/session/agent.go @@ -417,9 +417,10 @@ func newAgent(config Config, client Completer) (*Agent, error) { // not reachable yet. agent.scrubBlindImagePartsLocked(agent.model) // AND IT KEEPS WHAT IT SPENT. The total is the sum of the file's usage - // lines (sessionfile.go), so a conversation reopened tomorrow reports the - // money it actually cost rather than the money this process has spent so - // far — which for a resumed session was always nothing. + // lines and late-receipt call lines (sessionfile.go), so a conversation + // reopened tomorrow reports the money it actually cost rather than the + // money this process has spent so far — which for a resumed session was + // always nothing. // // THIS ALSO RESTORES THE SPEND RAIL, and that is the intended change: the // rail is this conversation's own ceiling (rail.go), so a conversation @@ -427,7 +428,7 @@ func newAgent(config Config, client Completer) (*Agent, error) { // the process was a ceiling on a window, not on a conversation. agent.usage = file.RestoredUsage() // AND A JOURNAL WRITTEN BEFORE THE TOTAL WAS KEPT GETS ONE NOW. The sum - // above came out of the file's own usage lines, so this costs the read + // above came out of the file's own charge records, so this costs the read // of a meta.json and — only when it has no figure at all — one write // (placemeta.go's [Agent.stampRestoredSpend]). Without it a conversation // that has not spoken since the field arrived would read as free on diff --git a/internal/session/digest.go b/internal/session/digest.go index 4ac2894d72..46dc8e87ce 100644 --- a/internal/session/digest.go +++ b/internal/session/digest.go @@ -31,7 +31,12 @@ type Digest struct { // Usage is one row per directly billed usage line, in file order. Usage []UsageLine - folded Usage + folded Usage + // Late receipts were banked outside every usage seal. Only Spend adds them; + // Usage stays the usage-line list its ledger readers already know. + lateUSD float64 + lateTokens int + placeholder string } @@ -44,6 +49,7 @@ var digestFolds = map[string]foldFunc{ "session": foldHeader, "title": foldTitle, "usage": foldUsage, + "call": foldCall, "message": foldMessage, } @@ -125,6 +131,14 @@ func foldUsage(d *Digest, _ []byte, e sessionEntry) { }) } +func foldCall(d *Digest, _ []byte, e sessionEntry) { + if e.Call == nil || e.Call.Arm == "" { + return + } + d.lateUSD += e.Call.CostUSD + d.lateTokens += e.Call.Input + e.Call.Output +} + func foldMessage(d *Digest, _ []byte, e sessionEntry) { if e.Role != "user" || e.Note { return @@ -140,15 +154,15 @@ func parseStamp(s string) time.Time { return t } -// Spend is the transcript's own total: money and input plus output tokens, the -// two figures [Meta] stamps. +// Spend is the transcript's own total: usage seals plus late-receipt call +// lines, in money and input plus output tokens, the two figures [Meta] stamps. func (d Digest) Spend() (usd float64, tokens int) { usd, tokens = d.folded.CostUSD, d.folded.Input+d.folded.Output for _, u := range d.Usage { usd += u.USD tokens += u.Input + u.Output } - return usd, tokens + return usd + d.lateUSD, tokens + d.lateTokens } // Derive lays what the transcript says over m and returns it. Fields the diff --git a/internal/session/loop.go b/internal/session/loop.go index e30855c0e0..c8d6676de1 100644 --- a/internal/session/loop.go +++ b/internal/session/loop.go @@ -4654,8 +4654,9 @@ func (a *Agent) reconciled(receipt provider.Reconciled) { // way: EVERY request this session makes writes one, and a request whose // usage block never arrived wrote none — so the journal's call lines summed // to barely half the measured session's bill, and exactly the expensive - // half was missing. The row is evidence and never spend; the money moved - // through the door above and nowhere else. + // half was missing. The live money moved through the door above and nowhere + // else; this row is the only record a replay can restore it from, because + // late money belongs to no turn seal. a.file.appendCall(armCall(receipt)) // A cancelled turn has already sealed, so its late charge is saved by // [Agent.bank]'s stamp above rather than waiting for another turn. diff --git a/internal/session/placemeta.go b/internal/session/placemeta.go index f0563dc96c..4320d6ea21 100644 --- a/internal/session/placemeta.go +++ b/internal/session/placemeta.go @@ -564,12 +564,12 @@ func isAnswer(message ai.Message) bool { // journal written before this field existed. // // IT FOLDS NOTHING AND READS NOTHING EXTRA. The resume already replayed the -// file and already summed its usage lines ([sessionFile.RestoredUsage], which -// is what the agent's live counters are restored from — agent.go), so this -// takes that sum and writes it once. A session with nothing restored, or one -// whose meta already carries a figure, writes nothing at all: the stamp above -// keeps it true from here on, and a rewrite per open would be a file touched by -// every window that merely looked. +// file and already summed its usage lines and late-receipt call lines +// ([sessionFile.RestoredUsage], which is what the agent's live counters are +// restored from — agent.go), so this takes that sum and writes it once. A +// session with nothing restored, or one whose meta already carries a figure, +// writes nothing at all: the stamp above keeps it true from here on, and a +// rewrite per open would be a file touched by every window that merely looked. func (a *Agent) stampRestoredSpend(restored Usage) { a.spendStampMu.Lock() defer a.spendStampMu.Unlock() diff --git a/internal/session/sessionfile.go b/internal/session/sessionfile.go index 3b0922f57e..d8ec49b980 100644 --- a/internal/session/sessionfile.go +++ b/internal/session/sessionfile.go @@ -502,10 +502,10 @@ type journalCreated struct { // journalCall is what ONE provider response reported, on its own line. // -// IT IS EVIDENCE AND NEVER SPEND. The seal above already carries every one of -// these numbers, summed; a replay that added these lines too would bill the -// session twice for the same calls. Nothing reads them back into the session's -// totals, and [replaySessionFile] says so where it drops them. +// AN ORDINARY ROW IS EVIDENCE AND NEVER SPEND. The seal above already carries +// every one of these numbers, summed; a replay that added these lines too would +// bill the session twice for the same calls. Only a late receipt, marked by +// Arm below, restores money from its own row rather than from a seal. // // It exists because a turn is sixty-odd requests with wildly different shapes — // a cold first call, then fifty that are almost all cache read — and the sum of @@ -541,9 +541,10 @@ type journalCreated struct { // fires when the first answer is slow, a receipt is fetched when a stream was // cut — and a journal that records every cheap call and none of the dear ones // answers a cost autopsy wrongly: the measured session's call lines summed to -// barely half its bill until the arms wrote theirs. The money is already banked -// by the reconciliation itself, so the row is evidence here and never spend, -// the same as every other line of this shape. +// barely half its bill until the arms wrote theirs. The money was banked LATE, +// outside every turn seal, so this row is the only record a replay has of that +// charge. Restoring it adds no turn: the receipt is a call, not a new step of +// the conversation. type journalCall struct { Model string `json:"model,omitempty"` Endpoint string `json:"endpoint,omitempty"` @@ -1242,7 +1243,8 @@ type sessionFile struct { stamps *stampIndex // restored is what this conversation had already spent when the file was - // opened: the SUM of its usage lines, replayed once and never updated after. + // opened: the SUM of its usage lines and late-receipt call lines, replayed + // once and never updated after. // It is the file's answer to "what did this cost before today", and the agent // seeds its own running total from it at construction (agent.go). Nothing // stores a second copy of the total — the lines are the record, and this is @@ -1251,7 +1253,8 @@ type sessionFile struct { } // RestoredUsage is what the conversation in this file had spent before it was -// opened, and the zero Usage for a file that is new or holds no usage lines. +// opened, and the zero Usage for a file that is new or holds no usage lines or +// late-receipt call lines. func (s *sessionFile) RestoredUsage() Usage { if s == nil { return Usage{} @@ -2203,9 +2206,9 @@ func readJournal(reader io.Reader, path string, rebuild bool) (replayedSession, messages = messages[:len(messages)-entry.Dropped] reasoning = reasoning[:len(reasoning)-entry.Dropped] case "usage": - // EVERY line is added, and none is ever taken back. This is the one - // arm that accumulates rather than rebuilds: a compaction below - // replaces the message window, and money spent before it stays spent. + // EVERY line is added, and none is ever taken back. Charges accumulate + // rather than rebuild: a compaction below replaces the message window, + // and money spent before it stays spent. if entry.Usage == nil { continue } @@ -2234,18 +2237,30 @@ func readJournal(reader io.Reader, path string, rebuild bool) (replayedSession, spent.Turns += used.Calls } case "call": - // NEVER BILLED, and this arm exists to say so rather than to leave it - // to the switch falling off the end. A call line is the SHAPE of one - // request — who served it, how much of its prompt was warm — and every - // dollar on it is already counted in the seal that closed its turn. - // Folding it into `spent` would bill the session twice for the same - // money. + // Ordinary call lines are NEVER BILLED, and this arm exists to say so + // rather than to leave it to the switch falling off the end. A call line + // is the SHAPE of one request — who served it, how much of its prompt + // was warm — and every ordinary dollar on it is already counted in the + // seal that closed its turn. Folding it into `spent` would bill the + // session twice for the same money. + // + // AN ARM IS THE EXCEPTION: its money was banked late, outside every + // seal, so this line is the only record a replay has of that charge. + // It restores one call and no turn, exactly as [Agent.reconciled] banked. // // IT IS READ FOR ITS SHAPE, which is what it was written for: the // size of each request and what came back from it, for a page that // watches this work with no lane to it ([RequestBooks]). if entry.Call != nil { requests.observe(*entry.Call) + if call := entry.Call; call.Arm != "" { + spent.Input += call.Input + spent.Output += call.Output + spent.CacheRead += call.CacheRead + spent.CacheWrite += call.CacheWrite + spent.CostUSD += call.CostUSD + spent.Calls++ + } } case "error": // DROPPED ON PURPOSE, for the reason a call line is, and one of its @@ -3259,8 +3274,8 @@ func (s *sessionFile) appendUsage(used Usage, model string, aux bool, role strin }) } -// appendCall writes ONE response's own accounting down, beside the seal that -// will sum it. +// appendCall writes ONE response's own accounting down. An ordinary response +// sits beside the seal that will sum it; a late receipt is its own record. // // A RESPONSE THAT REPORTED NO USAGE WRITES NOTHING. The emptiness law, and the // same test the seal keeps: a call with no tokens and no cost is a call the @@ -3453,7 +3468,8 @@ func (s *sessionFile) writeLine(entry any) bool { // records people share — the call log and the debug record — not here. payload = trace.ScrubRegistered(payload) entryLine, _ := entry.(sessionEntry) - if !s.appendLine(payload, entryLine.Type == "usage") { + paid := entryLine.Type == "usage" || entryLine.Type == "call" && entryLine.Call != nil && entryLine.Call.Arm != "" + if !s.appendLine(payload, paid) { return false } if s.onWrite != nil { @@ -3462,15 +3478,16 @@ func (s *sessionFile) writeLine(entry any) bool { return true } -func (s *sessionFile) appendLine(payload []byte, usage bool) bool { +func (s *sessionFile) appendLine(payload []byte, paid bool) bool { s.mu.Lock() defer s.mu.Unlock() if s.closed { - if !usage { + if !paid { return false } - // Paid work can outlive its stage. Keep its usage in the same journal - // under this file's mutex so reopening restores the full paid total. + // Paid work can outlive its stage. Keep its usage seals and late arm + // receipts in the same journal under this file's mutex so reopening + // restores the full paid total. file, err := os.OpenFile(s.file.Name(), os.O_WRONLY|os.O_APPEND, 0600) if err != nil { return false @@ -3491,7 +3508,7 @@ func (s *sessionFile) appendLine(payload []byte, usage bool) bool { // flock on its own — but it states the release at the place a reader looks for // it, and it puts the release before the close rather than as a side effect of // it. Nothing can slip into the gap: closed is already set, so no write reaches -// the file after this point, except paid usage appended by path. +// the file after this point, except paid accounting appended by path. func (s *sessionFile) Close() error { s.mu.Lock() defer s.mu.Unlock() diff --git a/internal/session/spend_replay_test.go b/internal/session/spend_replay_test.go new file mode 100644 index 0000000000..115c9c3bf0 --- /dev/null +++ b/internal/session/spend_replay_test.go @@ -0,0 +1,245 @@ +package session + +import ( + "context" + "fmt" + "math" + "path/filepath" + "testing" + "time" + + "github.com/Agent-Field/agentfield/sdk/go/ai" + "github.com/Agent-Field/codeaf/internal/provider" +) + +// Contract (a), (c): late reconciled and hedge calls restore their unsealed +// money and tokens as calls, never turns; ordinary call lines stay evidence. +func TestLateCallLinesRestoreSpendWithoutAddingTurns(t *testing.T) { + for _, test := range []struct { + name string + arm string + want Usage + }{ + { + name: "reconciled", arm: "reconciled", + want: Usage{Input: 90, Output: 25, CacheRead: 33, CacheWrite: 9, CostUSD: .375, Calls: 2, Turns: 1, Duration: 250 * time.Millisecond}, + }, + { + name: "hedge", arm: "hedge", + want: Usage{Input: 90, Output: 25, CacheRead: 33, CacheWrite: 9, CostUSD: .375, Calls: 2, Turns: 1, Duration: 250 * time.Millisecond}, + }, + { + name: "ordinary", + want: Usage{Input: 10, Output: 5, CacheRead: 3, CacheWrite: 2, CostUSD: .25, Calls: 1, Turns: 1, Duration: 250 * time.Millisecond}, + }, + } { + t.Run(test.name, func(t *testing.T) { + path := filepath.Join(t.TempDir(), "transcript.jsonl") + write(t, path, + `{"type":"session","version":1,"id":"abc","cwd":"/w","model":"m","timestamp":"t"}`, + `{"type":"call","call":{"model":"m","input":10,"output":5,"cacheRead":3,"cacheWrite":2,"costUsd":0.25}}`, + `{"type":"usage","usage":{"model":"m","input":10,"output":5,"cacheRead":3,"cacheWrite":2,"costUsd":0.25,"calls":1,"durationMs":250}}`, + fmt.Sprintf(`{"type":"call","call":{"model":"m","arm":%q,"input":80,"output":20,"cacheRead":30,"cacheWrite":7,"costUsd":0.125}}`, test.arm), + ) + file, _, err := openSessionFile(path, "/w", "m", "") + if err != nil { + t.Fatalf("open: %v", err) + } + t.Cleanup(func() { _ = file.Close() }) + if got := file.RestoredUsage(); got != test.want { + t.Fatalf("restored usage = %+v, want %+v", got, test.want) + } + }) + } +} + +// Contract (b): a cancelled .25 turn saves .375 after its late .125 receipt; +// reopening the same folder and spending .01 saves .385 rather than .26. +func TestALateReceiptSurvivesResumeAndTheNextSpendStamp(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "transcript.jsonl") + ledger := filepath.Join(t.TempDir(), UsageLedgerName) + held := make(chan struct{}) + first, workspace := newTestAgent(t, &scriptedCompleter{steps: []step{ + func(context.Context, []ai.Message) (*ai.Response, error) { + return priced(toolResponse("first", "ls", `{"path":"."}`), .25), nil + }, + func(ctx context.Context, _ []ai.Message) (*ai.Response, error) { + close(held) + <-ctx.Done() + return nil, ctx.Err() + }, + }}, func(config *Config) { + config.Place = Place{Dir: dir, Workspace: config.Workspace} + config.SessionFile = path + config.usageLedger = ledger + }) + events := mustSubmit(t, first, "look around") + waitMetaPatch(t, held) + first.Interrupt() + collect(t, events) + if meta := stampedMeta(t, first, dir); meta.SpentUSD != .25 || meta.Tokens != 27 { + t.Fatalf("cancelled turn saved %+v, want .25 and 27 tokens", meta) + } + if seals := turnUsageLines(t, path); len(seals) != 1 || seals[0].CostUSD != .25 { + t.Fatalf("cancelled turn seals = %+v, want one .25 seal", seals) + } + first.reconciled(provider.Reconciled{ + Billed: provider.Billed{Model: "test/model", PromptTokens: 80, CompletionTokens: 20, CachedTokens: 30, Cost: .125}, + Ref: "late-before-resume", Reason: "torn", Found: true, + }) + if meta := stampedMeta(t, first, dir); meta.SpentUSD != .375 || meta.Tokens != 127 { + t.Fatalf("late receipt saved %+v, want .375 and 127 tokens", meta) + } + if err := first.Close(); err != nil { + t.Fatalf("close before resume: %v", err) + } + + second, err := newAgent(Config{ + Workspace: workspace, Model: "test/model", System: "SYSTEM", + Place: Place{Dir: dir, Workspace: workspace}, SessionFile: path, usageLedger: ledger, + }, &scriptedCompleter{steps: []step{ + func(context.Context, []ai.Message) (*ai.Response, error) { + return priced(textResponse("done"), .01), nil + }, + }}) + if err != nil { + t.Fatalf("resume: %v", err) + } + t.Cleanup(second.SettleWrites) + t.Cleanup(func() { _ = second.Close() }) + if used := second.Usage(); used.CostUSD != .375 || used.Input != 100 || used.Output != 27 || used.CacheRead != 30 || used.Calls != 2 || used.Turns != 1 { + t.Errorf("resumed meter = %+v, want the seal plus its late receipt", used) + } + collect(t, mustSubmit(t, second, "finish up")) + meta := stampedMeta(t, second, dir) + if math.Abs(meta.SpentUSD-.385) > 1e-12 || meta.Tokens != 142 { + t.Fatalf("next turn saved %+v, want .385 and 142 tokens after resume", meta) + } + if used := second.Usage(); math.Abs(used.CostUSD-.385) > 1e-12 || used.Calls != 3 || used.Turns != 2 { + t.Fatalf("next turn meter = %+v, want .385, three calls and two conversation steps", used) + } +} + +// Contract (b), review 1 major finding: a receipt pending across Close must +// keep its paid arm line, so a resume restores .375 and the next turn saves +// .385. Ordinary call lines must still be refused after close. +func TestALateReceiptAfterCloseSurvivesResumeAndTheNextSpendStamp(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "transcript.jsonl") + ledger := filepath.Join(t.TempDir(), UsageLedgerName) + t.Cleanup(func() { StopUsageWriter(ledger) }) + held := make(chan struct{}) + first, workspace := newTestAgent(t, &scriptedCompleter{steps: []step{ + func(context.Context, []ai.Message) (*ai.Response, error) { + return priced(toolResponse("first", "ls", `{"path":"."}`), .25), nil + }, + func(ctx context.Context, _ []ai.Message) (*ai.Response, error) { + close(held) + <-ctx.Done() + return nil, ctx.Err() + }, + }}, func(config *Config) { + config.Place = Place{Dir: dir, Workspace: config.Workspace} + config.SessionFile = path + config.usageLedger = ledger + }) + events := mustSubmit(t, first, "look around") + waitMetaPatch(t, held) + settled := first.pendingReceipt() + first.Interrupt() + collect(t, events) + if meta := stampedMeta(t, first, dir); meta.SpentUSD != .25 || meta.Tokens != 27 { + t.Fatalf("cancelled turn saved %+v, want .25 and 27 tokens", meta) + } + if seals := turnUsageLines(t, path); len(seals) != 1 || seals[0].CostUSD != .25 { + t.Fatalf("cancelled turn seals = %+v, want one .25 seal", seals) + } + if err := first.Close(); err != nil { + t.Fatalf("close with a pending receipt: %v", err) + } + lines := len(readLines(t, path)) + if first.file.writeLine(sessionEntry{ + Type: "call", Call: &journalCall{Model: "test/model", Input: 1, Output: 1, CostUSD: 1}, Timestamp: stamp(), + }) { + t.Error("ordinary call line was accepted after close") + } + if got := len(readLines(t, path)); got != lines { + t.Fatalf("ordinary call changed the closed journal from %d to %d lines", lines, got) + } + first.reconciled(provider.Reconciled{ + Billed: provider.Billed{Model: "test/model", PromptTokens: 80, CompletionTokens: 20, CachedTokens: 30, Cost: .125}, + Ref: "late-after-close", Reason: "torn", Found: true, + }) + settled() + if meta := stampedMeta(t, first, dir); meta.SpentUSD != .375 || meta.Tokens != 127 { + t.Fatalf("post-close receipt saved %+v, want .375 and 127 tokens", meta) + } + if seals := turnUsageLines(t, path); len(seals) != 1 || seals[0].CostUSD != .25 { + t.Fatalf("post-close receipt changed turn seals: %+v", seals) + } + digest, err := ReadDigest(path) + if err != nil { + t.Fatalf("digest after close: %v", err) + } + if usd, tokens := digest.Spend(); usd != .375 || tokens != 127 { + t.Errorf("post-close digest spend = %v / %d tokens, want .375 / 127", usd, tokens) + } + + second, err := newAgent(Config{ + Workspace: workspace, Model: "test/model", System: "SYSTEM", + Place: Place{Dir: dir, Workspace: workspace}, SessionFile: path, usageLedger: ledger, + }, &scriptedCompleter{steps: []step{ + func(context.Context, []ai.Message) (*ai.Response, error) { + return priced(textResponse("done"), .01), nil + }, + }}) + if err != nil { + t.Fatalf("resume after post-close receipt: %v", err) + } + t.Cleanup(second.SettleWrites) + t.Cleanup(func() { _ = second.Close() }) + if used := second.Usage(); used.CostUSD != .375 || used.Input != 100 || used.Output != 27 || used.CacheRead != 30 || used.Calls != 2 || used.Turns != 1 { + t.Errorf("resumed meter = %+v, want the seal plus its post-close receipt", used) + } + collect(t, mustSubmit(t, second, "finish up")) + meta := stampedMeta(t, second, dir) + if math.Abs(meta.SpentUSD-.385) > 1e-12 || meta.Tokens != 142 { + t.Fatalf("next turn saved %+v, want .385 and 142 tokens after the post-close receipt", meta) + } + if used := second.Usage(); math.Abs(used.CostUSD-.385) > 1e-12 || used.Calls != 3 || used.Turns != 2 { + t.Fatalf("next turn meter = %+v, want .385, three calls and two conversation steps", used) + } +} + +// Contract (d), (c): the digest adds late reconciled and hedge receipts to +// saved spend while Usage still contains only the seal, never the call rows. +func TestDigestSpendIncludesLateReceiptsWithoutAddingUsageLines(t *testing.T) { + for _, arm := range []string{"reconciled", "hedge"} { + t.Run(arm, func(t *testing.T) { + path := filepath.Join(t.TempDir(), "transcript.jsonl") + write(t, path, + `{"type":"session","version":1,"id":"abc","cwd":"/w","model":"m","timestamp":"t"}`, + `{"type":"call","call":{"model":"m","input":10,"output":5,"cacheRead":3,"cacheWrite":2,"costUsd":0.25}}`, + `{"type":"usage","usage":{"model":"m","input":10,"output":5,"cacheRead":3,"cacheWrite":2,"costUsd":0.25,"calls":1}}`, + fmt.Sprintf(`{"type":"call","call":{"model":"m","arm":%q,"input":80,"output":20,"cacheRead":30,"cacheWrite":7,"costUsd":0.125}}`, arm), + ) + digest, err := ReadDigest(path) + if err != nil { + t.Fatalf("digest: %v", err) + } + if usd, tokens := digest.Spend(); usd != .375 || tokens != 115 { + t.Fatalf("digest spend = %v / %d tokens, want .375 / 115", usd, tokens) + } + if len(digest.Usage) != 1 { + t.Fatalf("digest Usage = %+v, want only the one usage seal", digest.Usage) + } + if used := digest.Usage[0]; used.USD != .25 || used.Input != 10 || used.Output != 5 || used.Calls != 1 { + t.Fatalf("digest Usage row = %+v, want the seal's figures alone", used) + } + if meta := digest.Derive(Meta{}); meta.SpentUSD != .375 || meta.Tokens != 115 { + t.Fatalf("derived metadata = %+v, want .375 and 115 tokens", meta) + } + }) + } +} From c9b6f05aa14f0754da77a077162d0bd792cc14c1 Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Thu, 12 Nov 2026 11:59:04 -0500 Subject: [PATCH 3/6] session: an abandoned turn's paid steps survive a restart Carries #2440 from the closed integration PR #2306 (workstreams/main); it never reached dev. Co-Authored-By: Claude Mythos 5.2 --- .../2440-abandoned-turn-step-seal.md | 13 + internal/manual/chat/sessions-and-rewind.md | 8 +- internal/session/abandon.go | 11 +- internal/session/abandon_spend_test.go | 464 ++++++++++++++++++ internal/session/loop.go | 52 +- internal/session/sessionfile.go | 32 +- 6 files changed, 535 insertions(+), 45 deletions(-) create mode 100644 docs/changes/unreleased/2440-abandoned-turn-step-seal.md create mode 100644 internal/session/abandon_spend_test.go diff --git a/docs/changes/unreleased/2440-abandoned-turn-step-seal.md b/docs/changes/unreleased/2440-abandoned-turn-step-seal.md new file mode 100644 index 0000000000..4d9f4e92bb --- /dev/null +++ b/docs/changes/unreleased/2440-abandoned-turn-step-seal.md @@ -0,0 +1,13 @@ +--- +kind: fixed +title: A turn stopped at the second stage keeps its paid steps across a restart +pr: 2440 +surface: [chat, engine] +invalidates: + - "Was true: a turn let go of at the stop bound recorded its spend only on its abandoned line, which replay never bills, so a restart could save a lower total. Is true now: the turn's paid conversation steps are written as a usage seal without duration, and replay restores them." +--- + +The abandoned line stays evidence and is never billed; errands on the turn's clock +keep their own usage lines. A settle turn stopping at its bound and a final stage +report now write the same step seal, without duration. Transcripts written +before this change cannot recover an abandoned turn's missing steps. diff --git a/internal/manual/chat/sessions-and-rewind.md b/internal/manual/chat/sessions-and-rewind.md index 27505cf9f3..a4e2e8e54c 100644 --- a/internal/manual/chat/sessions-and-rewind.md +++ b/internal/manual/chat/sessions-and-rewind.md @@ -315,9 +315,11 @@ requests it took and how long it lasted. A late receipt writes a marked `call` l There is one line per **completed turn**, and one more for each call made beside a turn — naming the conversation, the memory reflex, the check on a reply, a task's fold-up, a -planner, a harness run. Those are marked `aux` so a reader can tell the turn itself from -the work done around it. A turn that spent nothing writes no line at all, because a zero is -absence, not a fact worth a line. +planner, a harness run. Those are marked `aux`. A second-stage stop that lets go of a +turn while tools still run also saves its paid conversation steps in a `usage` seal, +without duration. Its `abandoned` line records the stop and last-known spend as evidence; +it is never added to the bill. A turn that spent nothing writes no usage line. +Older transcripts without that step seal cannot restore the missing steps. One exception to "one line per turn": a turn that was answered by **more than one model** — the model hopped mid-turn — writes **one line per model that answered**, each with that diff --git a/internal/session/abandon.go b/internal/session/abandon.go index b62f39d99c..0fbbeacb41 100644 --- a/internal/session/abandon.go +++ b/internal/session/abandon.go @@ -37,9 +37,10 @@ import ( // // THE LINE IS THE ACCEPTANCE AND NOT THE DECORATION. A turn that is let go of // while it may still be spending is a turn nobody could otherwise account for: -// the seal that carries a turn's cost is written when a turn ENDS, and the whole -// definition of an abandoned turn is that nobody waited for its end. The money -// itself already reaches the machine's ledger per call rather than per turn +// the abandoned line records the last-known spend while nobody waits for the end. +// The displaced loop still writes its conversation-step usage seal on exit; +// errands already have their own seals. This line remains evidence only. The +// money itself reaches the machine's ledger per call rather than per turn // (usage_ledger.go, issue #269), so the bill is not lost — what was lost was any // record that a turn had been let go of at all. // @@ -272,8 +273,8 @@ const toolAbandonedWord = "the person stopped this turn and it was not waited fo // abandoned reports that the turn this context belongs to has been let go of. // -// IT IS A NON-BLOCKING READ and it is the answer to "may I still write?". A turn -// that has been abandoned must not touch the session's transcript again: the +// IT IS A NON-BLOCKING READ and it answers "may I still write messages?". A turn +// that has been abandoned must not append to the session's messages again: the // session has moved on and a.messages belongs to whatever turn the person // started next, so an append from here is one conversation growing another's // rows. False is the ordinary answer and false is what every context with no diff --git a/internal/session/abandon_spend_test.go b/internal/session/abandon_spend_test.go new file mode 100644 index 0000000000..9fce93a0ff --- /dev/null +++ b/internal/session/abandon_spend_test.go @@ -0,0 +1,464 @@ +package session + +import ( + "context" + "encoding/json" + "math" + "path/filepath" + "reflect" + "sync" + "testing" + "time" + + "github.com/Agent-Field/agentfield/sdk/go/ai" + "github.com/Agent-Field/codeaf/internal/provider" +) + +// Contract A1, A2: abandoned steps survive a restart, and a later stamp grows +// from .26 to .27 rather than forgetting the abandoned .25. +func TestAnAbandonedTurnsStepSpendSurvivesResumeAndTheNextStamp(t *testing.T) { + first, entered, dir, path, ledger := abandonSpendAgent(t, &scriptedCompleter{steps: []step{ + func(context.Context, []ai.Message) (*ai.Response, error) { + return priced(toolResponse("paid", "wedge", `{}`), .25), nil + }, + func(context.Context, []ai.Message) (*ai.Response, error) { + return priced(textResponse("after the stop"), .01), nil + }, + }}) + events := mustSubmit(t, first, "park") + waitMetaPatch(t, entered) + first.Interrupt() + if _, ok := first.Abandon(AbandonStopTimeout); !ok { + t.Fatal("the parked turn was not abandoned") + } + collect(t, events) + waitAbandonSpendExit(t, path, 1) + seals := turnUsageLines(t, path) + if len(seals) != 1 || seals[0].CostUSD != .25 || seals[0].DurationMS != 0 { + t.Fatalf("abandoned seals = %+v, want one .25 seal with no duration", seals) + } + if lines := abandonedLines(t, path); len(lines) != 1 { + t.Fatalf("abandoned lines = %+v, want exactly one", lines) + } + collect(t, mustSubmit(t, first, "continue")) + before := stampedMeta(t, first, dir) + if math.Abs(before.SpentUSD-.26) > 1e-12 || before.Tokens != 42 { + t.Fatalf("before restart saved %+v, want .26 and 42 tokens", before) + } + live := first.Usage() + if math.Abs(live.CostUSD-.26) > 1e-12 || live.Input != 30 || live.Output != 12 || live.Calls != 2 || live.Turns != 2 { + t.Fatalf("live meter = %+v, want .26, 30/12 tokens, two calls and two conversation steps", live) + } + if err := first.Close(); err != nil { + t.Fatalf("close before restart: %v", err) + } + second, err := newAgent(Config{ + Workspace: first.config.Workspace, Model: "test/model", System: "SYSTEM", + Place: Place{Dir: dir, Workspace: first.config.Workspace}, SessionFile: path, usageLedger: ledger, + }, &scriptedCompleter{steps: []step{ + func(context.Context, []ai.Message) (*ai.Response, error) { + return priced(textResponse("done"), .01), nil + }, + }}) + if err != nil { + t.Fatalf("resume: %v", err) + } + t.Cleanup(second.SettleWrites) + t.Cleanup(func() { _ = second.Close() }) + assertAbandonSpendMeter(t, second.Usage(), live) + collect(t, mustSubmit(t, second, "finish")) + after := stampedMeta(t, second, dir) + if math.Abs(after.SpentUSD-.27) > 1e-12 || after.SpentUSD < before.SpentUSD || after.Tokens != 57 { + t.Fatalf("next stamp = %+v, want .27 and 57 tokens, above the previous %+v", after, before) + } + if used := second.Usage(); used.Calls != 3 || used.Turns != 3 { + t.Fatalf("next meter = %+v, want three calls and three conversation steps", used) + } +} + +// Contract A3, A5: on-clock errands retain only their own seals, late receipts +// retain their arm lines, and abandonment changes neither duration nor ledger. +func TestAnAbandonedTurnIsJournaledOnceWithItsErrands(t *testing.T) { + agent, entered, _, path, ledger := abandonSpendAgent(t, &scriptedCompleter{steps: []step{ + func(context.Context, []ai.Message) (*ai.Response, error) { + response := priced(toolResponse("paid", "wedge", `{}`), .25) + response.Usage.CacheReadInputTokens = 3 + response.Usage.CacheCreationInputTokens = 2 + return response, nil + }, + }}) + events := mustSubmit(t, agent, "park") + waitMetaPatch(t, entered) + errands := priced(textResponse("checked"), .125) + errands.Usage.CacheReadInputTokens = 4 + errands.Usage.CacheCreationInputTokens = 1 + agent.addAuxiliaryUsageAs(errands, "test/helper", 1, "guardian") + agent.reconciled(provider.Reconciled{ + Billed: provider.Billed{Model: "test/model", PromptTokens: 80, CompletionTokens: 20, CachedTokens: 30, Cost: .125}, + Ref: "late-abandoned", Reason: "torn", Found: true, + }) + live := agent.Usage() + assertAbandonSpendMeter(t, live, Usage{CostUSD: .5, Input: 110, Output: 32, CacheRead: 37, CacheWrite: 3, Calls: 3, Turns: 1}) + FlushUsage() + before, err := ReadUsage(ledger, time.Time{}) + if err != nil || len(before) != 3 { + t.Fatalf("ledger before abandon = %+v, %v; want three paid calls", before, err) + } + agent.Interrupt() + spend, ok := agent.Abandon(AbandonStopTimeout) + if !ok || spend.CostUSD != .375 || spend.Calls != 2 { + t.Fatalf("abandon = %+v, %v; want only the step and on-clock errand", spend, ok) + } + collect(t, events) + waitAbandonSpendExit(t, path, 1) + if _, again := agent.Abandon(AbandonStopTimeout); again { + t.Fatal("a second abandon let go of the same turn") + } + if lines := abandonedLines(t, path); len(lines) != 1 || lines[0] != (journalAbandoned{ + Reason: string(AbandonStopTimeout), Input: 30, Output: 12, CacheRead: 7, CacheWrite: 3, CostUSD: .375, Calls: 2, + }) { + t.Fatalf("abandoned evidence = %+v, want the original mixed tally without the late receipt", lines) + } + seals := turnUsageLines(t, path) + if len(seals) != 1 || seals[0].CostUSD != .25 || seals[0].Calls != 1 || seals[0].DurationMS != 0 { + t.Fatalf("step seals = %+v, want exactly one .25 step with no duration", seals) + } + if lines := journalUsageLines(t, path); len(lines) != 2 || !lines[0].Aux || lines[0].CostUSD != .125 || lines[0].Role != "guardian" { + t.Fatalf("usage rows = %+v, want one auxiliary errand and one step seal", lines) + } + var arms int + for _, line := range readLines(t, path) { + var entry sessionEntry + if json.Unmarshal([]byte(line), &entry) == nil && entry.Call != nil && entry.Call.Arm != "" { + arms++ + if entry.Call.Arm != "reconciled" || entry.Call.CostUSD != .125 { + t.Fatalf("late receipt changed: %+v", entry.Call) + } + } + } + if arms != 1 { + t.Fatalf("late arms = %d, want one", arms) + } + replayed, err := replaySessionFile(path) + if err != nil { + t.Fatalf("replay: %v", err) + } + assertAbandonSpendMeter(t, replayed.usage, live) + if got := agent.Usage(); got != live { + t.Fatalf("abandon changed the live meter or duration: before %+v, after %+v", live, got) + } + FlushUsage() + after, err := ReadUsage(ledger, time.Time{}) + if err != nil || !reflect.DeepEqual(after, before) { + t.Fatalf("abandon changed the machine ledger: before %+v, after %+v, %v", before, after, err) + } +} + +// Contract A4, A3: digest metadata and replay agree on abandoned step spend, +// including the ordinary per-model split when two models answered the turn. +func TestAnAbandonedTurnsDigestAgreesWithReplay(t *testing.T) { + for _, split := range []bool{false, true} { + name := "one model" + if split { + name = "two models" + } + t.Run(name, func(t *testing.T) { + steps := []step{} + want := Usage{CostUSD: .25, Input: 20, Output: 7, Calls: 1, Turns: 1} + seals := 1 + if split { + steps = append(steps, func(context.Context, []ai.Message) (*ai.Response, error) { + response := priced(toolResponse("earlier", "ls", `{"path":"."}`), .125) + response.Model = "test/earlier" + return response, nil + }) + want = Usage{CostUSD: .375, Input: 40, Output: 14, Calls: 2, Turns: 2} + seals = 2 + } + steps = append(steps, func(context.Context, []ai.Message) (*ai.Response, error) { + response := priced(toolResponse("paid", "wedge", `{}`), .25) + response.Model = "test/model" + return response, nil + }) + agent, entered, _, path, _ := abandonSpendAgent(t, &scriptedCompleter{steps: steps}) + events := mustSubmit(t, agent, "park") + waitMetaPatch(t, entered) + agent.Interrupt() + if _, ok := agent.Abandon(AbandonStopTimeout); !ok { + t.Fatal("the parked turn was not abandoned") + } + collect(t, events) + waitAbandonSpendExit(t, path, seals) + live := agent.Usage() + if err := agent.Close(); err != nil { + t.Fatalf("close before reopen: %v", err) + } + file, _, err := openSessionFile(path, agent.config.Workspace, "test/model", "") + if err != nil { + t.Fatalf("reopen: %v", err) + } + t.Cleanup(func() { _ = file.Close() }) + assertAbandonSpendMeter(t, file.RestoredUsage(), want) + assertAbandonSpendMeter(t, file.RestoredUsage(), live) + digest, err := ReadDigest(path) + if err != nil { + t.Fatalf("digest: %v", err) + } + if usd, tokens := digest.Spend(); usd != want.CostUSD || tokens != want.Input+want.Output { + t.Fatalf("digest spend = %v / %d tokens, want %+v", usd, tokens, want) + } + if meta := digest.Derive(Meta{}); meta.SpentUSD != want.CostUSD || meta.Tokens != want.Input+want.Output { + t.Fatalf("derived metadata = %+v, want %+v", meta, want) + } + lines := turnUsageLines(t, path) + if len(lines) != seals || len(digest.Usage) != seals { + t.Fatalf("turn seals = %+v, digest usage = %+v, want %d", lines, digest.Usage, seals) + } + for _, line := range lines { + cost := .25 + if line.Model == "test/earlier" && split { + cost = .125 + } else if line.Model != "test/model" { + t.Fatalf("unexpected seal model: %+v", line) + } + if line.CostUSD != cost || line.Input != 20 || line.Output != 7 || line.Calls != 1 || line.DurationMS != 0 { + t.Fatalf("per-model seal = %+v, want its own paid call with no duration", line) + } + } + }) + } +} + +// Contract A5: an abandoned request with no usage writes only its evidence, +// keeping the live duration and machine ledger unchanged. +func TestAnAbandonedTurnThatSpentNothingWritesNoSeal(t *testing.T) { + agent, entered, _, path, ledger := abandonSpendAgent(t, &scriptedCompleter{steps: []step{ + func(context.Context, []ai.Message) (*ai.Response, error) { + response := toolResponse("unpaid", "wedge", `{}`) + response.Usage = nil + return response, nil + }, + }}) + events := mustSubmit(t, agent, "park without usage") + waitMetaPatch(t, entered) + before := agent.Usage() + if before != (Usage{}) { + t.Fatalf("unpaid request banked %+v", before) + } + agent.Interrupt() + if spend, ok := agent.Abandon(AbandonStopTimeout); !ok || spend != (Usage{}) { + t.Fatalf("abandon = %+v, %v; want no spend", spend, ok) + } + collect(t, events) + waitAbandonSpendExit(t, path, 0) + if _, again := agent.Abandon(AbandonStopTimeout); again { + t.Fatal("the empty turn was abandoned twice") + } + if lines := abandonedLines(t, path); len(lines) != 1 || lines[0] != (journalAbandoned{Reason: string(AbandonStopTimeout)}) { + t.Fatalf("empty abandoned evidence = %+v, want exactly one reason-only line", lines) + } + if seals := journalUsageLines(t, path); len(seals) != 0 { + t.Fatalf("an empty abandoned turn wrote usage seals: %+v", seals) + } + if after := agent.Usage(); after != before { + t.Fatalf("empty abandon changed meter or duration: before %+v, after %+v", before, after) + } + if usd, rows := ledgerUSD(t, ledger); usd != 0 || rows != 0 { + t.Fatalf("empty abandon wrote machine ledger spend: %v in %d rows", usd, rows) + } +} + +// Contract A6: historical abandoned evidence never becomes a charge, even +// when its conversation steps have no exit seal to recover. +func TestAnOldAbandonedLineIsStillNotBilled(t *testing.T) { + path := filepath.Join(t.TempDir(), "transcript.jsonl") + write(t, path, + `{"type":"session","version":1,"id":"abc","cwd":"/w","model":"m","timestamp":"t"}`, + `{"type":"usage","usage":{"model":"m","input":10,"output":5,"cacheRead":3,"cacheWrite":2,"costUsd":0.25,"calls":1,"durationMs":250}}`, + `{"type":"call","call":{"model":"m","input":20,"output":7,"costUsd":0.5}}`, + `{"type":"abandoned","abandoned":{"reason":"stop timeout","input":20,"output":7,"costUsd":0.5,"calls":1}}`, + ) + file, _, err := openSessionFile(path, "/w", "m", "") + if err != nil { + t.Fatalf("open historical journal: %v", err) + } + t.Cleanup(func() { _ = file.Close() }) + want := Usage{Input: 10, Output: 5, CacheRead: 3, CacheWrite: 2, CostUSD: .25, Calls: 1, Turns: 1, Duration: 250 * time.Millisecond} + if got := file.RestoredUsage(); got != want { + t.Fatalf("historical replay = %+v, want only the existing seal %+v", got, want) + } + digest, err := ReadDigest(path) + if err != nil { + t.Fatalf("historical digest: %v", err) + } + if usd, tokens := digest.Spend(); usd != .25 || tokens != 15 { + t.Fatalf("historical digest spend = %v / %d tokens, want .25 / 15", usd, tokens) + } +} + +// Contract A3: a displaced turn returning through its ordinary cancellation +// ending writes that seal alone, without an additional abandoned exit seal. +func TestAnAbandonedTurnEndingOrdinarilySealsOnlyOnce(t *testing.T) { + held, release := make(chan struct{}), make(chan struct{}) + var once sync.Once + letReturn := func() { once.Do(func() { close(release) }) } + agent, _, _, path, _ := abandonSpendAgent(t, &scriptedCompleter{steps: []step{ + func(context.Context, []ai.Message) (*ai.Response, error) { + return priced(toolResponse("paid", "ls", `{"path":"."}`), .25), nil + }, + func(ctx context.Context, _ []ai.Message) (*ai.Response, error) { + close(held) + <-release + return nil, ctx.Err() + }, + }}) + t.Cleanup(letReturn) + events := mustSubmit(t, agent, "read and wait") + waitMetaPatch(t, held) + live := agent.Usage() + agent.Interrupt() + if _, ok := agent.Abandon(AbandonStopTimeout); !ok { + t.Fatal("the held request was not abandoned") + } + collect(t, events) + letReturn() + waitAbandonSpendExit(t, path, 1) + if seals := turnUsageLines(t, path); len(seals) != 1 || seals[0].CostUSD != .25 || seals[0].Calls != 1 { + t.Fatalf("ordinary ending seals = %+v, want just one .25 seal", seals) + } + if lines := abandonedLines(t, path); len(lines) != 1 { + t.Fatalf("ordinary ending abandoned evidence = %+v, want one", lines) + } + replayed, err := replaySessionFile(path) + if err != nil { + t.Fatalf("ordinary ending replay: %v", err) + } + assertAbandonSpendMeter(t, replayed.usage, live) +} + +// Contract A3, exit audit: the settle bound and final-report returns must seal +// their paid steps once, without duration or ordinary turn-ending acts. +func TestOtherTurnEndingsSealTheirStepSpend(t *testing.T) { + for _, ending := range []string{"settle bound", "final report"} { + t.Run(ending, func(t *testing.T) { + ctx := context.Background() + tool, args := "ls", `{"path":"."}` + var configure func(*Config) + if ending == "settle bound" { + ctx = withSettleWake(ctx, settleWake{ceiling: 1}) + } else { + tool, args = "stage_result", `{"done":true,"output":"finished"}` + configure = func(c *Config) { + c.Stage = &fakeStageDoor{} + c.Unattended = true + } + } + agent, _, _, path, _ := abandonSpendAgent(t, &scriptedCompleter{steps: []step{ + func(context.Context, []ai.Message) (*ai.Response, error) { + return priced(toolResponse("paid", tool, args), .25), nil + }, + }}, configure) + events, err := agent.Submit(ctx, "finish at the boundary") + if err != nil { + t.Fatalf("submit: %v", err) + } + collect(t, events) + seals := turnUsageLines(t, path) + if len(seals) != 1 || seals[0].CostUSD != .25 || seals[0].Calls != 1 || seals[0].DurationMS != 0 { + t.Fatalf("%s seals = %+v, want one .25 seal with no duration", ending, seals) + } + replayed, err := replaySessionFile(path) + if err != nil { + t.Fatalf("%s replay: %v", ending, err) + } + assertAbandonSpendMeter(t, replayed.usage, Usage{CostUSD: .25, Input: 20, Output: 7, Calls: 1, Turns: 1}) + live := agent.Usage() + assertAbandonSpendMeter(t, replayed.usage, live) + if live.Duration != 0 { + t.Fatalf("%s added live duration: %v", ending, live.Duration) + } + }) + } +} + +// abandonSpendAgent uses the existing releasable wedge and constructor with a +// fixed session folder and private ledger, so resume observes the same record. +func abandonSpendAgent(t *testing.T, completer Completer, configure ...func(*Config)) (*Agent, <-chan struct{}, string, string, string) { + t.Helper() + if scripted, ok := completer.(*scriptedCompleter); ok { + scripted.aside = func(messages []ai.Message) (*ai.Response, bool) { + if isCaptionCall(messages) || isTitleCall(messages) { + response := textResponse("") + response.Usage = nil + return response, true + } + return nil, false + } + } + dir := t.TempDir() + path := filepath.Join(dir, "transcript.jsonl") + ledger := filepath.Join(t.TempDir(), UsageLedgerName) + t.Cleanup(func() { StopUsageWriter(ledger) }) + agent, _ := newTestAgent(t, completer, func(c *Config) { + c.Place = Place{Dir: dir, Workspace: c.Workspace} + c.SessionFile = path + c.usageLedger = ledger + for _, mutate := range configure { + if mutate != nil { + mutate(c) + } + } + }) + entered, release := make(chan struct{}), make(chan struct{}) + t.Cleanup(func() { close(release) }) + agent.tools = append(agent.tools, releasableWedgeTool(entered, release)) + definitions, err := toolDefinitions(agent.tools) + if err != nil { + t.Fatalf("tool definitions: %v", err) + } + agent.definitions = definitions + return agent, entered, dir, path, ledger +} + +// waitAbandonSpendExit waits for the loop's deferred pace row, which follows +// its exit seal even when the abandoned turn's event stream closed earlier. +func waitAbandonSpendExit(t *testing.T, path string, seals int) { + t.Helper() + deadline := time.Now().Add(10 * time.Second) + for { + var ended bool + var written []journalUsage + for _, line := range readLines(t, path) { + var entry sessionEntry + if json.Unmarshal([]byte(line), &entry) != nil { + continue + } + if entry.Type == "pace" { + ended = true + } + if entry.Type == "usage" && entry.Usage != nil && !entry.Usage.Aux { + written = append(written, *entry.Usage) + } + } + if ended { + if len(written) != seals { + t.Fatalf("turn exit wrote %+v, want %d step seals", written, seals) + } + return + } + if time.Now().After(deadline) { + t.Fatal("the displaced turn did not journal its exit within 10 seconds") + } + time.Sleep(5 * time.Millisecond) + } +} + +// assertAbandonSpendMeter compares the accounting that must survive replay; +// duration is rounded to milliseconds by the journal's existing format. +func assertAbandonSpendMeter(t *testing.T, got, want Usage) { + t.Helper() + if math.Abs(got.CostUSD-want.CostUSD) > 1e-12 || got.Input != want.Input || got.Output != want.Output || + got.CacheRead != want.CacheRead || got.CacheWrite != want.CacheWrite || got.Calls != want.Calls || got.Turns != want.Turns { + t.Fatalf("replayed meter = %+v, want live accounting %+v", got, want) + } +} diff --git a/internal/session/loop.go b/internal/session/loop.go index c8d6676de1..fc8fb48e99 100644 --- a/internal/session/loop.go +++ b/internal/session/loop.go @@ -1017,6 +1017,10 @@ func (a *Agent) runTurn(ctx context.Context, hub *eventHub, user userMessage) (f // asked here only for the sentence it carries ([Agent.markSettleBound]). if count, tripped := a.settleBoundTripped(ctx, &turn, settleCalls); tripped { a.markSettleBound(count) + // Keep the step spend without duration or ordinary turn-ending acts; + // abandonment can displace this turn while the bound is marked. + turn.Duration = 0 + a.file.appendUsage(turn, model, false, "") return false } @@ -1481,20 +1485,23 @@ func (a *Agent) runTurn(ctx context.Context, hub *eventHub, user userMessage) (f // the fold, the readings that ride beside it — and the law says it is // milliseconds. pace.resultsIn(time.Now()) - // AND A TURN THIS SESSION HAS ALREADY LET GO OF STOPS HERE, WRITING - // NOTHING. [waitBatch] is the one wait in this loop that can return with - // its work still running, and everything below this line writes: the tool - // messages go into a.messages, which by now belongs to whatever turn the - // person started after they stopped this one. A left-behind turn appending + // AND A TURN THIS SESSION HAS ALREADY LET GO OF STOPS HERE, KEEPING ONLY + // ITS STEP SPEND. [waitBatch] can return with its work still running, + // and everything below this line writes: the tool messages go into + // a.messages, which belongs to the turn the person started next. + // A left-behind turn appending // its results there would be another conversation's transcript growing // tool rows nobody asked for. // - // It is read here and nowhere else because this is the only boundary an - // abandoned turn can reach — the batch above is what it was parked in, and - // its own cleanup already knows to do nothing (agent.go's turnSeq). The - // read is non-blocking and costs an ordinary turn one closed-channel test - // per tool round. + // The local tally holds conversation steps only; on-clock errands already + // wrote their own auxiliary usage lines. Bank moved the live meter and + // machine ledger, so this writes only the journal seal, with no duration + // or ordinary turn-ending acts. Each exit keeps its own seal; + // this return cannot reach another ending. Cleanup already knows to leave + // the displaced turn alone (agent.go's turnSeq). if abandoned(ctx) { + turn.Duration = 0 + a.file.appendUsage(turn, model, false, "") return false } if severed != nil { @@ -1513,6 +1520,10 @@ func (a *Agent) runTurn(ctx context.Context, hub *eventHub, user userMessage) (f } if a.stageBatchFinal(calls, results) { + // Keep the step spend without duration or ordinary turn-ending acts; + // abandonment can displace this turn after the batch check above. + turn.Duration = 0 + a.file.appendUsage(turn, model, false, "") return true } // The batch can settle after its result lines and each call's `took` @@ -1838,11 +1849,10 @@ func (a *Agent) keepSteeredPartial(partial *partialBuffer, reasoning *reasoningB // sealTurn stamps the turn's wall duration, folds it into the session total, // and writes the turn down. // -// THIS IS THE ONE PLACE A TURN'S COST REACHES THE JOURNAL, and it is here -// because every turn shape in the package ends through it: the loop above, the -// harness turn, the orchestrated turn and the image turn, on the answering path -// and on both failing ones. A write anywhere else would be a turn shape that -// silently kept no record. +// Ordinary endings write their cost here: the loop above, the harness turn, +// the orchestrated turn and the image turn. The abandoned tool-batch, settle-bound +// and final stage-report exits write only their step seals, without duration or +// ordinary turn-ending acts that could affect a replacement turn after abandonment. // // AND IT IS NOT WHERE THE MACHINE'S LEDGER IS WRITTEN, which is the whole of // issue #269. A seal is the shape of a TURN — its duration, its place in the @@ -1865,7 +1875,7 @@ func (a *Agent) sealTurn(turn Usage, started time.Time, model string) Usage { a.mu.Unlock() a.file.appendUsage(turn, model, false, "") // AND THE SESSION'S RUNNING TOTAL IS STAMPED BESIDE IT, for the reason this - // function is the one place the journal is written: what a conversation has + // function owns ordinary turn endings: what a conversation has // cost is a fact every reader of the machine wants and only the transcript // holds, and a surface listing every session on disk cannot open every // transcript to find it (placemeta.go's [Agent.stampSpend]). A turn that @@ -4560,11 +4570,11 @@ func (a *Agent) bank(call bankedCall) { // rather than the one the seat was built on: a fallback mid-run is a // different model doing the work (competence.go). a.answered.add(call.model, call.used) - // AND THE RUNNING TURN'S OWN SHARE MOVES ON THE SAME CALL, for this door's - // own reason: an abandoned turn never writes the seal that would carry its - // cost, so the only honest figure to journal it with is the one accumulated - // here (abandon.go's [Agent.Abandon]). It is reset when a turn opens and is - // meaningless outside one, which is why it is not exported. + // The running turn's last-known spend includes on-clock errands as well as + // conversation steps. Abandon records it as evidence only; the loop's local + // step tally supplies the seal, and errands write their own auxiliary seals. + // It is reset when a turn opens and is meaningless outside one, which is why + // it is not exported. if !call.late { a.turnSpend.Input += call.used.Input a.turnSpend.Output += call.used.Output diff --git a/internal/session/sessionfile.go b/internal/session/sessionfile.go index d8ec49b980..b56d0dddbc 100644 --- a/internal/session/sessionfile.go +++ b/internal/session/sessionfile.go @@ -446,12 +446,11 @@ type journalRule struct { // journalAbandoned is ONE TURN NOBODY WAITED FOR THE END OF. // -// IT IS THE ONE LINE THAT COULD NOT BE RECONSTRUCTED. Every other record of what -// a turn did is written when the turn ENDS — the seal carries its cost, the -// principal line carries its decision — and the whole definition of an abandoned -// turn is that its ending never came. Before this line a turn let go of at a -// bound was indistinguishable in this file from a turn that simply stopped -// talking, which is exactly the reading that made issue #265's four minutes +// IT IS THE ONE LINE THAT COULD NOT BE RECONSTRUCTED. The seal carries a turn's +// cost, but cannot say that the session let go of it while tools still ran. +// Before this line a turn let go of at a bound was indistinguishable in this +// file from a turn that simply stopped talking, which is exactly the reading +// that made issue #265's four minutes // unaccountable: the surface was freed, the person moved on, and the file said a // turn had been stopped and nothing about the fact that something may still have // been running under it. @@ -459,9 +458,10 @@ type journalRule struct { // Reason is why the turn was let go of ([AbandonReason]); today the only one is // the stop bound. The token and money figures are THE TURN'S LAST KNOWN SPEND — // what [Agent.bank] had moved by the moment the door was opened — and they are -// EVIDENCE AND NEVER SPEND, for [journalCall]'s reason exactly: every one of -// those calls already wrote its own line and moved the machine's ledger, so a -// replay that summed this one too would bill the abandoned turn twice. +// EVIDENCE AND NEVER SPEND: the displaced loop writes its conversation steps +// in a usage seal, while on-clock errands already wrote auxiliary usage lines. +// The calls also moved the machine's ledger when banked. Replaying this tally +// as spend would count those charges twice. // // A turn abandoned before it had spent anything writes the line with no figures // on it, which is the emptiness law: the fact worth recording is that the turn @@ -2281,13 +2281,13 @@ func readJournal(reader io.Reader, path string, rebuild bool) (replayedSession, } } case "abandoned": - // DROPPED ON PURPOSE, for the reason a call line is and one of its - // own. Every dollar on it is already counted — the calls it sums each - // wrote their own line and each moved the machine's ledger when they - // were made (usage_ledger.go) — so folding it in would bill the - // abandoned turn twice. And it is not a message: what the turn had - // said before it was let go of is already in the transcript above it, - // and a resumed session must open on that rather than on a note about + // Evidence only: conversation steps belong to the displaced loop's + // usage seal, and errands to their own auxiliary usage lines. This + // tally mixes both, so billing it would count errands twice even in + // older journals whose abandoned turns have no step seal. It is not + // a message either: what the turn said before it was let go of is + // already in the transcript above it. A resumed session opens on + // that rather than on a note about // how the last one ended. The line is for whoever reads the file. case "mark", "ceiling", "division", "carry", "failure", "flight": // DROPPED ON PURPOSE, for the reason a call line is: these are the From 3800cf420cf1491d4640f29f1840dd0c955ccd30 Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Thu, 12 Nov 2026 11:59:46 -0500 Subject: [PATCH 4/6] changes: the carried spend entries name this pull request (#2612) Co-Authored-By: Claude Mythos 5.2 --- ...doned-turn-step-seal.md => 2612-abandoned-turn-step-seal.md} | 2 +- ...414-replay-late-receipts.md => 2612-replay-late-receipts.md} | 2 +- ...413-turn-tool-live-spend.md => 2612-turn-tool-live-spend.md} | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) rename docs/changes/unreleased/{2440-abandoned-turn-step-seal.md => 2612-abandoned-turn-step-seal.md} (98%) rename docs/changes/unreleased/{2414-replay-late-receipts.md => 2612-replay-late-receipts.md} (98%) rename docs/changes/unreleased/{2413-turn-tool-live-spend.md => 2612-turn-tool-live-spend.md} (98%) diff --git a/docs/changes/unreleased/2440-abandoned-turn-step-seal.md b/docs/changes/unreleased/2612-abandoned-turn-step-seal.md similarity index 98% rename from docs/changes/unreleased/2440-abandoned-turn-step-seal.md rename to docs/changes/unreleased/2612-abandoned-turn-step-seal.md index 4d9f4e92bb..a1fae42dca 100644 --- a/docs/changes/unreleased/2440-abandoned-turn-step-seal.md +++ b/docs/changes/unreleased/2612-abandoned-turn-step-seal.md @@ -1,7 +1,7 @@ --- kind: fixed title: A turn stopped at the second stage keeps its paid steps across a restart -pr: 2440 +pr: 2612 surface: [chat, engine] invalidates: - "Was true: a turn let go of at the stop bound recorded its spend only on its abandoned line, which replay never bills, so a restart could save a lower total. Is true now: the turn's paid conversation steps are written as a usage seal without duration, and replay restores them." diff --git a/docs/changes/unreleased/2414-replay-late-receipts.md b/docs/changes/unreleased/2612-replay-late-receipts.md similarity index 98% rename from docs/changes/unreleased/2414-replay-late-receipts.md rename to docs/changes/unreleased/2612-replay-late-receipts.md index 4cef8d6e70..38443ce43e 100644 --- a/docs/changes/unreleased/2414-replay-late-receipts.md +++ b/docs/changes/unreleased/2612-replay-late-receipts.md @@ -1,7 +1,7 @@ --- kind: fixed title: A resumed conversation keeps its late receipts and never saves a lower spend -pr: 2414 +pr: 2612 surface: [chat, engine, app] invalidates: - Replay restored a conversation's spend from usage seals only, so a restart forgot late receipts and the next turn could save a lower total. diff --git a/docs/changes/unreleased/2413-turn-tool-live-spend.md b/docs/changes/unreleased/2612-turn-tool-live-spend.md similarity index 98% rename from docs/changes/unreleased/2413-turn-tool-live-spend.md rename to docs/changes/unreleased/2612-turn-tool-live-spend.md index 84acadc799..eca06f6876 100644 --- a/docs/changes/unreleased/2413-turn-tool-live-spend.md +++ b/docs/changes/unreleased/2612-turn-tool-live-spend.md @@ -1,7 +1,7 @@ --- kind: added title: Tool step events carry the conversation's live spend so far -pr: 2413 +pr: 2612 surface: [app, docs] invalidates: - A running turn said nothing about its conversation's spend until it ended; turn.tool now carries the live figure as spent_usd. From 6354eb8095bb10a68cc65f3833ede79542f34d6e Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Thu, 12 Nov 2026 18:21:56 -0500 Subject: [PATCH 5/6] manual: a late receipt's call line is the one a resume adds up (#2612) The transcript-format section of sessions-and-rewind.md still said every `call` line is never added up. Since this pull request a resume adds a call line marked with an `arm` once, as the usage and call-line sections of the same page already say. The section now names that exception, and a test keeps every paragraph that says call lines are not added up naming the arm beside it. Co-Authored-By: Claude Opus 5.5 --- internal/manual/chat/sessions-and-rewind.md | 5 ++-- internal/session/spend_replay_test.go | 30 +++++++++++++++++++++ 2 files changed, 33 insertions(+), 2 deletions(-) diff --git a/internal/manual/chat/sessions-and-rewind.md b/internal/manual/chat/sessions-and-rewind.md index a4e2e8e54c..112fc97b7b 100644 --- a/internal/manual/chat/sessions-and-rewind.md +++ b/internal/manual/chat/sessions-and-rewind.md @@ -292,8 +292,9 @@ line types are `session` (the header: version, id, working directory, model, tim `message`, `compaction`, `rewind`, `title`, `usage` and `call`. Message content is written as text. A `usage` line records what one piece of work cost, and on a resume those lines are added back up — which is why the bill survives a restart. A `call` line records one -request's own shape and is **never** added up: the money on it is already counted in the -`usage` line that closed its turn. +request's own shape and is normally **not** added up: the money on it is already counted in +the `usage` line that closed its turn. The exception is a late receipt, a `call` line marked +with an `arm`: its money is in no `usage` line, so a resume adds it once. The session id is 16 random hex characters, minted when the file is created and replayed unchanged on every resume — that is what keeps one conversation one identity across days. diff --git a/internal/session/spend_replay_test.go b/internal/session/spend_replay_test.go index 115c9c3bf0..7e1b62f5a7 100644 --- a/internal/session/spend_replay_test.go +++ b/internal/session/spend_replay_test.go @@ -4,7 +4,9 @@ import ( "context" "fmt" "math" + "os" "path/filepath" + "strings" "testing" "time" @@ -243,3 +245,31 @@ func TestDigestSpendIncludesLateReceiptsWithoutAddingUsageLines(t *testing.T) { }) } } + +// Contract (a), manual: the shipped page must not tell a reader that every call +// line is left out of the bill. A resume adds a line marked with an arm +// (TestLateCallLinesRestoreSpendWithoutAddingTurns), so any paragraph that +// says call lines are never added up has to name that exception beside it. +func TestTheManualNamesTheArmWhereverItSaysCallLinesAreNotAddedUp(t *testing.T) { + page, err := os.ReadFile("../manual/chat/sessions-and-rewind.md") + if err != nil { + t.Fatal(err) + } + var denials int + for _, paragraph := range strings.Split(string(page), "\n\n") { + words := strings.ToLower(strings.Join(strings.Fields(paragraph), " ")) + if !strings.Contains(words, "`call` line") && !strings.Contains(words, "call lines") { + continue + } + if !strings.Contains(words, "nothing adds them up") && !(strings.Contains(words, "never") && strings.Contains(words, "added up")) { + continue + } + denials++ + if !strings.Contains(paragraph, "`arm`") { + t.Errorf("this paragraph says call lines are not added up without naming the late-receipt arm a resume adds:\n%s", paragraph) + } + } + if denials == 0 { + t.Fatal("the page no longer says which call lines a resume leaves out") + } +} From eca765e991f07a5b6726fc61ad7864aa9d3a9788 Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Thu, 12 Nov 2026 18:22:05 -0500 Subject: [PATCH 6/6] session: a step seal saves the total Home reads, as a turn seal does (#2612) The abandoned tool-batch, settle-bound and final stage-report exits wrote their step seal to the journal but never stamped meta.json. Their steps bank as turn calls, which Agent.bank leaves for the seal to stamp, so meta.json kept the figure from before them. After a restart the meter (/cost, the status row, agents.budget.get) restored the journal's total while meta.json (agents.list spent_usd, Home) kept the older one, because stampRestoredSpend fills only a missing figure. A .01 title followed by a .25 step at any of the three exits read $0.26 on the meter and $0.01 in meta.json until the next ordinary turn sealed. The three exits now share sealSteps, which writes the step seal without duration and owes the deferred spend stamp. The stamp reads the live books and writes only the spend fields, so it changes nothing a replacement turn owns. The manual and the change entry say so. Co-Authored-By: Claude Opus 5.5 --- .../2612-abandoned-turn-step-seal.md | 9 +- internal/manual/chat/sessions-and-rewind.md | 8 +- internal/session/abandon_spend_test.go | 82 +++++++++++++++++++ internal/session/loop.go | 36 ++++++-- 4 files changed, 120 insertions(+), 15 deletions(-) diff --git a/docs/changes/unreleased/2612-abandoned-turn-step-seal.md b/docs/changes/unreleased/2612-abandoned-turn-step-seal.md index a1fae42dca..f642414c35 100644 --- a/docs/changes/unreleased/2612-abandoned-turn-step-seal.md +++ b/docs/changes/unreleased/2612-abandoned-turn-step-seal.md @@ -2,12 +2,15 @@ kind: fixed title: A turn stopped at the second stage keeps its paid steps across a restart pr: 2612 -surface: [chat, engine] +surface: [chat, engine, app] invalidates: - "Was true: a turn let go of at the stop bound recorded its spend only on its abandoned line, which replay never bills, so a restart could save a lower total. Is true now: the turn's paid conversation steps are written as a usage seal without duration, and replay restores them." --- The abandoned line stays evidence and is never billed; errands on the turn's clock keep their own usage lines. A settle turn stopping at its bound and a final stage -report now write the same step seal, without duration. Transcripts written -before this change cannot recover an abandoned turn's missing steps. +report now write the same step seal, without duration. Each step seal also saves +the conversation's total to `meta.json`, so after a restart the saved figure Home +and `agents.list` read matches the meter the resumed conversation restores. +Transcripts written before this change cannot recover an abandoned turn's missing +steps. diff --git a/internal/manual/chat/sessions-and-rewind.md b/internal/manual/chat/sessions-and-rewind.md index 112fc97b7b..683442f081 100644 --- a/internal/manual/chat/sessions-and-rewind.md +++ b/internal/manual/chat/sessions-and-rewind.md @@ -318,9 +318,11 @@ There is one line per **completed turn**, and one more for each call made beside naming the conversation, the memory reflex, the check on a reply, a task's fold-up, a planner, a harness run. Those are marked `aux`. A second-stage stop that lets go of a turn while tools still run also saves its paid conversation steps in a `usage` seal, -without duration. Its `abandoned` line records the stop and last-known spend as evidence; -it is never added to the bill. A turn that spent nothing writes no usage line. -Older transcripts without that step seal cannot restore the missing steps. +without duration. It saves the conversation's new total to `meta.json` too, which is +what Home reads, so that figure matches the transcript after a restart. Its `abandoned` +line records the stop and last-known spend as evidence; it is never added to the bill. +A turn that spent nothing writes no usage line. Older transcripts without that step +seal cannot restore the missing steps. One exception to "one line per turn": a turn that was answered by **more than one model** — the model hopped mid-turn — writes **one line per model that answered**, each with that diff --git a/internal/session/abandon_spend_test.go b/internal/session/abandon_spend_test.go index 9fce93a0ff..2260dc6f94 100644 --- a/internal/session/abandon_spend_test.go +++ b/internal/session/abandon_spend_test.go @@ -381,6 +381,88 @@ func TestOtherTurnEndingsSealTheirStepSpend(t *testing.T) { } } +// Contract A2, exit audit: a step seal saves the total Home and agents.list +// read, as an ordinary seal does. A title has already saved .01, so meta.json +// holds a figure a resume will not replace; after a .25 step ends at an +// abandoned batch, a settle bound or a final stage report and the engine +// restarts, the meter (/cost, the status row, agents.budget.get), meta.json +// (agents.list spent_usd, Home) and the journal must all read .26. +func TestAStepSealSavesTheTotalAResumeRestores(t *testing.T) { + for _, ending := range []string{"abandoned", "settle bound", "final report"} { + t.Run(ending, func(t *testing.T) { + ctx := context.Background() + tool, args := "wedge", `{}` + var configure func(*Config) + switch ending { + case "settle bound": + ctx = withSettleWake(ctx, settleWake{ceiling: 1}) + tool, args = "ls", `{"path":"."}` + case "final report": + tool, args = "stage_result", `{"done":true,"output":"finished"}` + configure = func(c *Config) { + c.Stage = &fakeStageDoor{} + c.Unattended = true + } + } + agent, entered, dir, path, ledger := abandonSpendAgent(t, &scriptedCompleter{steps: []step{ + func(context.Context, []ai.Message) (*ai.Response, error) { + return priced(toolResponse("paid", tool, args), .25), nil + }, + }}, configure) + agent.addDetachedUsageAs(priced(textResponse("named"), .01), "test/helper", 1, "title") + if saved := stampedMeta(t, agent, dir); math.Abs(saved.SpentUSD-.01) > 1e-12 { + t.Fatalf("the title saved %+v, want .01 before the paid step", saved) + } + events, err := agent.Submit(ctx, "finish at the boundary") + if err != nil { + t.Fatalf("submit: %v", err) + } + if ending == "abandoned" { + waitMetaPatch(t, entered) + agent.Interrupt() + if _, ok := agent.Abandon(AbandonStopTimeout); !ok { + t.Fatal("the parked turn was not abandoned") + } + } + collect(t, events) + waitAbandonSpendExit(t, path, 1) + live := agent.Usage() + if math.Abs(live.CostUSD-.26) > 1e-12 { + t.Fatalf("live meter = %+v, want .26", live) + } + saved := stampedMeta(t, agent, dir) + if err := agent.Close(); err != nil { + t.Fatalf("close before restart: %v", err) + } + second, err := newAgent(Config{ + Workspace: agent.config.Workspace, Model: "test/model", System: "SYSTEM", + Place: Place{Dir: dir, Workspace: agent.config.Workspace}, SessionFile: path, usageLedger: ledger, + }, &scriptedCompleter{}) + if err != nil { + t.Fatalf("resume: %v", err) + } + t.Cleanup(func() { _ = second.Close() }) + second.SettleWrites() + restored := second.Usage() + listed, err := SavedSpend(dir) + if err != nil { + t.Fatalf("saved spend: %v", err) + } + digest, err := ReadDigest(path) + if err != nil { + t.Fatalf("digest: %v", err) + } + journal, _ := digest.Spend() + if math.Abs(saved.SpentUSD-live.CostUSD) > 1e-12 || saved.Tokens != live.Input+live.Output { + t.Errorf("before restart meta.json = $%v / %d tokens, meter = $%v / %d tokens", saved.SpentUSD, saved.Tokens, live.CostUSD, live.Input+live.Output) + } + if math.Abs(restored.CostUSD-.26) > 1e-12 || math.Abs(listed-restored.CostUSD) > 1e-12 || math.Abs(journal-restored.CostUSD) > 1e-12 { + t.Errorf("after restart meter (/cost) = $%v, meta.json (agents.list) = $%v, journal = $%v; want all .26", restored.CostUSD, listed, journal) + } + }) + } +} + // abandonSpendAgent uses the existing releasable wedge and constructor with a // fixed session folder and private ledger, so resume observes the same record. func abandonSpendAgent(t *testing.T, completer Completer, configure ...func(*Config)) (*Agent, <-chan struct{}, string, string, string) { diff --git a/internal/session/loop.go b/internal/session/loop.go index fc8fb48e99..e58ad9e123 100644 --- a/internal/session/loop.go +++ b/internal/session/loop.go @@ -1019,8 +1019,7 @@ func (a *Agent) runTurn(ctx context.Context, hub *eventHub, user userMessage) (f a.markSettleBound(count) // Keep the step spend without duration or ordinary turn-ending acts; // abandonment can displace this turn while the bound is marked. - turn.Duration = 0 - a.file.appendUsage(turn, model, false, "") + a.sealSteps(turn, model) return false } @@ -1500,8 +1499,7 @@ func (a *Agent) runTurn(ctx context.Context, hub *eventHub, user userMessage) (f // this return cannot reach another ending. Cleanup already knows to leave // the displaced turn alone (agent.go's turnSeq). if abandoned(ctx) { - turn.Duration = 0 - a.file.appendUsage(turn, model, false, "") + a.sealSteps(turn, model) return false } if severed != nil { @@ -1522,8 +1520,7 @@ func (a *Agent) runTurn(ctx context.Context, hub *eventHub, user userMessage) (f if a.stageBatchFinal(calls, results) { // Keep the step spend without duration or ordinary turn-ending acts; // abandonment can displace this turn after the batch check above. - turn.Duration = 0 - a.file.appendUsage(turn, model, false, "") + a.sealSteps(turn, model) return true } // The batch can settle after its result lines and each call's `took` @@ -1851,8 +1848,9 @@ func (a *Agent) keepSteeredPartial(partial *partialBuffer, reasoning *reasoningB // // Ordinary endings write their cost here: the loop above, the harness turn, // the orchestrated turn and the image turn. The abandoned tool-batch, settle-bound -// and final stage-report exits write only their step seals, without duration or -// ordinary turn-ending acts that could affect a replacement turn after abandonment. +// and final stage-report exits write only their step seals ([Agent.sealSteps]), +// without duration or ordinary turn-ending acts that could affect a replacement +// turn after abandonment. // // AND IT IS NOT WHERE THE MACHINE'S LEDGER IS WRITTEN, which is the whole of // issue #269. A seal is the shape of a TURN — its duration, its place in the @@ -1891,6 +1889,25 @@ func (a *Agent) sealTurn(turn Usage, started time.Time, model string) Usage { return turn } +// sealSteps writes the paid conversation steps of a turn that ends without +// [Agent.sealTurn]: an abandoned tool batch, a settle bound, or a final stage +// report. The seal carries no duration, and none of the turn-ending acts run, +// because after abandonment the session already belongs to the next turn. +// +// IT SAVES THE RUNNING TOTAL BESIDE THE SEAL, as an ordinary seal does. The +// steps banked as turn calls, which [Agent.bank] leaves for the seal to stamp. +// Without this stamp meta.json kept the figure from before these steps. A +// restart then restored the larger journal total on the meter while Home and +// agents.list read the older figure ([Agent.stampRestoredSpend] fills only a +// missing figure). The stamp is the deferred, coalesced spend write. +// It reads the live books and writes only the spend fields, so it changes +// nothing a replacement turn owns. +func (a *Agent) sealSteps(turn Usage, model string) { + turn.Duration = 0 + a.file.appendUsage(turn, model, false, "") + a.stampSpend() +} + // tellWrote tells the seal watch that the chat's record changed, when it wants // to know: a write behind the last seal is sealed by the watch once it stops. func (a *Agent) tellWrote() { @@ -4617,7 +4634,8 @@ func (a *Agent) bank(call bankedCall) { // NOWHERE ELSE. A late receipt, an errand's call, memory work or a remote // task's money can arrive after the turn sealed, and Home must not read less // than /cost (dev #2577). A step of the conversation is stamped once, when - // its turn seals, so a turn saves its spend once however many calls it made + // its turn seals or its step seal is written ([Agent.sealSteps]), so a turn + // saves its spend once however many calls it made // (app/v3's TestLateRecoveredCancellationSavesCumulativeSpendOnce). if !call.turn { a.stampSpend()