fix(cli): serialize concurrent subagent cost propagation

packages/opencode/src/kilocode/session/cost-propagation.ts

@@ -5,6 +5,28 @@ import { MessageV2 } from "@/session/message-v2"
 import { SessionID, MessageID } from "@/session/schema"
 
 export namespace KiloCostPropagation {
+  /**
+   * Per-key promise chain that serializes concurrent `propagate` calls against
+   * the same parent message. Prevents lost updates when the LLM launches
+   * several `task` tool calls in parallel (each release stage races to
+   * read-modify-write the same parent cost field).
+   */
+  const locks = new Map<string, Promise<void>>()
+
+  function acquire(key: string): Promise<() => void> {
+    const prev = locks.get(key) ?? Promise.resolve()
+    let release!: () => void
+    const current = new Promise<void>((r) => (release = r))
+    const chain = prev.catch(() => {}).then(() => current)
+    locks.set(key, chain)
+    return prev
+      .catch(() => {})
+      .then(() => () => {
+        release()
+        if (locks.get(key) === chain) locks.delete(key)
+      })
+  }
+
   /**
    * Total assistant-message cost in a session. Because each subagent propagates
    * its own total into the parent assistant message when it finishes, this sum
@@ -22,8 +44,8 @@ export namespace KiloCostPropagation {
    * Add `amount` to the given parent assistant message's cost. No-op when
    * `amount` is non-positive or the target is not an assistant message.
    *
-   * Caller must guarantee serial access to the parent message — concurrent
-   * calls against the same `(sid, mid)` are not atomic (read-modify-write).
+   * Concurrent calls against the same parent are serialized internally so the
+   * read-modify-write cannot lose updates when subagents complete in parallel.
    */
   export const propagate = Effect.fn("KiloCostPropagation.propagate")(function* (
     sessions: Session.Interface,
@@ -32,9 +54,16 @@ export namespace KiloCostPropagation {
     amount: number,
   ) {
     if (!(amount > 0)) return
-    const parent = yield* Effect.sync(() => MessageV2.get({ sessionID: sid, messageID: mid }))
-    if (parent.info.role !== "assistant") return
-    parent.info.cost += amount
-    yield* sessions.updateMessage(parent.info)
+    yield* Effect.acquireUseRelease(
+      Effect.promise(() => acquire(`${sid}:${mid}`)),
+      () =>
+        Effect.gen(function* () {
+          const parent = yield* Effect.sync(() => MessageV2.get({ sessionID: sid, messageID: mid }))
+          if (parent.info.role !== "assistant") return
+          parent.info.cost += amount
+          yield* sessions.updateMessage(parent.info)
+        }),
+      (release) => Effect.sync(() => release()),
+    )
   })
 }

packages/opencode/test/kilocode/cost-propagation.test.ts

@@ -0,0 +1,94 @@
+// Verifies KiloCostPropagation.propagate() serializes concurrent writes to
+// the same parent assistant message. Without the internal lock, parallel
+// subagent completions race on read-modify-write and lose deltas (#6321).
+
+import { afterEach, describe, expect } from "bun:test"
+import { Effect, Layer } from "effect"
+import { Bus } from "../../src/bus"
+import * as CrossSpawnSpawner from "../../src/effect/cross-spawn-spawner"
+import { KiloCostPropagation } from "../../src/kilocode/session/cost-propagation"
+import { Instance } from "../../src/project/instance"
+import { ProviderID, ModelID } from "../../src/provider/schema"
+import { Session } from "../../src/session"
+import { MessageV2 } from "../../src/session/message-v2"
+import { MessageID } from "../../src/session/schema"
+import { Log } from "../../src/util"
+import { provideTmpdirInstance } from "../fixture/fixture"
+import { testEffect } from "../lib/effect"
+
+Log.init({ print: false })
+
+afterEach(async () => {
+  await Instance.disposeAll()
+})
+
+const ref = {
+  providerID: ProviderID.make("test"),
+  modelID: ModelID.make("test-model"),
+}
+
+const it = testEffect(Layer.mergeAll(Session.defaultLayer, Bus.layer, CrossSpawnSpawner.defaultLayer))
+
+const seed = Effect.fn("CostPropagationTest.seed")(function* () {
+  const sessions = yield* Session.Service
+  const chat = yield* sessions.create({ title: "parent" })
+  const user = yield* sessions.updateMessage({
+    id: MessageID.ascending(),
+    role: "user",
+    sessionID: chat.id,
+    agent: "build",
+    model: ref,
+    time: { created: Date.now() },
+  })
+  const assistant: MessageV2.Assistant = {
+    id: MessageID.ascending(),
+    role: "assistant",
+    parentID: user.id,
+    sessionID: chat.id,
+    mode: "build",
+    agent: "build",
+    cost: 0,
+    path: { cwd: "/tmp", root: "/tmp" },
+    tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } },
+    modelID: ref.modelID,
+    providerID: ref.providerID,
+    time: { created: Date.now() },
+  }
+  yield* sessions.updateMessage(assistant)
+  return { chat, assistant }
+})
+
+describe("KiloCostPropagation.propagate", () => {
+  it.live("sums deltas correctly under parallel execution", () =>
+    provideTmpdirInstance(() =>
+      Effect.gen(function* () {
+        const sessions = yield* Session.Service
+        const { chat, assistant } = yield* seed()
+        const deltas = [0.01, 0.02, 0.04, 0.08, 0.16, 0.32, 0.64, 1.28]
+        yield* Effect.all(
+          deltas.map((d) => KiloCostPropagation.propagate(sessions, chat.id, assistant.id, d)),
+          { concurrency: "unbounded" },
+        )
+        const parent = yield* Effect.sync(() => MessageV2.get({ sessionID: chat.id, messageID: assistant.id }))
+        expect(parent.info.role).toBe("assistant")
+        if (parent.info.role !== "assistant") return
+        const total = deltas.reduce((a, b) => a + b, 0)
+        expect(parent.info.cost).toBeCloseTo(total, 6)
+      }),
+    ),
+  )
+
+  it.live("is a no-op when amount is non-positive", () =>
+    provideTmpdirInstance(() =>
+      Effect.gen(function* () {
+        const sessions = yield* Session.Service
+        const { chat, assistant } = yield* seed()
+        yield* KiloCostPropagation.propagate(sessions, chat.id, assistant.id, 0)
+        yield* KiloCostPropagation.propagate(sessions, chat.id, assistant.id, -1.5)
+        const parent = yield* Effect.sync(() => MessageV2.get({ sessionID: chat.id, messageID: assistant.id }))
+        if (parent.info.role !== "assistant") return
+        expect(parent.info.cost).toBe(0)
+      }),
+    ),
+  )
+})