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/2612-abandoned-turn-step-seal.md b/docs/changes/unreleased/2612-abandoned-turn-step-seal.md new file mode 100644 index 0000000000..f642414c35 --- /dev/null +++ b/docs/changes/unreleased/2612-abandoned-turn-step-seal.md @@ -0,0 +1,16 @@ +--- +kind: fixed +title: A turn stopped at the second stage keeps its paid steps across a restart +pr: 2612 +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. 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/docs/changes/unreleased/2612-replay-late-receipts.md b/docs/changes/unreleased/2612-replay-late-receipts.md new file mode 100644 index 0000000000..38443ce43e --- /dev/null +++ b/docs/changes/unreleased/2612-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: 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. + - 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/docs/changes/unreleased/2612-turn-tool-live-spend.md b/docs/changes/unreleased/2612-turn-tool-live-spend.md new file mode 100644 index 0000000000..eca06f6876 --- /dev/null +++ b/docs/changes/unreleased/2612-turn-tool-live-spend.md @@ -0,0 +1,12 @@ +--- +kind: added +title: Tool step events carry the conversation's live spend so far +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. +--- + +`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. diff --git a/internal/manual/chat/sessions-and-rewind.md b/internal/manual/chat/sessions-and-rewind.md index 41e279334c..683442f081 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. @@ -310,20 +311,26 @@ 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 -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. 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 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 +361,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/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..2260dc6f94 --- /dev/null +++ b/internal/session/abandon_spend_test.go @@ -0,0 +1,546 @@ +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) + } + }) + } +} + +// 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) { + 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/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..e58ad9e123 100644 --- a/internal/session/loop.go +++ b/internal/session/loop.go @@ -1017,6 +1017,9 @@ 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. + a.sealSteps(turn, model) return false } @@ -1481,20 +1484,22 @@ 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) { + a.sealSteps(turn, model) return false } if severed != nil { @@ -1513,6 +1518,9 @@ 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. + a.sealSteps(turn, model) return true } // The batch can settle after its result lines and each call's `took` @@ -1838,11 +1846,11 @@ 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 ([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 @@ -1865,7 +1873,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 @@ -1881,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() { @@ -4560,11 +4587,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 @@ -4607,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() @@ -4654,8 +4682,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..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 @@ -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 @@ -2266,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 @@ -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..7e1b62f5a7 --- /dev/null +++ b/internal/session/spend_replay_test.go @@ -0,0 +1,275 @@ +package session + +import ( + "context" + "fmt" + "math" + "os" + "path/filepath" + "strings" + "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) + } + }) + } +} + +// 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") + } +}