Align Item, event, reasoning and usage serialization

CONTRIBUTING.md

@@ -332,8 +332,9 @@ common field.
 Runtime history uses the existing Core PostgreSQL database: one sanitized row per
 periodic observation, seven-day retention and bounded reads. It is best-effort
 operational evidence, not execution or Usage authority. The execution owner samples
-by default every 30 seconds. Existing canonical Session Usage supplies token
-snapshots; never aggregate provider counters as model tokens. Preserve missing data
+by default every 30 seconds. Core's internal measured Session usage (every
+recorded root Turn snapshot, active Turns included) supplies token snapshots, not
+the public Session usage rule; never aggregate provider counters as model tokens. Preserve missing data
 and reset CPU derivation across compute incarnations or counter regressions.
 The bounded asynchronous database writer and optional OTLP exporter have independent
 queues; external telemetry outages must not stall local history or execution.
@@ -2177,8 +2178,11 @@ replaced; do not carry obsolete compatibility code forward to satisfy this secti
   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; Subagent Turn
-  listings are not a summable accounting ledger. Native measurement coverage
+  Raw payloads in Core. Public Session totals cover recorded root Turns and are
+  null while any root Turn has not ended or once one ends with unknown usage;
+  Runtime telemetry uses the separate measured sum of recorded snapshots. Subagent
+  Turn listings are not a summable accounting ledger. A cumulative native total that has not advanced
+  past the Turn's baseline is not a measurement of that Turn. 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.
@@ -2304,17 +2308,18 @@ replaced; do not carry obsolete compatibility code forward to satisfy this secti
   missing or foreign Sessions keep one 404. Identical saved results remain retryable after termination
   without applying them again. The execution input cursor skips function results;
   their separate native receipts still determine application. Public result events
-  validate variant-specific fields and required values before admission; retain
+  validate variant-specific fields and required values before admission; store
   omitted versus null error/output and ordered text/image parts. Inline function
   tools resolve into the immutable configuration with explicit
   `defer_loading=false`. Validate required strings and parameter objects before
   persistence; reject unsupported deferred discovery. Omitted/null/empty tool
   lists resolve to no tools. The public worker selects or waits for a same-tenant
   device advertising `function_tools` when the Session has functions.
-  Function results are Session input Items: emit `item.added` without an output
+  Function results are Session input Items: emit `item.added` with a null output
   index, and never emit `item.done`, whose upstream union only allows agent output.
-  Project their public output/error from the saved submission, including missing
-  versus null fields; native content normalization must not change public history.
+  Project their public output/error from the saved submission; the wire always
+  carries both, null when not submitted, while stored payloads keep the submitted
+  presence. Native content normalization must not change public history.
 - Codex function application requires a matching live native dynamic-tool completion,
   including root thread/Turn/call identity, function, success and ordered content.
   Writing its JSON-RPC response is not application. The adapter owns pending
@@ -2473,7 +2478,16 @@ replaced; do not carry obsolete compatibility code forward to satisfy this secti
   Completion text replaces accumulated deltas. Item merging must not mutate the
   incoming observation or the previous snapshot: public text delta events read the
   original fragment after merging, while Items retain the accumulated text.
-  Copy the content slice before replacing its text pointer. Keep partial output on termination;
+  Copy the content slice before replacing its text pointer. Assistant text
+  Items follow the official event sequence: `item.added` in progress with empty
+  content, an empty `content_part.added`, deltas, then the done events. A first
+  observation without its own fragment (a non-streamed native final) carries its
+  unchanged text in one delta; this frames the text and never alters it. Wire-only
+  explicit nulls (`phase`, function result `output`/`error`, Item event
+  `output_index`, Agent `reasoning` keys) come from response marshalling. Stored
+  Item payloads keep their original encoding through `Item.MarshalStored`, so
+  replayed child Items still compare equal, and stored configuration keeps
+  omitting unset reasoning keys. Keep partial output on termination;
   do not turn an unfinished call into a successful result. Thinking fragments are
   internal observations, not a claim of upstream reasoning-item support.
 

apps/parsar-daemon/internal/agent/codex/session_usage.go

@@ -55,6 +55,13 @@ func (s *Session) onUsageUpdated(raw json.RawMessage) {
 		}
 		s.usageTotal = *p.TokenUsage.Total
 		u := subtractUsage(s.usageTotal, s.usageBaseline)
+		if u.InputTokens == 0 && u.OutputTokens == 0 && u.TotalTokens == 0 {
+			// The thread total has not advanced past this Turn's baseline. It
+			// reports no usage for this Turn, e.g. when native sends it after an
+			// interrupt before any response reported usage; publishing the
+			// difference would turn missing usage into measured zeros.
+			return
+		}
 		observed = &u
 	} else if p.Usage != nil && s.usageTurnID != "" && (p.TurnID == "" || p.TurnID == s.usageTurnID) {
 		u := *p.Usage

apps/parsar-daemon/internal/agent/codex/session_usage_test.go

@@ -152,3 +152,41 @@ func TestAbnormalTerminationTransmitsKnownUsage(t *testing.T) {
 		t.Fatalf("known usage missing from Done: %+v", done.Usage)
 	}
 }
+
+// Native sends the unchanged thread total when a Turn is interrupted before any
+// response reported usage (EVT-24). That measures nothing for this Turn, so the
+// cancellation outcome keeps usage unknown instead of reporting zeros.
+func TestUnadvancedThreadTotalIsNotTurnUsage(t *testing.T) {
+	previous := `{"inputTokens":13444,"cachedInputTokens":6656,"outputTokens":120,"reasoningOutputTokens":43,"totalTokens":13564}`
+	for name, replay := range map[string]string{
+		"resumed thread": `{"threadId":"thread","turnId":"previous","tokenUsage":{"total":` + previous + `}}`,
+		"fresh thread":   "",
+	} {
+		t.Run(name, func(t *testing.T) {
+			out := make(chan proto.Envelope, 8)
+			s := &Session{runID: "run", out: out, cancelCtx: t.Context(), cfg: defaultSessionConfig()}
+			s.setThreadID("thread")
+			total := `{"inputTokens":0,"cachedInputTokens":0,"outputTokens":0,"reasoningOutputTokens":0,"totalTokens":0}`
+			if replay != "" {
+				s.onUsageUpdated(json.RawMessage(replay))
+				total = previous
+			}
+			s.onTurnStarted(json.RawMessage(`{"threadId":"thread","turn":{"id":"current"}}`))
+			s.onUsageUpdated(json.RawMessage(`{"threadId":"thread","turnId":"current","tokenUsage":{"total":` + total + `}}`))
+			if len(out) != 0 || s.latestUsage != nil {
+				t.Fatalf("unadvanced total published as usage: %+v", s.latestUsage)
+			}
+			outcome := s.CancellationOutcome()
+			if outcome.Usage.Tokens != nil || outcome.Usage.Provider != "" {
+				t.Fatalf("cancellation reported unknown usage: %+v", outcome.Usage)
+			}
+			// A later advanced total is still this Turn's measurement.
+			advanced := `{"inputTokens":13500,"cachedInputTokens":6656,"outputTokens":130,"reasoningOutputTokens":43,"totalTokens":13630}`
+			s.onUsageUpdated(json.RawMessage(`{"threadId":"thread","turnId":"current","tokenUsage":{"total":` + advanced + `}}`))
+			got := s.CancellationOutcome().Usage.Tokens
+			if got == nil || got.TotalTokens == 0 || len(out) != 1 {
+				t.Fatalf("advanced usage lost: %+v", got)
+			}
+		})
+	}
+}

apps/web/src/features/dashboard/RuntimeObservabilityContent.tsx

@@ -31,9 +31,11 @@ import {
   formatDashboardDuration,
   formatDashboardTimestamp,
   formatDashboardTokens,
+  reportedSessionTokens,
   runtimeObservationStatusLabel,
   type RuntimeDashboardRow,
 } from "./dashboard-model";
+import { holdLastReported } from "./held-usage";
 import type { RuntimeDashboardSnapshot } from "./runtime-snapshot";
 import { RuntimeTrendPanel, type RuntimeHistoryLoader } from "./RuntimeTrendPanel";
 
@@ -287,7 +289,21 @@ export function RuntimeObservabilityContent({
   loadRuntimeHistory: RuntimeHistoryLoader;
   onOpenSession: (sessionId: string) => void;
 }) {
-  const model = useMemo(() => buildRuntimeDashboardModel(snapshot.sessions, snapshot.observations), [snapshot]);
+  // Hold each Session's last reported tokens across snapshots, so the summary
+  // total does not drop while a Turn withholds public Session usage.
+  const [heldTokens, setHeldTokens] = useState(() => ({
+    snapshot,
+    totals: holdLastReported(new Map<string, number>(), reportedSessionTokens(snapshot.sessions)),
+  }));
+  let tokenTotals = heldTokens.totals;
+  if (heldTokens.snapshot !== snapshot) {
+    tokenTotals = holdLastReported(heldTokens.totals, reportedSessionTokens(snapshot.sessions));
+    setHeldTokens({ snapshot, totals: tokenTotals });
+  }
+  const model = useMemo(
+    () => buildRuntimeDashboardModel(snapshot.sessions, snapshot.observations, tokenTotals),
+    [snapshot, tokenTotals],
+  );
   const summary = model.summary;
   return (
     <>

apps/web/src/features/dashboard/dashboard-model.test.ts

@@ -11,8 +11,10 @@ import {
   formatDashboardTimestamp,
   formatDashboardBytes,
   formatDashboardDuration,
+  reportedSessionTokens,
   runtimeObservationStatusLabel,
 } from "./dashboard-model";
+import { holdLastReported } from "./held-usage";
 
 function agent(id: string, overrides: Partial<SavedAgent> = {}): SavedAgent {
   return {
@@ -295,6 +297,44 @@ describe("Dashboard loaded-snapshot model", () => {
     expect(formatDashboardDuration(90)).toBe("1m 30s");
   });
 
+  it("holds each Session's last reported tokens in the summary while public usage is null", () => {
+    const running = session("11111111-1111-4111-8111-111111111111", { usage: usage(21) });
+    const other = session("22222222-2222-4222-8222-222222222222", { usage: usage(5) });
+    const observation = (id: string): RuntimeObservation => ({
+      id,
+      object: "agent.runtime_observation",
+      session_id: id,
+      environment_id: null,
+      mode: "none",
+      provider_type: null,
+      instance: { kind: "none", allocation_id: null, device_id: null, connection_generation: null },
+      status: "unsupported",
+      reason: "runtime_mode_not_observable",
+      allocation_created_at: null,
+      resolved_at: 225,
+      observed_at: null,
+      started_at: null,
+      cpu: null,
+      memory: null,
+    });
+    const observations = [observation(running.id), observation(other.id)];
+    let held = holdLastReported(new Map<string, number>(), reportedSessionTokens([running, other]));
+    expect(buildRuntimeDashboardModel([running, other], observations, held).summary)
+      .toMatchObject({ totalTokens: 26, tokenCoverageCount: 2 });
+
+    // A Turn starts: public usage is withheld, the summary keeps the last total.
+    const withheld = { ...running, status: "in_progress", usage: null } as AgentSession;
+    held = holdLastReported(held, reportedSessionTokens([withheld, { ...other, usage: usage(9) }]));
+    const model = buildRuntimeDashboardModel([withheld, { ...other, usage: usage(9) }], observations, held);
+    expect(model.summary).toMatchObject({ totalTokens: 30, tokenCoverageCount: 2 });
+    expect(model.rows.find((row) => row.session.id === withheld.id)?.session.totalTokens).toBeNull();
+    expect(buildRuntimeDashboardModel([withheld, other], observations).summary.totalTokens).toBe(5);
+
+    // A newly reported value replaces the held one; unlisted Sessions are dropped.
+    held = holdLastReported(held, reportedSessionTokens([{ ...running, usage: usage(40) }]));
+    expect([...held]).toEqual([[running.id, 40]]);
+  });
+
   it("does not infer a released allocation lifetime from the current resolution time", () => {
     const stopped = session("11111111-1111-4111-8111-111111111111");
     const observation: RuntimeObservation = {

apps/web/src/features/dashboard/dashboard-model.ts

@@ -275,9 +275,19 @@ function elapsedSeconds(start: number | null, end: number | null): number | null
   return end - start;
 }
 
+/** Each Session's reported total tokens, null when its public usage is null. */
+export function reportedSessionTokens(sessions: readonly AgentSession[]): Map<string, number | null> {
+  return new Map(sessions.map((session) => [session.id, canonicalUsage(session.usage)?.total_tokens ?? null]));
+}
+
+/**
+ * Rows show current public usage. The summary token total adds each Session's
+ * held last reported total (see holdLastReported) while its usage is null.
+ */
 export function buildRuntimeDashboardModel(
   sessions: readonly AgentSession[],
   observations: readonly RuntimeObservation[],
+  heldTokens: ReadonlyMap<string, number> = new Map(),
 ): RuntimeDashboardModel {
   const sessionsById = new Map(sessions.map((session) => [session.id, session]));
   const rows: RuntimeDashboardRow[] = [];
@@ -359,8 +369,9 @@ export function buildRuntimeDashboardModel(
       unsupportedRuntimeCount += 1;
     }
 
-    if (sessionRow.totalTokens !== null) {
-      const next = safeAdd(totalTokens, sessionRow.totalTokens);
+    const sessionTokens = sessionRow.totalTokens ?? heldTokens.get(session.id) ?? null;
+    if (sessionTokens !== null) {
+      const next = safeAdd(totalTokens, sessionTokens);
       if (next !== null) {
         totalTokens = next;
         tokensKnown = true;

apps/web/src/features/dashboard/held-usage.ts

@@ -0,0 +1,17 @@
+/**
+ * Public Session usage is null while a root Turn runs and after a Turn ends
+ * unmeasured. Dashboard token totals keep each listed Session's last reported
+ * value instead, so they neither drop to zero nor break. A newly reported value
+ * replaces the held one, and Sessions no longer listed are dropped.
+ */
+export function holdLastReported<T>(
+  previous: ReadonlyMap<string, T>,
+  current: ReadonlyMap<string, T | null>,
+): Map<string, T> {
+  const held = new Map<string, T>();
+  for (const [sessionId, value] of current) {
+    const kept = value ?? previous.get(sessionId);
+    if (kept !== undefined) held.set(sessionId, kept);
+  }
+  return held;
+}

apps/web/src/features/dashboard/runtime-trends.test.ts

@@ -235,6 +235,31 @@ describe("Runtime live-window trends", () => {
     expect(samples.at(-1)?.targets[0]?.cpuRatio ?? null).toBeNull();
   });
 
+  it("leaves a gap while a Session's public usage is null and spreads the next report", () => {
+    const pending = (at: number) => {
+      const value = snapshot(at);
+      value.sessions[0] = { ...value.sessions[0]!, status: "in_progress", usage: null };
+      return value;
+    };
+    let samples = appendRuntimeTrendSample([], snapshot(60_000, { input: 100, output: 20 }));
+    samples = appendRuntimeTrendSample(samples, pending(120_000));
+    // The held total keeps its report time and is never a zero-rate measurement.
+    expect(samples.at(-1)?.tokenTotals).toEqual([{
+      sessionId: "11111111-1111-4111-8111-111111111111", sampledAt: 60_000, inputTokens: 100, outputTokens: 20, held: true,
+    }]);
+    samples = appendRuntimeTrendSample(samples, pending(180_000));
+    samples = appendRuntimeTrendSample(samples, snapshot(240_000, { input: 400, output: 80 }));
+    samples = appendRuntimeTrendSample(samples, snapshot(300_000, { input: 460, output: 90 }));
+    expect(tokenThroughput(samples)).toEqual([
+      { sampledAt: 60_000, inputPerMinute: null, outputPerMinute: null },
+      { sampledAt: 120_000, inputPerMinute: null, outputPerMinute: null },
+      { sampledAt: 180_000, inputPerMinute: null, outputPerMinute: null },
+      // 300 input and 60 output tokens over the three minutes since the last report.
+      { sampledAt: 240_000, inputPerMinute: 100, outputPerMinute: 20 },
+      { sampledAt: 300_000, inputPerMinute: 60, outputPerMinute: 10 },
+    ]);
+  });
+
   it("keeps Session-set churn as a gap instead of publishing partial throughput", () => {
     const first = snapshot(60_000, { input: 100, output: 20 });
     const second = snapshot(120_000, { input: 220, output: 50 });

apps/web/src/features/dashboard/runtime-trends.ts

@@ -1,5 +1,6 @@
 import type { AgentSession, RuntimeObservation } from "@agents-core-web/agents-client";
 
+import { holdLastReported } from "./held-usage";
 import type { RuntimeDashboardSnapshot } from "./runtime-snapshot";
 
 export const RUNTIME_TREND_WINDOW_MS = 60 * 60 * 1_000;
@@ -44,6 +45,8 @@ export interface RuntimeTrendTokenTotal {
   sampledAt: number;
   inputTokens: number;
   outputTokens: number;
+  /** Kept from the report at sampledAt while public Session usage is null. */
+  held?: true;
 }
 
 export interface TokenThroughputSample {
@@ -104,6 +107,17 @@ function tokenTotals(sessions: readonly AgentSession[], sampledAt: number): Runt
   });
 }
 
+// Keep a listed Session's last reported totals, with their report time, while
+// its public usage is null. tokenRate treats a held total as unknown, so those
+// intervals are gaps; the next reported total is spread over the time since the
+// last report.
+function carryTokenTotals(previous: RuntimeTrendSample, next: RuntimeTrendSample, snapshot: RuntimeDashboardSnapshot): void {
+  const reported = new Map<string, RuntimeTrendTokenTotal | null>(snapshot.sessions.map((session) => [session.id, null]));
+  for (const total of next.tokenTotals) reported.set(total.sessionId, total);
+  const prior = new Map(previous.tokenTotals.map((total) => [total.sessionId, { ...total, held: true as const }]));
+  next.tokenTotals = [...holdLastReported(prior, reported).values()];
+}
+
 export function runtimeTrendSample(snapshot: RuntimeDashboardSnapshot): RuntimeTrendSample {
   const sessions = new Map(snapshot.sessions.map((session) => [session.id, session]));
   const observed = snapshot.observations.flatMap((observation) => {
@@ -244,15 +258,17 @@ function tokenRate(
     const prior = previousTotals.get(current.sessionId);
     return prior ? [[prior, current]] : [];
   });
+  // A total held from an earlier report is not a measurement at this sample.
+  const measured = (current: RuntimeTrendTokenTotal) => !current.held;
   const inputRates = pairs.map(([prior, current]) => {
     const elapsedMinutes = (current.sampledAt - prior.sampledAt) / 60_000;
     const delta = current.inputTokens - prior.inputTokens;
-    return elapsedMinutes > 0 && delta >= 0 ? delta / elapsedMinutes : null;
+    return measured(current) && elapsedMinutes > 0 && delta >= 0 ? delta / elapsedMinutes : null;
   });
   const outputRates = pairs.map(([prior, current]) => {
     const elapsedMinutes = (current.sampledAt - prior.sampledAt) / 60_000;
     const delta = current.outputTokens - prior.outputTokens;
-    return elapsedMinutes > 0 && delta >= 0 ? delta / elapsedMinutes : null;
+    return measured(current) && elapsedMinutes > 0 && delta >= 0 ? delta / elapsedMinutes : null;
   });
   const inputRate = inputRates.length > 0 && inputRates.every((rate) => rate !== null)
     ? inputRates.reduce<number>((total, rate) => total + (rate ?? 0), 0)
@@ -289,6 +305,7 @@ export function appendRuntimeTrendSample(
     .slice(-(maximum - 1));
   const previous = retained.at(-1);
   if (previous) {
+    carryTokenTotals(previous, next, snapshot);
     Object.assign(next, tokenRate(previous, next));
     applyCPURatios(next, cpuRatios(previous, next));
   }