fix(core): reconcile promoted prompt retries from messages (#40664)

packages/core/src/session/pending.ts

@@ -12,10 +12,8 @@ import {
   User,
   UserData,
 } from "@opencode-ai/schema/session-pending"
-import { Event } from "@opencode-ai/schema/event"
 import type { Database } from "../database/database"
 import { Bus } from "../bus"
-import { EventTable } from "../event/sql"
 import { KeyedMutex } from "../effect/keyed-mutex"
 import { SessionEvent } from "./event"
 import { SessionMessage } from "./message"
@@ -37,11 +35,7 @@ const decodeUser = Schema.decodeUnknownSync(UserData)
 const encodeUser = Schema.encodeSync(UserData)
 const decodeSynthetic = Schema.decodeUnknownSync(SyntheticData)
 const encodeSynthetic = Schema.encodeSync(SyntheticData)
-const decodeAdmittedEvent = Schema.decodeUnknownOption(SessionEvent.InputAdmitted.data)
-const admittedEventType = Bus.versionedType(
-  SessionEvent.InputAdmitted.type,
-  SessionEvent.InputAdmitted.durable.version,
-)
+const decodeMessage = Schema.decodeUnknownSync(SessionMessage.Info)
 const inboxLocks = KeyedMutex.makeUnsafe<SessionSchema.ID>()
 
 export class LifecycleConflict extends Schema.TaggedErrorClass<LifecycleConflict>()(
@@ -103,46 +97,35 @@ export const compaction = Effect.fn("SessionPending.compaction")(function* (
   return entry.type === "compaction" ? entry : undefined
 })
 
-/**
- * Reconstruct the admitted record for a pending row that was already consumed
- * by promotion. The projected `session_message` row proves promotion happened;
- * the durable `session.input.admitted` event retains the exact admitted
- * message, including delivery.
- */
-const promotedFromHistory = Effect.fn("SessionPending.promotedFromHistory")(function* (
+const promotedFromMessage = Effect.fn("SessionPending.promotedFromMessage")(function* (
   db: DatabaseService,
   sessionID: SessionSchema.ID,
   id: SessionMessage.ID,
+  delivery: Delivery,
 ) {
-  const message = yield* db
+  const row = yield* db
     .select()
     .from(SessionMessageTable)
     .where(eq(SessionMessageTable.id, id))
     .get()
     .pipe(Effect.orDie)
-  if (message === undefined) return undefined
-  if (message.session_id !== sessionID || (message.type !== "user" && message.type !== "synthetic"))
+  if (row === undefined) return undefined
+  if (row.session_id !== sessionID || (row.type !== "user" && row.type !== "synthetic"))
     return yield* Effect.die(new LifecycleConflict({ id }))
-  const rows = yield* db
-    .select()
-    .from(EventTable)
-    .where(and(eq(EventTable.aggregate_id, sessionID), eq(EventTable.type, admittedEventType)))
-    .all()
-    .pipe(Effect.orDie)
-  for (const row of rows) {
-    const decoded = decodeAdmittedEvent(row.data)
-    if (decoded._tag !== "Some" || decoded.value.inputID !== id) continue
-    const base = {
-      id,
-      sessionID,
-      timeCreated: DateTime.makeUnsafe(row.created),
-    }
-    return decoded.value.input.type === "user"
-      ? User.make({ ...base, ...decoded.value.input })
-      : Synthetic.make({ ...base, ...decoded.value.input })
-  }
-  // A projected message without an admitted event in this aggregate (for
-  // example fork-copied history) is not a retryable admission.
+  const message = decodeMessage({ ...row.data, id: row.id, type: row.type })
+  const base = { id, sessionID, timeCreated: message.time.created, delivery }
+  if (message.type === "user")
+    return User.make({
+      ...base,
+      type: "user",
+      data: decodeUser(message),
+    })
+  if (message.type === "synthetic")
+    return Synthetic.make({
+      ...base,
+      type: "synthetic",
+      data: decodeSynthetic(message),
+    })
   return yield* Effect.die(new LifecycleConflict({ id }))
 })
 
@@ -160,7 +143,7 @@ export const admit = Effect.fn("SessionPending.admit")(function* (
     if (existing.type === "compaction") return yield* Effect.die(new LifecycleConflict({ id: request.id }))
     return existing
   }
-  const promoted = yield* promotedFromHistory(db, request.sessionID, request.id)
+  const promoted = yield* promotedFromMessage(db, request.sessionID, request.id, request.input.delivery)
   if (promoted !== undefined) return promoted
   return yield* bus
     .publish(SessionEvent.InputAdmitted, {
@@ -426,7 +409,7 @@ const publish = Effect.fn("SessionPending.publish")(function* (
         .pipe(
           Effect.catchDefect((defect) =>
             defect instanceof LifecycleConflict
-              ? promotedFromHistory(db, sessionID, entry.id).pipe(
+              ? promotedFromMessage(db, sessionID, entry.id, entry.delivery).pipe(
                   Effect.flatMap((stored) => (stored !== undefined ? Effect.void : Effect.die(defect))),
                 )
               : Effect.die(defect),

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

@@ -284,13 +284,9 @@ describe("Session.create", () => {
       })
       expect(yield* SessionPending.find(db, forkContext[0].id)).toBeUndefined()
       expect(yield* SessionPending.find(db, forkContext[1].id)).toBeUndefined()
-      // Fork-copied messages have no admitted event in the fork aggregate, so
-      // reusing their IDs as prompt IDs is conflicting reuse, not a retry.
       expect(
-        yield* session
-          .prompt({ id: forkContext[0].id, sessionID: forked.id, text: "First", resume: false })
-          .pipe(Effect.flip),
-      ).toMatchObject({ _tag: "Session.PromptConflictError", messageID: forkContext[0].id })
+        yield* session.prompt({ id: forkContext[0].id, sessionID: forked.id, text: "First", resume: false }),
+      ).toMatchObject({ id: forkContext[0].id, type: "user", data: { text: "First" } })
 
       yield* session.prompt({
         sessionID: parent.id,

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

@@ -553,6 +553,47 @@ describe("Session.prompt", () => {
     }),
   )
 
+  it.effect("reconciles an exact retry from the promoted message without admission history", () =>
+    Effect.gen(function* () {
+      yield* setup
+      const session = yield* Session.Service
+      const bus = yield* Bus.Service
+      const { db } = yield* Database.Service
+      const input = { sessionID, id: messageID, text: "Fix the failing tests", resume: false }
+      const first = yield* session.prompt(input)
+      yield* SessionPending.promote(db, bus, sessionID, "steer")
+      yield* db
+        .delete(EventTable)
+        .where(eq(EventTable.aggregate_id, sessionID))
+        .run()
+        .pipe(Effect.orDie)
+
+      const retried = yield* session.prompt(input)
+
+      expect(retried).toMatchObject({ id: first.id, type: "user", data: { text: first.data.text } })
+      expect(yield* session.messages({ sessionID })).toMatchObject([
+        { id: messageID, type: "user", text: "Fix the failing tests" },
+      ])
+    }),
+  )
+
+  it.effect("ignores delivery when retrying a promoted message", () =>
+    Effect.gen(function* () {
+      yield* setup
+      const session = yield* Session.Service
+      const bus = yield* Bus.Service
+      const { db } = yield* Database.Service
+      const input = { sessionID, id: messageID, text: "Fix the failing tests", resume: false }
+      yield* session.prompt(input)
+      yield* SessionPending.promote(db, bus, sessionID, "steer")
+
+      const retried = yield* session.prompt({ ...input, delivery: "queue" })
+
+      expect(retried).toMatchObject({ id: messageID, type: "user", data: { text: input.text } })
+      expect(yield* admitted(messageID)).toBeUndefined()
+    }),
+  )
+
   it.effect("wakes execution when an exact prompt retry recovers a committed message", () =>
     Effect.gen(function* () {
       yield* setup