diff --git a/packages/core/src/session/pending.ts b/packages/core/src/session/pending.ts index 986d624a6f4..2b9bdac8be0 100644 --- a/packages/core/src/session/pending.ts +++ b/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() export class LifecycleConflict extends Schema.TaggedErrorClass()( @@ -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), diff --git a/packages/core/test/session-create.test.ts b/packages/core/test/session-create.test.ts index cdc8175fb3b..06023e2009f 100644 --- a/packages/core/test/session-create.test.ts +++ b/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, diff --git a/packages/core/test/session-prompt.test.ts b/packages/core/test/session-prompt.test.ts index b292a11c434..a735cd82e95 100644 --- a/packages/core/test/session-prompt.test.ts +++ b/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