Fix execution history projections and persist live usage

CONTRIBUTING.md

@@ -1651,6 +1651,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

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

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), &notification); 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")
+	}
+}

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"}}`))

apps/web/e2e/agents-lifecycle.spec.ts

@@ -2945,7 +2945,8 @@ test("presents an honest searchable Trace workbench without changing the convers
   await search.fill("");
 
   const patchRow = trace.locator(".trace-ledger-row-tools").filter({ hasText: "apply_patch" }).first();
-  await expect(patchRow.locator('.trace-row-duration [aria-hidden="true"]')).toHaveText("41 ms");
+  await expect(patchRow.locator('.trace-row-duration [aria-hidden="true"]')).toHaveText("—");
+  await expect(patchRow.locator(".trace-row-duration")).toHaveAttribute("title", "Core does not provide per-item timing.");
   await patchRow.click();
   const detail = page.getByRole("complementary", { name: "Trace item details" });
   await expect(detail).toBeVisible();
@@ -3072,6 +3073,8 @@ test("renders Parsar patches as accessible read-only diffs in desktop and narrow
   const failedStep = failedTrace.locator('[data-trace-step="patch_failed"]');
   await failedStep.getByRole("button", { name: /apply_patch/ }).click();
   await expect(failedStep.locator('[data-patch-status="failed"]')).toContainText("Failed");
+  await failedStep.getByText("Raw result").click();
+  await expect(failedStep).toContainText("fixture failure");
   const fallback = page.locator('[data-trace-step="patch_alternate"]');
   await fallback.getByRole("button", { name: /apply_patch/ }).click();
   await expect(fallback).toContainText("malformed alternate shape");

apps/web/e2e/fixture-core.mjs

@@ -20,9 +20,10 @@ function patchItems() {
         { path: "src/delete.ts", kind: { type: "delete", move_path: null }, diff: "@@ -1 +0,0 @@\n-removed" },
       ] },
     },
-    { id: "patch_output", turn_id: "turn_patch_completed", type: "function_call_output", status: "completed", call_id: "call_completed", output: { applied: true }, duration_ms: 41 },
+    { id: "patch_output", turn_id: "turn_patch_completed", type: "function_call_output", status: "completed", call_id: "call_completed", output: '{"applied": true}' },
     { id: "patch_running", turn_id: "turn_patch_running", type: "function_call", status: "in_progress", name: "apply_patch", call_id: "call_running", arguments: { changes: [{ path: "src/running.ts", kind: { type: "update" }, diff: "@@ -1 +1 @@\n-wait\n+working" }] } },
-    { id: "patch_failed", turn_id: "turn_patch_failed", type: "function_call", status: "failed", name: "apply_patch", call_id: "call_failed", arguments: { changes: [{ path: "src/failed.ts", kind: { type: "delete" }, diff: "@@ -1 +0,0 @@\n-failed" }] }, error: { message: "fixture failure" } },
+    { id: "patch_failed", turn_id: "turn_patch_failed", type: "function_call", status: "failed", name: "apply_patch", call_id: "call_failed", arguments: { changes: [{ path: "src/failed.ts", kind: { type: "delete" }, diff: "@@ -1 +0,0 @@\n-failed" }] } },
+    { id: "patch_failed_output", turn_id: "turn_patch_failed", type: "function_call_output", status: "failed", call_id: "call_failed", error: "fixture failure" },
     { id: "patch_alternate", turn_id: "turn_patch_alternate", type: "function_call", status: "in_progress", name: "apply_patch", call_id: "call_alternate", arguments: { patch: "*** Begin Patch\nmalformed alternate shape" } },
   ];
 }

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 = <><Icon className="trace-step-icon" size={14} strokeWidth={1.5} aria-hidden="true" /><span className="trace-step-verb">{verb}</span><span className="trace-step-target" title={target}>{target}</span>{item.status !== "completed" ? <StatusIcon status={itemStatusKind(item.status)} title={item.status.replaceAll("_", " ")} /> : null}{item.duration_ms ? <span className="trace-duration">{formatDuration(item.duration_ms)}</span> : null}{expandable ? <ChevronRight className={`trace-step-chevron ${open ? "open" : ""}`} size={14} strokeWidth={1.5} aria-hidden="true" /> : null}</>;
+  const row = <><Icon className="trace-step-icon" size={14} strokeWidth={1.5} aria-hidden="true" /><span className="trace-step-verb">{verb}</span><span className="trace-step-target" title={target}>{target}</span>{item.status != null && item.status !== "completed" ? <StatusIcon status={itemStatusKind(item.status)} title={item.status.replaceAll("_", " ")} /> : null}{item.duration_ms ? <span className="trace-duration">{formatDuration(item.duration_ms)}</span> : null}{expandable ? <ChevronRight className={`trace-step-chevron ${open ? "open" : ""}`} size={14} strokeWidth={1.5} aria-hidden="true" /> : null}</>;
   return <li className="trace-step" data-trace-step={item.id}>{expandable ? <button className="trace-step-row" type="button" aria-expanded={open} onClick={() => setOpen((value) => !value)}>{row}</button> : <div className="trace-step-row">{row}</div>}{expandable ? <TraceCollapse open={open}>{patch ? <ApplyPatchDiffViewer item={item} patch={patch} result={result} /> : <div className="trace-step-details">{args !== undefined && args !== null ? <div><p>Arguments</p><pre>{pretty(args)}</pre></div> : null}{result !== undefined && result !== null ? <div><p>Result</p><pre>{pretty(result)}</pre></div> : null}</div>}</TraceCollapse> : null}</li>;
 }