diff --git a/packages/opencode/src/session/message-v2.ts b/packages/opencode/src/session/message-v2.ts index 1bea9f52c3ec..6ee3de01c55f 100644 --- a/packages/opencode/src/session/message-v2.ts +++ b/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 diff --git a/packages/opencode/src/session/processor.ts b/packages/opencode/src/session/processor.ts index 710fd47682b3..4a68e0bf99fe 100644 --- a/packages/opencode/src/session/processor.ts +++ b/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 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 } @@ -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, diff --git a/packages/opencode/src/session/prompt.ts b/packages/opencode/src/session/prompt.ts index 69dca1211a83..472c1aecf0ab 100644 --- a/packages/opencode/src/session/prompt.ts +++ b/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, }) diff --git a/packages/opencode/src/session/tools.ts b/packages/opencode/src/session/tools.ts index 805d2253b8b6..815a87818f7e 100644 --- a/packages/opencode/src/session/tools.ts +++ b/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 + processor: Pick 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 }, diff --git a/packages/opencode/test/session/compaction.test.ts b/packages/opencode/test/session/compaction.test.ts index 73a46689e2d5..008f230a2272 100644 --- a/packages/opencode/test/session/compaction.test.ts +++ b/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)), diff --git a/packages/opencode/test/session/prompt.test.ts b/packages/opencode/test/session/prompt.test.ts index 818035e4275c..9272720dcb07 100644 --- a/packages/opencode/test/session/prompt.test.ts +++ b/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 = [] + 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 + 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)