From aafc28c23ebf85515fa247e0be6d08a1c77f53ee Mon Sep 17 00:00:00 2001 From: saladday <1203511142@qq.com> Date: Tue, 22 Sep 2026 21:32:19 +0800 Subject: [PATCH 1/9] test: preserve observed usage through worker restart --- CONTRIBUTING.md | 10 ++++++++ internal/agentdaemon/proto/inbound.go | 13 ++++++---- .../internal/store/public_execution_test.go | 25 +++++++++++++++++++ 3 files changed, 43 insertions(+), 5 deletions(-) diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 906921a59..673c051f9 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -1649,6 +1649,16 @@ replaced; do not carry obsolete compatibility code forward to satisfy this secti Cancellation receipts carry the stopped engine's continuity snapshot when no Done is emitted. Preserve separately reported usage on failure; do not add the same counters again when Done also includes them. + Usage frames carry cumulative snapshots for the current execution, not deltas. + Adapters publish observed snapshots promptly through the same ordered stream; + waiting for Done unnecessarily loses known measurements if the Runtime stops. + Core replaces complete valid token breakdowns and preserves the last committed + measurement on interruption. Missing measurements remain unknown. Do not infer + token consumption from context occupancy or estimated costs, or parse native + Raw payloads in Core. Session totals cover recorded root Turns; mixed root/child + Turn listings are not a summable accounting ledger. Native measurement coverage + and exact provider/model attribution remain explicit qualification boundaries. + No separate public usage event or historical SSE replay is introduced. - The dispatcher is an internal entry point used by the standalone service worker. Legacy internal `daemon` configuration is not a public Environment type. Self-hosted enrollment uses exact local binding; further pending interactions diff --git a/internal/agentdaemon/proto/inbound.go b/internal/agentdaemon/proto/inbound.go index 20dc08183..1b2770ee0 100644 --- a/internal/agentdaemon/proto/inbound.go +++ b/internal/agentdaemon/proto/inbound.go @@ -52,7 +52,8 @@ const ( // frame arrives. TypeInteractionDecisionAck = "interaction_decision_ack" - // TypeUsage reports incremental token / cost usage. + // TypeUsage reports a cumulative usage snapshot for the current execution. + // Repeated snapshots, including the final Done snapshot, replace rather than add. TypeUsage = "usage" // TypeError signals the prompt failed. Daemon MUST emit a Done @@ -197,10 +198,9 @@ func (p PromptForUserChoicePayload) EffectiveQuestions() []PromptForUserChoiceQu }} } -// Usage mirrors server/internal/store.UsageInput on the wire — field -// names and JSON tags identical so the connector boundary copies with -// a one-liner translator. Redeclared (not imported) because this -// package must stay free of server/internal dependencies. +// TokenUsage is a complete cumulative measurement for the current execution. +// Adapters publish observed snapshots promptly; historical resume totals belong +// to the adapter baseline, not this execution. Unknown fields must not become zero. type TokenUsage struct { InputTokens int64 `json:"input_tokens"` CachedInputTokens int64 `json:"cached_input_tokens"` @@ -209,6 +209,9 @@ type TokenUsage struct { TotalTokens int64 `json:"total_tokens"` } +// Usage retains native evidence independently of the optional public breakdown. +// Tokens is absent when a complete, correctly scoped measurement is unavailable. +// Core does not reconstruct token counts, prices or provider identity from Raw. type Usage struct { Tokens *TokenUsage `json:"tokens,omitempty"` Provider string `json:"provider,omitempty"` diff --git a/services/agents-api/internal/store/public_execution_test.go b/services/agents-api/internal/store/public_execution_test.go index c6b00bf61..14557e3e6 100644 --- a/services/agents-api/internal/store/public_execution_test.go +++ b/services/agents-api/internal/store/public_execution_test.go @@ -117,6 +117,30 @@ func TestWorkerRestartReconcilesClaimedButPreservesQueuedWork(t *testing.T) { if _, err := h.s.TransitionTurn(ctx, h.tenant, h.session.ID, first.TurnID, store.TurnTransition{ExpectedStatus: store.TurnQueued, Status: store.TurnInProgress}); err != nil { t.Fatal(err) } + // A native measurement committed before process loss must survive startup + // reconciliation even when no Done frame can be recovered. + usage := json.RawMessage(`{"tokens":{"input_tokens":10,"cached_input_tokens":4,"output_tokens":3,"reasoning_output_tokens":2,"total_tokens":13}}`) + if err := h.s.AppendTurnEvents(ctx, h.tenant, h.session.ID, first.TurnID, 1, []store.ExecutionEvent{{Kind: proto.TypeUsage, Payload: usage}}); err != nil { + t.Fatal(err) + } + checkMeasurement := func() { + t.Helper() + turn, err := h.s.GetTurn(ctx, h.tenant, h.session.ID, first.TurnID) + if err != nil { + t.Fatal(err) + } + var got struct { + TotalTokens int64 `json:"total_tokens"` + } + if json.Unmarshal(turn.Usage, &got) != nil || got.TotalTokens != 13 { + t.Fatalf("lost observed usage: %s", turn.Usage) + } + session, err := h.s.GetSession(ctx, h.tenant, h.session.ID) + if err != nil || string(session.Usage) != string(turn.Usage) { + t.Fatalf("Session and Turn measurement differ: %+v %v", session, err) + } + } + checkMeasurement() queued := publicSession(t, h, "queued") if _, err := h.s.SubmitMessage(ctx, h.tenant, queued.ID, "first", json.RawMessage(`{"text":"Not sent"}`)); err != nil { t.Fatal(err) @@ -152,4 +176,5 @@ func TestWorkerRestartReconcilesClaimedButPreservesQueuedWork(t *testing.T) { if err := restarted.Run(stopped); err != context.Canceled { t.Fatal(err) } + checkMeasurement() } From a671f008c8a39dd8ddd07785bfa8ba35b1ce8c97 Mon Sep 17 00:00:00 2001 From: saladday <1203511142@qq.com> Date: Tue, 22 Sep 2026 21:35:41 +0800 Subject: [PATCH 2/9] fix: publish Codex usage snapshots during active turns --- .../internal/agent/codex/session_usage.go | 30 ++- .../agent/codex/session_usage_live_test.go | 197 ++++++++++++++++++ .../agent/codex/session_usage_test.go | 8 +- 3 files changed, 223 insertions(+), 12 deletions(-) create mode 100644 apps/parsar-daemon/internal/agent/codex/session_usage_live_test.go diff --git a/apps/parsar-daemon/internal/agent/codex/session_usage.go b/apps/parsar-daemon/internal/agent/codex/session_usage.go index 7f03a9c54..9f9b310ff 100644 --- a/apps/parsar-daemon/internal/agent/codex/session_usage.go +++ b/apps/parsar-daemon/internal/agent/codex/session_usage.go @@ -33,8 +33,16 @@ func (s *Session) onUsageUpdated(raw json.RawMessage) { if s.terminal.Load() { return } + var observed *TurnUsage s.usageMu.Lock() - defer s.usageMu.Unlock() + defer func() { + s.usageMu.Unlock() + // Native notifications are ordered by the RPC reader. Publish before it + // reads completion, without holding the snapshot lock across backpressure. + if observed != nil { + s.emitUsage(*observed) + } + }() if p.TokenUsage != nil && p.TokenUsage.Total != nil && p.TurnID != "" { // app-server replays the previous thread total after resume and // before turn/started. It establishes a baseline, not new usage. @@ -47,19 +55,20 @@ func (s *Session) onUsageUpdated(raw json.RawMessage) { } s.usageTotal = *p.TokenUsage.Total u := subtractUsage(s.usageTotal, s.usageBaseline) - s.latestUsage = &u - return - } - if p.Usage != nil && s.usageTurnID != "" && (p.TurnID == "" || p.TurnID == s.usageTurnID) { + observed = &u + } else if p.Usage != nil && s.usageTurnID != "" && (p.TurnID == "" || p.TurnID == s.usageTurnID) { u := *p.Usage - s.latestUsage = &u + observed = &u + } + if observed != nil { + s.latestUsage = observed } } func subtractUsage(total, baseline TurnUsage) TurnUsage { return TurnUsage{ observed: true, - complete: total.complete && (!baseline.observed || baseline.complete) && + complete: total.completeTokens() && (!baseline.observed || baseline.completeTokens()) && total.InputTokens >= baseline.InputTokens && total.OutputTokens >= baseline.OutputTokens && total.CachedInputTokens >= baseline.CachedInputTokens && total.ReasoningOutputTokens >= baseline.ReasoningOutputTokens && total.TotalTokens >= baseline.TotalTokens, @@ -93,9 +102,14 @@ func (u *TurnUsage) UnmarshalJSON(raw []byte) error { return nil } +func (u TurnUsage) completeTokens() bool { + return u.complete && u.CachedInputTokens <= u.InputTokens && u.ReasoningOutputTokens <= u.OutputTokens && + u.InputTokens <= u.TotalTokens && u.OutputTokens == u.TotalTokens-u.InputTokens +} + func (s *Session) usagePayload(u TurnUsage) proto.Usage { result := proto.Usage{Provider: "openai", Model: s.resolvedModel, InputTokens: int32(u.InputTokens), OutputTokens: int32(u.OutputTokens)} - if u.complete { + if u.completeTokens() { result.Tokens = &proto.TokenUsage{InputTokens: int64(u.InputTokens), OutputTokens: int64(u.OutputTokens), CachedInputTokens: int64(u.CachedInputTokens), ReasoningOutputTokens: int64(u.ReasoningOutputTokens), TotalTokens: int64(u.TotalTokens)} } return result diff --git a/apps/parsar-daemon/internal/agent/codex/session_usage_live_test.go b/apps/parsar-daemon/internal/agent/codex/session_usage_live_test.go new file mode 100644 index 000000000..0672b6433 --- /dev/null +++ b/apps/parsar-daemon/internal/agent/codex/session_usage_live_test.go @@ -0,0 +1,197 @@ +package codex + +import ( + "context" + "encoding/json" + "io" + "runtime" + "testing" + "time" + + "github.com/MiniMax-AI-Dev/parsar/internal/agentdaemon/proto" +) + +const activeUsageNotification = `{"method":"thread/tokenUsage/updated","params":{"threadId":"thread","turnId":"turn","tokenUsage":{"total":{"inputTokens":30,"cachedInputTokens":4,"outputTokens":10,"reasoningOutputTokens":2,"totalTokens":40}}}}` + +func TestUsagePublishedBeforeCompletion(t *testing.T) { + out := make(chan proto.Envelope, 8) + s := &Session{runID: "run", out: out, cancelCtx: t.Context(), cfg: defaultSessionConfig()} + s.setThreadID("thread") + s.onUsageUpdated(json.RawMessage(`{"threadId":"thread","turnId":"old","tokenUsage":{"total":{"inputTokens":10,"cachedInputTokens":1,"outputTokens":3,"reasoningOutputTokens":1,"totalTokens":13}}}`)) + if len(s.out) != 0 { + t.Fatal("resume baseline was published as current usage") + } + s.onTurnStarted(json.RawMessage(`{"threadId":"thread","turn":{"id":"turn"}}`)) + var notification struct { + Params json.RawMessage `json:"params"` + } + if err := json.Unmarshal([]byte(activeUsageNotification), ¬ification); err != nil { + t.Fatal(err) + } + s.onUsageUpdated(notification.Params) + s.onUsageUpdated(notification.Params) + if s.terminal.Load() || len(s.out) != 2 { + t.Fatalf("expected repeated live snapshots before completion, got %d", len(s.out)) + } + var payload proto.UsagePayload + event := <-out + if err := event.DecodePayload(&payload); err != nil { + t.Fatal(err) + } + want := proto.TokenUsage{InputTokens: 20, CachedInputTokens: 3, OutputTokens: 7, ReasoningOutputTokens: 1, TotalTokens: 27} + if event.Type != proto.TypeUsage || event.ID != "run" || payload.Tokens == nil || *payload.Tokens != want { + t.Fatalf("wrong live snapshot: %+v %+v", event, payload) + } + repeated := <-out + if repeated.Type != proto.TypeUsage || string(repeated.Payload) != string(event.Payload) { + t.Fatal("repeated cumulative snapshot changed the reported measurement") + } + // No terminal event is needed to observe or retain the current measurement. + if got := s.CancellationOutcome().Usage.Tokens; got == nil || *got != want { + t.Fatalf("live observation missing from cancellation outcome: %+v", got) + } +} + +func TestUsageBackpressureKeepsSnapshotReadableAndCompletionOrdered(t *testing.T) { + s, _, server := cancellationTestSession(t) + out := make(chan proto.Envelope, 1) + s.out = out + s.out <- proto.Envelope{Type: "occupied"} + s.registerHandlers() + s.setThreadID("thread") + s.onTurnStarted(json.RawMessage(`{"threadId":"thread","turn":{"id":"turn"}}`)) + written := make(chan error, 1) + go func() { + _, err := io.WriteString(server.ToClient, activeUsageNotification+"\n"+`{"method":"turn/completed","params":{"threadId":"thread","turn":{"id":"turn","status":"completed"}}}`+"\n") + written <- err + }() + waitForUsageBackpressure(t, s) + assertReadableUsageSnapshot(t, s) + if s.terminal.Load() { + t.Fatal("completion overtook the blocked live usage") + } + <-out + for _, kind := range []string{proto.TypeUsage, proto.TypeUsage, proto.TypeDone} { + select { + case event := <-out: + if event.Type != kind { + t.Fatalf("event %q, want %q", event.Type, kind) + } + if kind == proto.TypeDone { + var done proto.DonePayload + if err := event.DecodePayload(&done); err != nil || done.Usage.Tokens == nil || done.Usage.Tokens.TotalTokens != 40 { + t.Fatalf("completion lost its snapshot: %+v %v", done, err) + } + } + case <-time.After(time.Second): + t.Fatal("ordered native output did not resume") + } + } + if err := <-written; err != nil { + t.Fatal(err) + } +} + +func TestCancellationReleasesUsageBackpressure(t *testing.T) { + s, _, server := cancellationTestSession(t) + s.out = make(chan proto.Envelope, 1) + s.out <- proto.Envelope{Type: "occupied"} + s.registerHandlers() + s.setThreadID("thread") + s.onTurnStarted(json.RawMessage(`{"threadId":"thread","turn":{"id":"turn"}}`)) + requests := collectCancellationRequests(t, server, "response timeout") + written := make(chan error, 1) + go func() { + _, err := io.WriteString(server.ToClient, activeUsageNotification+"\n") + written <- err + }() + waitForUsageBackpressure(t, s) + assertReadableUsageSnapshot(t, s) + ctx, cancel := context.WithTimeout(t.Context(), 4*time.Second) + defer cancel() + if err := s.Cancel(ctx); err != nil { + t.Fatal("cancellation could not release usage backpressure", err) + } + if s.cancelCtx.Err() == nil || len(<-requests) != 1 { + t.Fatal("native cancellation did not settle") + } + if err := <-written; err != nil { + t.Fatal(err) + } + assertReadableUsageSnapshot(t, s) +} + +func waitForUsageBackpressure(t *testing.T, s *Session) { + t.Helper() + deadline := time.Now().Add(time.Second) + for s.outMu.TryLock() { + s.outMu.Unlock() + if time.Now().After(deadline) { + t.Fatal("native usage did not reach the blocked output") + } + runtime.Gosched() + } +} + +func assertReadableUsageSnapshot(t *testing.T, s *Session) { + t.Helper() + read := make(chan proto.DonePayload, 1) + go func() { read <- s.CancellationOutcome() }() + select { + case got := <-read: + if got.Usage.Tokens == nil || got.Usage.Tokens.TotalTokens != 40 { + t.Fatalf("missing observed snapshot: %+v", got.Usage) + } + case <-time.After(time.Second): + t.Fatal("output backpressure held the usage snapshot lock") + } +} + +func TestLiveUsageRequiresConsistentCompleteBreakdown(t *testing.T) { + for name, counters := range map[string]string{ + "missing": `{"inputTokens":5,"outputTokens":2}`, + "null": `{"inputTokens":5,"cachedInputTokens":0,"outputTokens":2,"reasoningOutputTokens":null,"totalTokens":7}`, + "negative": `{"inputTokens":5,"cachedInputTokens":-1,"outputTokens":2,"reasoningOutputTokens":0,"totalTokens":7}`, + "total": `{"inputTokens":5,"cachedInputTokens":0,"outputTokens":2,"reasoningOutputTokens":0,"totalTokens":8}`, + "cache": `{"inputTokens":5,"cachedInputTokens":6,"outputTokens":2,"reasoningOutputTokens":0,"totalTokens":7}`, + "reasoning": `{"inputTokens":5,"cachedInputTokens":0,"outputTokens":2,"reasoningOutputTokens":3,"totalTokens":7}`, + "large sum": `{"inputTokens":9223372036854775807,"cachedInputTokens":0,"outputTokens":1,"reasoningOutputTokens":0,"totalTokens":9223372036854775807}`, + } { + t.Run(name, func(t *testing.T) { + out := make(chan proto.Envelope, 1) + s := &Session{out: out, cancelCtx: t.Context(), cfg: defaultSessionConfig()} + s.setThreadID("thread") + s.onTurnStarted(json.RawMessage(`{"threadId":"thread","turn":{"id":"turn"}}`)) + s.onUsageUpdated(json.RawMessage(`{"threadId":"thread","turnId":"turn","tokenUsage":{"total":` + counters + `}}`)) + if len(s.out) != 1 { + t.Fatal("native snapshot was not published") + } + var usage proto.UsagePayload + event := <-out + if err := event.DecodePayload(&usage); err != nil || usage.Tokens != nil { + t.Fatalf("incomplete native snapshot became public token usage: %+v %v", usage, err) + } + }) + } +} + +func TestLiveUsageRejectsInconsistentResumeDelta(t *testing.T) { + out := make(chan proto.Envelope, 2) + s := &Session{out: out, cancelCtx: t.Context(), cfg: defaultSessionConfig()} + s.setThreadID("thread") + s.onUsageUpdated(json.RawMessage(`{"threadId":"thread","turnId":"old","tokenUsage":{"total":{"inputTokens":100,"cachedInputTokens":90,"outputTokens":50,"reasoningOutputTokens":10,"totalTokens":150}}}`)) + s.onTurnStarted(json.RawMessage(`{"threadId":"thread","turn":{"id":"turn"}}`)) + // Both thread snapshots are complete, but cache growth exceeds input growth. + s.onUsageUpdated(json.RawMessage(`{"threadId":"thread","turnId":"turn","tokenUsage":{"total":{"inputTokens":110,"cachedInputTokens":105,"outputTokens":55,"reasoningOutputTokens":11,"totalTokens":165}}}`)) + var usage proto.UsagePayload + event := <-out + if err := event.DecodePayload(&usage); err != nil || usage.Tokens != nil { + t.Fatalf("inconsistent per-Turn delta became public token usage: %+v %v", usage, err) + } + s.onTurnCompleted(json.RawMessage(`{"threadId":"thread","turn":{"id":"turn","status":"completed"}}`)) + previous := *s.latestUsage + s.onUsageUpdated(json.RawMessage(`{"threadId":"thread","turnId":"turn","tokenUsage":{"total":{"inputTokens":130,"cachedInputTokens":105,"outputTokens":60,"reasoningOutputTokens":12,"totalTokens":190}}}`)) + if *s.latestUsage != previous { + t.Fatal("late usage changed the completed Turn") + } +} diff --git a/apps/parsar-daemon/internal/agent/codex/session_usage_test.go b/apps/parsar-daemon/internal/agent/codex/session_usage_test.go index 4a01b15c3..d0de87f6b 100644 --- a/apps/parsar-daemon/internal/agent/codex/session_usage_test.go +++ b/apps/parsar-daemon/internal/agent/codex/session_usage_test.go @@ -11,7 +11,7 @@ import ( ) func TestNativeTokenUsageAcrossRuns(t *testing.T) { - out := make(chan proto.Envelope, 4) + out := make(chan proto.Envelope, 8) s := &Session{runID: "run", out: out, cancelCtx: context.Background(), cfg: sessionConfig{logger: slog.New(slog.NewTextHandler(io.Discard, nil))}} s.setThreadID("thread") @@ -62,7 +62,7 @@ func TestNativeTokenUsageAcrossRuns(t *testing.T) { } func TestNativeTokenUsageFreshThreadAndIgnoredPayloads(t *testing.T) { - s := &Session{} + s := &Session{out: make(chan proto.Envelope, 8), cancelCtx: t.Context(), cfg: defaultSessionConfig()} s.setThreadID("thread") s.onTurnStarted(json.RawMessage(`{"threadId":"thread","turn":{"id":"current"}}`)) s.onUsageUpdated(json.RawMessage(`{"threadId":"thread","turnId":"current","tokenUsage":{"total":{"inputTokens":321,"outputTokens":45}}}`)) @@ -81,7 +81,7 @@ func TestNativeTokenUsageFreshThreadAndIgnoredPayloads(t *testing.T) { } func TestLegacyTurnUsagePayload(t *testing.T) { - s := &Session{} + s := &Session{out: make(chan proto.Envelope, 8), cancelCtx: t.Context(), cfg: defaultSessionConfig()} s.setThreadID("thread") s.onTurnStarted(json.RawMessage(`{"threadId":"thread","turn":{"id":"current"}}`)) s.onUsageUpdated(json.RawMessage(`{"threadId":"thread","usage":{"inputTokens":123,"outputTokens":12}}`)) @@ -95,7 +95,7 @@ func TestLegacyTurnUsagePayload(t *testing.T) { } func TestCompleteTokenBreakdownAndCancellation(t *testing.T) { - s := &Session{} + s := &Session{out: make(chan proto.Envelope, 8), cancelCtx: t.Context(), cfg: defaultSessionConfig()} s.setThreadID("thread") s.onUsageUpdated(json.RawMessage(`{"threadId":"thread","turnId":"old","tokenUsage":{"total":{"inputTokens":100,"cachedInputTokens":20,"outputTokens":50,"reasoningOutputTokens":10,"totalTokens":150}}}`)) s.onTurnStarted(json.RawMessage(`{"threadId":"thread","turn":{"id":"new"}}`)) From 7d6b981014384704d1594fd49a5c6c659a8968e6 Mon Sep 17 00:00:00 2001 From: saladday <1203511142@qq.com> Date: Tue, 22 Sep 2026 21:35:40 +0800 Subject: [PATCH 3/9] fix(client): align history reads and live execution projections --- .../features/sessions/items/ItemRenderers.tsx | 2 +- .../sessions/items/item-status.test.tsx | 19 ++ packages/agents-client/src/client.ts | 277 ++--------------- .../agents-client/src/history-projection.ts | 278 ++++++++++++++++++ .../src/history-recovery.test.ts | 253 ++++++++++++++++ .../agents-client/src/response-projection.ts | 37 +++ packages/agents-client/src/types.ts | 26 +- .../agents-client/src/usage-projection.ts | 32 ++ 8 files changed, 669 insertions(+), 255 deletions(-) create mode 100644 apps/web/src/features/sessions/items/item-status.test.tsx create mode 100644 packages/agents-client/src/history-projection.ts create mode 100644 packages/agents-client/src/history-recovery.test.ts create mode 100644 packages/agents-client/src/response-projection.ts create mode 100644 packages/agents-client/src/usage-projection.ts diff --git a/apps/web/src/features/sessions/items/ItemRenderers.tsx b/apps/web/src/features/sessions/items/ItemRenderers.tsx index 775d77d6c..f5a4cb2c9 100644 --- a/apps/web/src/features/sessions/items/ItemRenderers.tsx +++ b/apps/web/src/features/sessions/items/ItemRenderers.tsx @@ -104,7 +104,7 @@ function WorkStep({ item }: { item: SessionItem }) { : undefined; const patch = supported && item.type === "function_call" && item.name === "apply_patch" ? parseParsarApplyPatch(item.arguments) : null; const expandable = args !== undefined && args !== null || result !== undefined && result !== null; - const row = <>