fix(core): include provider wait in token throughput timing

packages/client/src/effect/api/api.ts

@@ -688,6 +688,7 @@ export type SessionLogOutput =
             readonly agent: Agent.ID
             readonly model: Model.Ref
             readonly snapshot?: (string & Brand.Brand<"Snapshot.ID">) | undefined
+            readonly started?: DateTime.Utc | undefined
           }
         }
       | {

packages/client/src/promise/generated/types.ts

@@ -748,7 +748,14 @@ export type SessionStepStarted = {
   type: "session.step.started"
   durable: { aggregateID: string; seq: number; version: 1 }
   location?: LocationRef
-  data: { sessionID: string; assistantMessageID: string; agent: string; model: ModelRef; snapshot?: string }
+  data: {
+    sessionID: string
+    assistantMessageID: string
+    agent: string
+    model: ModelRef
+    snapshot?: string
+    started?: number
+  }
 }
 
 export type SessionStepStreamed = {

packages/client/src/solid/data.ts

@@ -806,6 +806,7 @@ export function createData(config: CreateDataInput) {
             existing.finish = undefined
             existing.rawFinish = undefined
             existing.providerState = undefined
+            if (event.data.started !== undefined) existing.time.created = event.data.started
             existing.time.streamed = undefined
             existing.time.completed = undefined
             if (event.data.snapshot) existing.snapshot = { ...existing.snapshot, start: event.data.snapshot }
@@ -824,7 +825,7 @@ export function createData(config: CreateDataInput) {
             metadata: event.metadata,
             content: [],
             snapshot: event.data.snapshot ? { start: event.data.snapshot } : undefined,
-            time: { created: event.created },
+            time: { created: event.data.started ?? event.created },
           })
         })
         return

packages/client/test/solid-data.test.ts

@@ -14,6 +14,59 @@ const session = (viewed: number): SessionInfo => ({
   location: { directory: "/project" },
 })
 
+test("projects request start timestamps and preserves legacy retry timing", async () => {
+  const listeners = new Set<Parameters<CreateDataInput["event"]["listen"]>[0]>()
+  const api = OpenCode.make({
+    baseUrl: "http://opencode.local",
+    fetch: async () => Response.json({ data: [], cursor: {} }),
+  })
+  const setup = createRoot((dispose) => ({
+    data: createData({
+      api: () => api,
+      directory: "/project",
+      event: {
+        on: () => () => {},
+        listen(handler) {
+          listeners.add(handler)
+          return () => listeners.delete(handler)
+        },
+      },
+    }),
+    dispose,
+  }))
+
+  try {
+    await setup.data.session.message.sync("ses_refresh")
+    const cases = [
+      { id: "legacy", started: undefined, expected: 2_690 },
+      { id: "legacy", started: undefined, expected: 2_690 },
+      { id: "current", started: 0, expected: 0 },
+      { id: "current", started: 5_000, expected: 5_000 },
+      { id: "current", started: undefined, expected: 5_000 },
+    ]
+    for (const [index, item] of cases.entries()) {
+      const event: OpenCodeEvent = {
+        id: `evt_started_${index}`,
+        created: 2_690 + index * 3_000,
+        type: "session.step.started",
+        durable: { aggregateID: "ses_refresh", seq: index + 1, version: 1 },
+        data: {
+          sessionID: "ses_refresh",
+          assistantMessageID: item.id,
+          agent: "build",
+          model: { id: "model", providerID: "provider" },
+          ...(item.started === undefined ? {} : { started: item.started }),
+        },
+      }
+      listeners.forEach((listener) => listener({ name: event.type, details: event }))
+      expect(setup.data.session.message.get("ses_refresh", item.id)?.time.created).toBe(item.expected)
+    }
+    expect(setup.data.session.message.list("ses_refresh")).toHaveLength(2)
+  } finally {
+    setup.dispose()
+  }
+})
+
 test("revalidates after an event overtakes an active session read", async () => {
   let release!: () => void
   const gate = new Promise<void>((resolve) => (release = resolve))

packages/core/src/session/message-updater.ts

@@ -205,6 +205,7 @@ export function update(adapter: Adapter, event: SessionEvent.DurableEvent) {
                 draft.finish = undefined
                 draft.rawFinish = undefined
                 draft.providerState = undefined
+                if (event.data.started !== undefined) draft.time.created = event.data.started
                 draft.time.streamed = undefined
                 draft.time.completed = undefined
                 if (event.data.snapshot) draft.snapshot = { ...draft.snapshot, start: event.data.snapshot }
@@ -228,7 +229,7 @@ export function update(adapter: Adapter, event: SessionEvent.DurableEvent) {
               agent: event.data.agent,
               model: event.data.model,
               metadata: event.metadata,
-              time: { created },
+              time: { created: event.data.started ?? created },
               content: [],
               snapshot: event.data.snapshot ? { start: event.data.snapshot } : undefined,
             }),

packages/core/src/session/runner/publish-llm-event.ts

@@ -3,7 +3,7 @@ import type { Agent } from "@opencode-ai/schema/agent"
 import type { Model } from "@opencode-ai/schema/model"
 import type { RelativePath } from "@opencode-ai/schema/schema"
 import type { Snapshot } from "@opencode-ai/schema/snapshot"
-import { Effect, Fiber, Iterable } from "effect"
+import { type DateTime, Effect, Fiber, Iterable } from "effect"
 import { isReadonlyArrayNonEmpty } from "effect/Array"
 import { Bus } from "../../bus.js"
 import { SessionEvent } from "../event.js"
@@ -20,6 +20,7 @@ type Input = {
   readonly model: Model.Ref
   readonly providerMetadataKey: string
   readonly snapshot?: Snapshot.ID
+  readonly started?: DateTime.Utc
   readonly assistantMessageID: SessionMessage.ID
 }
 
@@ -104,6 +105,7 @@ export const createLLMEventPublisher = (bus: Pick<Bus.Interface, "publish">, inp
       model: input.model,
       assistantMessageID,
       snapshot: input.snapshot,
+      started: input.started,
     })
     return assistantMessageID
   })

packages/core/src/session/runner/step.ts

@@ -10,7 +10,7 @@ import {
   type ToolCall,
 } from "@opencode-ai/ai"
 import type { Agent } from "@opencode-ai/schema/agent"
-import { Cause, Data, Effect, Exit, Fiber, Option, Stream } from "effect"
+import { Cause, Data, DateTime, Effect, Exit, Fiber, Option, Stream } from "effect"
 import { SessionError } from "@opencode-ai/schema/session-error"
 import { Bus } from "../../bus.js"
 import { Permission } from "../../permission.js"
@@ -77,6 +77,7 @@ export const make = Effect.gen(function* () {
       model: input.model.ref,
       providerMetadataKey: input.model.model.route.providerMetadataKey ?? input.model.model.provider,
       snapshot: startSnapshot,
+      started: yield* DateTime.now,
     })
     const toolRuns: Array<{
       readonly call: ToolCall

packages/core/test/session-runner.test.ts

@@ -76,7 +76,7 @@ import { SessionSystemPrompt } from "@opencode-ai/core/session/system-prompt"
 import { ID, Model } from "@opencode-ai/core/model"
 import { Location } from "@opencode-ai/core/location"
 import { Provider } from "@opencode-ai/core/provider"
-import { Cause, Context, Deferred, Effect, Exit, Fiber, Layer, Queue, Schema, Scope, Stream } from "effect"
+import { Cause, Context, DateTime, Deferred, Effect, Exit, Fiber, Layer, Queue, Schema, Scope, Stream } from "effect"
 import { TestClock } from "effect/testing"
 import { asc, desc, eq, sql } from "drizzle-orm"
 import { testEffect } from "./lib/effect"
@@ -2815,6 +2815,42 @@ describe("SessionRunnerLLM", () => {
     expect(systemTexts(s.requests[1])).toContain("Replacement context")
   })
 
+  scenario("includes provider wait in the token throughput interval for a buffered reply", function* (s) {
+    yield* s.admit("Hello")
+    const requested = yield* Deferred.make<void>()
+    yield* s.llm.push(
+      Stream.fromEffect(Deferred.succeed(requested, undefined).pipe(Effect.andThen(Effect.sleep(2_690)))).pipe(
+        Stream.flatMap(() => Stream.make(LLMEvent.stepStart({ index: 0 }))),
+        Stream.concat(
+          Stream.fromEffect(Effect.sleep(10)).pipe(
+            Stream.flatMap(() =>
+              Stream.fromIterable(
+                TestLLM.complete(
+                  { reason: { normalized: "stop" }, usage: { inputTokens: 100, outputTokens: 12 } },
+                  LLMEvent.textStart({ id: "text" }),
+                  LLMEvent.textDelta({ id: "text", text: "Hello. What are you working on?" }),
+                  LLMEvent.textEnd({ id: "text" }),
+                ).slice(1),
+              ),
+            ),
+          ),
+        ),
+      ),
+    )
+    const run = yield* Effect.forkChild(s.resume)
+    yield* Deferred.await(requested)
+    yield* TestClock.adjust(2_700)
+    yield* Fiber.join(run)
+
+    const assistant = requireAssistant(yield* s.context)
+    expect(assistant.tokens?.output).toBe(12)
+    expect(DateTime.toEpochMillis(assistant.time.streamed!) - DateTime.toEpochMillis(assistant.time.created)).toBe(
+      2_700,
+    )
+    yield* replaySessionProjection(sessionID)
+    expect(requireAssistant(yield* s.context).time).toEqual(assistant.time)
+  })
+
   scenario("consumes the full provider stream before recording its boundary and settling local tools", function* (s) {
     yield* s.admit("Echo this")
     const tail = yield* Deferred.make<void>()
@@ -4375,14 +4411,18 @@ describe("SessionRunnerLLM", () => {
   scenario("bounds jittered exponential backoff for eligible pre-output failures", function* (s) {
     yield* s.admit("Retry transport")
     yield* s.llm.push(Stream.fail(providerUnavailable()))
-    yield* s.llm.push(TestLLM.text("Recovered", "retry-success"))
+    yield* s.llm.push(
+      Stream.fromEffect(Effect.sleep(400)).pipe(
+        Stream.flatMap(() => Stream.fromIterable(TestLLM.text("Recovered", "retry-success"))),
+      ),
+    )
 
     const scheduled = yield* subscribeRetries(s)
     const run = yield* s.resume.pipe(Effect.forkChild)
     yield* Queue.take(scheduled)
     yield* TestClock.adjust("1599 millis")
     expect(s.requests).toHaveLength(1)
-    yield* TestClock.adjust("801 millis")
+    yield* TestClock.adjust("1201 millis")
     yield* Fiber.join(run)
 
     expect(s.requests).toHaveLength(2)
@@ -4395,6 +4435,8 @@ describe("SessionRunnerLLM", () => {
     ])
     yield* replaySessionProjection(sessionID)
     expect((yield* s.context).filter((message) => message.type === "assistant")).toHaveLength(1)
+    const assistant = requireAssistant(yield* s.context)
+    expect(DateTime.toEpochMillis(assistant.time.streamed!) - DateTime.toEpochMillis(assistant.time.created)).toBe(400)
   })
 
   scenario("does not start another physical attempt after interruption during retry backoff", function* (s) {

packages/schema/src/session-event.ts

@@ -6,7 +6,7 @@ import { Event } from "./event.js"
 import { FinishReason } from "./llm.js"
 import { Content } from "./tool.js"
 import { Model } from "./model.js"
-import { NonNegativeInt, PositiveInt, RelativePath } from "./schema.js"
+import { DateTimeUtcFromMillis, NonNegativeInt, PositiveInt, RelativePath } from "./schema.js"
 import { FileAttachment } from "./prompt.js"
 import { SessionID } from "./session-id.js"
 import { SessionMetadata } from "./session-metadata.js"
@@ -315,6 +315,8 @@ export namespace Step {
       agent: Agent.ID,
       model: Model.Ref,
       snapshot: Snapshot.ID.pipe(optional),
+      /** Request start, before waiting for provider output. Absent on older events. */
+      started: DateTimeUtcFromMillis.pipe(optional),
     },
   })
   export type Started = typeof Started.Type

packages/schema/test/event.test.ts

@@ -1,9 +1,25 @@
 import { describe, expect, test } from "bun:test"
-import { Schema } from "effect"
+import { DateTime, Schema } from "effect"
 import { Event } from "../src/event.js"
 import { EventLog } from "../src/event-log.js"
+import { SessionEvent } from "../src/session-event.js"
 
 describe("public event schemas", () => {
+  test("step start accepts historical payloads and round-trips request timing", () => {
+    const decode = Schema.decodeUnknownSync(SessionEvent.Step.Started.data)
+    const encode = Schema.encodeSync(SessionEvent.Step.Started.data)
+    const data = {
+      sessionID: "ses_test",
+      assistantMessageID: "msg_test",
+      agent: "build",
+      model: { id: "model", providerID: "provider" },
+    }
+    expect(encode(decode(data))).toEqual(data)
+    expect(encode({ ...decode(data), started: undefined })).toEqual(data)
+    expect(encode({ ...decode(data), started: DateTime.makeUnsafe(0) })).toEqual({ ...data, started: 0 })
+    expect(decode({ ...data, started: 1_000 }).started).toEqual(DateTime.makeUnsafe(1_000))
+  })
+
   test("definition is pure", () => {
     const definitions = Event.inventory()
     Event.ephemeral({ type: "test.pure", schema: { value: Schema.String } })