refactor(opencode): cut per-step turn prep

packages/opencode/src/session/message-v2.ts

@@ -25,6 +25,7 @@ import { and } from "drizzle-orm"
 import { desc } from "drizzle-orm"
 import { eq } from "drizzle-orm"
 import { inArray } from "drizzle-orm"
+import { gt } from "drizzle-orm"
 import { lt } from "drizzle-orm"
 import { or } from "drizzle-orm"
 import { MessageTable, PartTable, SessionTable } from "@opencode-ai/core/session/sql"
@@ -92,6 +93,9 @@ const part = (row: typeof PartTable.$inferSelect) =>
     messageID: row.message_id,
   }) as Part
 
+const newer = (row: Cursor) =>
+  or(gt(MessageTable.time_created, row.time), and(eq(MessageTable.time_created, row.time), gt(MessageTable.id, row.id)))
+
 const older = (row: Cursor) =>
   or(lt(MessageTable.time_created, row.time), and(eq(MessageTable.time_created, row.time), lt(MessageTable.id, row.id)))
 
@@ -489,6 +493,23 @@ export function stream(sessionID: SessionID) {
   })
 }
 
+/** Messages after the (time_created, id) cursor; tuple-shaped like older() so the session index serves it. */
+export const after = Effect.fn("MessageV2.after")(function* (input: {
+  sessionID: SessionID
+  time: number
+  id: MessageID
+}) {
+  const { db } = yield* Database.Service
+  const rows = yield* db
+    .select()
+    .from(MessageTable)
+    .where(and(eq(MessageTable.session_id, input.sessionID), newer(input)))
+    .orderBy(MessageTable.time_created, MessageTable.id)
+    .all()
+    .pipe(Effect.orDie)
+  return yield* hydrate(db, rows)
+})
+
 export function parts(messageID: MessageID) {
   return Effect.gen(function* () {
     const { db } = yield* Database.Service

packages/opencode/src/session/processor.ts

@@ -1,8 +1,9 @@
+import { KeyedMutex } from "@opencode-ai/core/effect/keyed-mutex"
 import { LayerNode } from "@opencode-ai/core/effect/layer-node"
 import { PermissionV1 } from "@opencode-ai/core/v1/permission"
 import { Image } from "@/image/image"
 import { SessionV1 } from "@opencode-ai/core/v1/session"
-import { Cause, Deferred, Effect, Exit, Layer, Context, Scope, Schema } from "effect"
+import { Cause, Deferred, Effect, Exit, Fiber, Layer, Context, Scope, Schema } from "effect"
 import * as Stream from "effect/Stream"
 import { Agent } from "@/agent/agent"
 import { Config } from "@/config/config"
@@ -33,6 +34,9 @@ export interface Handle {
   readonly message: SessionV1.Assistant
   readonly firstTokenAt: number | undefined
   readonly requestStartAt: number | undefined
+  readonly firstRequestStartAt: number | undefined
+  readonly snapshotMs: number
+  readonly awaitSnapshot: Effect.Effect<string | undefined>
   readonly updateToolCall: (
     toolCallID: string,
     update: (part: SessionV1.ToolPart) => SessionV1.ToolPart,
@@ -75,6 +79,8 @@ interface ProcessorContext extends Input {
   currentText: SessionV1.TextPart | undefined
   firstTokenAt: number | undefined
   requestStartAt: number | undefined
+  firstRequestStartAt: number | undefined
+  snapshotMs: number
   reasoningMap: Record<string, SessionV1.ReasoningPart>
 }
 
@@ -100,26 +106,50 @@ const layer = Layer.effect(
     const database = yield* Database.Service
 
     const create = Effect.fn("SessionProcessor.create")(function* (input: Input) {
-      // Pre-capture snapshot before the LLM stream starts. The AI SDK
-      // may execute tools internally before emitting start-step events,
-      // so capturing inside the event handler can be too late.
-      const initialSnapshot = yield* snapshot.track()
       const ctx: ProcessorContext = {
         assistantMessage: input.assistantMessage,
         sessionID: input.sessionID,
         model: input.model,
         toolcalls: {},
         shouldBreak: false,
-        snapshot: initialSnapshot,
+        snapshot: undefined,
+        snapshotMs: 0,
         blocked: false,
         needsCompaction: false,
         currentText: undefined,
         firstTokenAt: undefined,
         requestStartAt: undefined,
+        firstRequestStartAt: undefined,
         reasoningMap: {},
       }
       let aborted = false
 
+      // Runs concurrently with the provider request; awaitSnapshot is the barrier
+      // every tool execution and step event takes before touching the result.
+      const snapshotStart = Date.now()
+      const snapshotFiber = yield* snapshot.track().pipe(
+        Effect.tap(() => Effect.sync(() => (ctx.snapshotMs = Date.now() - snapshotStart))),
+        Effect.orDie,
+        Effect.forkIn(scope),
+      )
+      // step-finish clears ctx.snapshot so every step diffs against its own baseline: the
+      // forked fiber serves only the first capture; later steps track fresh, as before.
+      // Serialized so concurrent tool executions share one baseline instead of racing tracks.
+      let snapshotJoined = false
+      const snapshotGate = KeyedMutex.makeUnsafe<"snapshot">()
+      const awaitSnapshot = snapshotGate.withLock("snapshot")(
+        Effect.gen(function* () {
+          if (ctx.snapshot !== undefined) return ctx.snapshot
+          if (!snapshotJoined) {
+            snapshotJoined = true
+            ctx.snapshot = yield* Fiber.join(snapshotFiber)
+          } else {
+            ctx.snapshot = yield* snapshot.track().pipe(Effect.orDie)
+          }
+          return ctx.snapshot
+        }),
+      )
+
       const parse = (e: unknown) =>
         MessageV2.fromError(e, {
           providerID: input.model.providerID,
@@ -429,7 +459,7 @@ const layer = Layer.effect(
             throw new Error(value.message)
 
           case "step-start":
-            if (!ctx.snapshot) ctx.snapshot = yield* snapshot.track()
+            yield* awaitSnapshot
             yield* session.updatePart({
               id: PartID.ascending(),
               messageID: ctx.assistantMessage.id,
@@ -646,6 +676,8 @@ const layer = Layer.effect(
             ctx.reasoningMap = {}
             yield* status.set(ctx.sessionID, { type: "busy" })
             ctx.requestStartAt = Date.now()
+            // Retries re-stamp requestStartAt; keep the first attempt so prep_ms excludes backoff.
+            if (ctx.firstRequestStartAt === undefined) ctx.firstRequestStartAt = ctx.requestStartAt
             const stream = llm.stream(streamInput)
 
             yield* stream.pipe(
@@ -701,6 +733,13 @@ const layer = Layer.effect(
         get requestStartAt() {
           return ctx.requestStartAt
         },
+        get firstRequestStartAt() {
+          return ctx.firstRequestStartAt
+        },
+        get snapshotMs() {
+          return ctx.snapshotMs
+        },
+        awaitSnapshot,
         updateToolCall,
         completeToolCall,
         process,

packages/opencode/src/session/prompt.ts

@@ -1090,15 +1090,48 @@ const layer = Layer.effect(
         const turnStart = Date.now()
         let firstTokenAt: number | undefined
         let firstTokenReqStart: number | undefined
+        let firstRequestStartAt: number | undefined
+        let ttftStep: number | undefined
+        let historyMs = 0
+        let cachedMsgs: SessionV1.WithParts[] | undefined
+        let toolsMs = 0
+        let contextMs = 0
+        let snapshotMs = 0
         const session = yield* sessions.get(sessionID).pipe(Effect.orDie)
 
+        // Per turn, not per step: an instruction file edited by the turn itself applies next turn.
+        const instructionsStart = Date.now()
+        const instructions = yield* instruction.system().pipe(Effect.orDie)
+        contextMs += Date.now() - instructionsStart
+
         while (true) {
           yield* status.set(sessionID, { type: "busy" })
           yield* Effect.logInfo("loop", { "session.id": sessionID, step })
 
-          let msgs = yield* MessageV2.filterCompactedEffect(sessionID).pipe(
-            Effect.provideService(Database.Service, database),
-          )
+          const historyStart = Date.now()
+          let msgs: SessionV1.WithParts[]
+          if (cachedMsgs === undefined) {
+            msgs = yield* MessageV2.filterCompactedEffect(sessionID).pipe(
+              Effect.provideService(Database.Service, database),
+            )
+          } else {
+            const newest = cachedMsgs.reduce((max, m) => (m.info.id > max.info.id ? m : max), cachedMsgs[0]!)
+            const fresh = yield* MessageV2.after({
+              sessionID,
+              time: newest.info.time.created,
+              id: newest.info.id,
+            }).pipe(Effect.provideService(Database.Service, database))
+            // Compaction/summary rows change the retained window; recompute it from scratch.
+            const rewritesHistory = fresh.some(
+              (m) => m.parts.some((p) => p.type === "compaction") || (m.info.role === "assistant" && m.info.summary),
+            )
+            msgs = rewritesHistory
+              ? yield* MessageV2.filterCompactedEffect(sessionID).pipe(
+                  Effect.provideService(Database.Service, database),
+                )
+              : [...cachedMsgs, ...fresh]
+          }
+          historyMs += Date.now() - historyStart
 
           const { user: lastUser, assistant: lastAssistant, finished: lastFinished, tasks } = MessageV2.latest(msgs)
 
@@ -1184,6 +1217,10 @@ const layer = Layer.effect(
           }
           const maxSteps = agent.steps ?? Infinity
           const isLastStep = step >= maxSteps
+          // Cache before reminders: apply pushes synthetic parts into the last user message
+          // in place, and re-caching them would duplicate the reminder on every step.
+          const lastUserIndex = msgs.findLastIndex((m) => m.info.role === "user")
+          cachedMsgs = msgs.map((m, i) => (i === lastUserIndex ? { ...m, parts: [...m.parts] } : m))
           msgs = yield* SessionReminders.apply({ messages: msgs, agent, session }).pipe(
             Effect.provideService(RuntimeFlags.Service, flags),
             Effect.provideService(FSUtil.Service, fsys),
@@ -1230,6 +1267,7 @@ const layer = Layer.effect(
             const bypassAgentCheck = lastUserMsg?.parts.some((p) => p.type === "agent") ?? false
             const promptOps = yield* ops()
 
+            const toolsStart = Date.now()
             const tools = yield* SessionTools.resolve({
               agent,
               session,
@@ -1246,6 +1284,7 @@ const layer = Layer.effect(
               Effect.provideService(Truncate.Service, truncate),
               Effect.provideService(RuntimeFlags.Service, flags),
             )
+            toolsMs += Date.now() - toolsStart
 
             if (lastUser.format?.type === "json_schema") {
               tools["StructuredOutput"] = createStructuredOutputTool({
@@ -1261,13 +1300,14 @@ const layer = Layer.effect(
 
             yield* plugin.trigger("experimental.chat.messages.transform", {}, { messages: msgs })
 
-            const [skills, env, instructions, mcpInstructions, modelMsgs] = yield* Effect.all([
+            const contextStart = Date.now()
+            const [skills, env, mcpInstructions, modelMsgs] = yield* Effect.all([
               sys.skills(agent),
               sys.environment(model),
-              instruction.system().pipe(Effect.orDie),
               sys.mcp(agent, session.permission),
               MessageV2.toModelMessagesEffect(msgs, model),
             ])
+            contextMs += Date.now() - contextStart
             const system = [
               ...env,
               ...instructions,
@@ -1292,9 +1332,16 @@ const layer = Layer.effect(
               toolChoice: format.type === "json_schema" ? "required" : undefined,
             })
 
+            // A tool-only first step emits no token, so the ttft pair can capture from a later
+            // step and straddle tool execution; prep_ms and ttft_step exist to keep that honest.
+            if (firstRequestStartAt === undefined && handle.firstRequestStartAt !== undefined) {
+              firstRequestStartAt = handle.firstRequestStartAt
+            }
+            snapshotMs += handle.snapshotMs
             if (firstTokenAt === undefined && handle.firstTokenAt !== undefined) {
               firstTokenAt = handle.firstTokenAt
               firstTokenReqStart = handle.requestStartAt
+              ttftStep = step
             }
 
             if (structured !== undefined) {
@@ -1354,6 +1401,12 @@ const layer = Layer.effect(
             firstTokenAt !== undefined && firstTokenReqStart !== undefined
               ? firstTokenAt - firstTokenReqStart
               : undefined,
+          prep_ms: firstRequestStartAt !== undefined ? firstRequestStartAt - turnStart : undefined,
+          ttft_step: ttftStep,
+          history_ms: historyMs,
+          tools_ms: toolsMs,
+          context_ms: contextMs,
+          snapshot_ms: snapshotMs,
           turn_ms: Date.now() - turnStart,
           steps: step,
         })

packages/opencode/src/session/tools.ts

@@ -43,7 +43,7 @@ export const resolve = Effect.fn("SessionTools.resolve")(function* (input: {
   agent: Agent.Info
   model: Provider.Model
   session: Session.Info
-  processor: Pick<SessionProcessor.Handle, "message" | "updateToolCall" | "completeToolCall">
+  processor: Pick<SessionProcessor.Handle, "message" | "updateToolCall" | "completeToolCall" | "awaitSnapshot">
   bypassAgentCheck: boolean
   messages: SessionV1.WithParts[]
   promptOps: TaskPromptOps
@@ -104,6 +104,8 @@ export const resolve = Effect.fn("SessionTools.resolve")(function* (input: {
         return run.promise(
           Effect.gen(function* () {
             const ctx = context(args, options)
+            // Tools can run before any step event; the pre-turn snapshot must land before a tool mutates files.
+            yield* input.processor.awaitSnapshot
             yield* plugin.trigger(
               "tool.execute.before",
               { tool: item.id, sessionID: ctx.sessionID, callID: ctx.callID },
@@ -402,6 +404,8 @@ export const resolve = Effect.fn("SessionTools.resolve")(function* (input: {
       run.promise(
         Effect.gen(function* () {
           const ctx = context(args, opts)
+          // MCP tools can mutate the worktree too (workspace-backed files); same barrier as builtins.
+          yield* input.processor.awaitSnapshot
           yield* plugin.trigger(
             "tool.execute.before",
             { tool: key, sessionID: ctx.sessionID, callID: opts.toolCallId },

packages/opencode/test/session/compaction.test.ts

@@ -204,6 +204,9 @@ function fake(
     },
     firstTokenAt: undefined,
     requestStartAt: undefined,
+    firstRequestStartAt: undefined,
+    snapshotMs: 0,
+    awaitSnapshot: Effect.succeed(undefined),
     updateToolCall: Effect.fn("TestSessionProcessor.updateToolCall")(() => Effect.succeed(undefined)),
     completeToolCall: Effect.fn("TestSessionProcessor.completeToolCall")(() => Effect.void),
     process: Effect.fn("TestSessionProcessor.process")(() => Effect.succeed(result)),

packages/opencode/test/session/prompt.test.ts

@@ -6,7 +6,7 @@ import { SessionProjector } from "@opencode-ai/core/session/projector"
 import { eq } from "drizzle-orm"
 import { EventV2Bridge } from "@/event-v2-bridge"
 import { expect } from "bun:test"
-import { Cause, Deferred, Duration, Effect, Exit, Fiber, Layer, Option } from "effect"
+import { Cause, Deferred, Duration, Effect, Exit, Fiber, Layer, Logger, Option, References } from "effect"
 import path from "path"
 import { fileURLToPath } from "url"
 import { NamedError } from "@opencode-ai/core/util/error"
@@ -814,6 +814,50 @@ it.instance("loop continues when finish is tool-calls", () =>
   }),
 )
 
+it.instance("turn.done logs prep_ms from the first request even when ttft captures from a later step", () =>
+  Effect.gen(function* () {
+    const { llm } = yield* useServerConfig(providerCfg)
+    const prompt = yield* SessionPrompt.Service
+    const sessions = yield* Session.Service
+    const session = yield* sessions.create({
+      title: "Turn timing",
+      permission: [{ permission: "*", pattern: "*", action: "allow" }],
+    })
+    yield* prompt.prompt({
+      sessionID: session.id,
+      agent: "build",
+      noReply: true,
+      parts: [{ type: "text", text: "hello" }],
+    })
+    yield* llm.tool("first", { value: "first" })
+    yield* llm.text("second")
+
+    const captured: Array<unknown> = []
+    const capture = Logger.make((options) => {
+      captured.push(options.message)
+    })
+    yield* prompt.loop({ sessionID: session.id }).pipe(
+      Effect.provide(Logger.layer([capture], { mergeWithExisting: true })),
+      Effect.provide(Layer.succeed(References.MinimumLogLevel, "Debug")),
+    )
+
+    const message = captured.find((item) => Array.isArray(item) && item[0] === "turn.done")
+    expect(message).toBeDefined()
+    if (!Array.isArray(message)) return
+    const fields = message[1] as Record<string, number | undefined>
+    expect(fields.steps).toBe(2)
+    // Step 1 was tool-only, so the ttft pair comes from step 2 — the cross-step hazard.
+    expect(fields.ttft_step).toBe(2)
+    // prep_ms must reflect step 1's request, which started before step 2's first token.
+    expect(typeof fields.prep_ms).toBe("number")
+    expect(fields.prep_ms!).toBeLessThanOrEqual(fields.ttft_ms!)
+    for (const phase of ["history_ms", "tools_ms", "context_ms", "snapshot_ms"] as const) {
+      expect(typeof fields[phase]).toBe("number")
+      expect(fields[phase]!).toBeGreaterThanOrEqual(0)
+    }
+  }),
+)
+
 it.instance("glob tool keeps instance context during prompt runs", () =>
   Effect.gen(function* () {
     const { dir, llm } = yield* useServerConfig(providerCfg)