diff --git a/packages/core/src/session/input.ts b/packages/core/src/session/input.ts index 14b613678d..2c6aeda460 100644 --- a/packages/core/src/session/input.ts +++ b/packages/core/src/session/input.ts @@ -1,6 +1,6 @@ export * as SessionInput from "./input" -import { and, asc, eq, isNull, lte } from "drizzle-orm" +import { and, asc, eq, isNull } from "drizzle-orm" import { DateTime, Effect, Schema } from "effect" import { Admitted, Delivery } from "@opencode-ai/schema/session-input" import type { Database } from "../database/database" @@ -145,26 +145,11 @@ export const projectPrompted = Effect.fn("SessionInput.projectPrompted")(functio return } + // Every Prompted event is published from an admitted inbox row, so a missing or + // divergent row on replay is an invariant violation. const stored = yield* find(db, input.id) - if (stored) { - if (!matchesProjection(stored, input) || stored.promotedSeq !== input.promotedSeq) - return yield* Effect.die(new LifecycleConflict({ id: input.id })) - return - } - - yield* db - .insert(SessionInputTable) - .values({ - id: input.id, - session_id: input.sessionID, - prompt: encodePrompt(input.prompt), - delivery: input.delivery, - admitted_seq: input.promotedSeq, - promoted_seq: input.promotedSeq, - time_created: DateTime.toEpochMillis(input.timeCreated), - }) - .run() - .pipe(Effect.orDie) + if (!stored || !matchesProjection(stored, input) || stored.promotedSeq !== input.promotedSeq) + return yield* Effect.die(new LifecycleConflict({ id: input.id })) }) export const hasPending = Effect.fn("SessionInput.hasPending")(function* ( @@ -195,9 +180,8 @@ export const equivalent = ( readonly prompt: Prompt readonly delivery: Delivery }, -) => input.delivery === expected.delivery && matchesPrompt(input, expected) - -const matchesPrompt = (input: Admitted, expected: { readonly sessionID: SessionSchema.ID; readonly prompt: Prompt }) => +) => + input.delivery === expected.delivery && input.sessionID === expected.sessionID && JSON.stringify(encodePrompt(input.prompt)) === JSON.stringify(encodePrompt(expected.prompt)) @@ -246,7 +230,6 @@ export const promoteSteers = Effect.fn("SessionInput.promoteSteers")(function* ( db: DatabaseService, events: EventV2.Interface, sessionID: SessionSchema.ID, - cutoff: number, ) { const rows = yield* db .select() @@ -256,7 +239,6 @@ export const promoteSteers = Effect.fn("SessionInput.promoteSteers")(function* ( eq(SessionInputTable.session_id, sessionID), isNull(SessionInputTable.promoted_seq), eq(SessionInputTable.delivery, "steer"), - lte(SessionInputTable.admitted_seq, cutoff), ), ) .orderBy(asc(SessionInputTable.admitted_seq)) diff --git a/packages/core/src/session/runner/llm.ts b/packages/core/src/session/runner/llm.ts index 6484e27c98..f9f5e37ffb 100644 --- a/packages/core/src/session/runner/llm.ts +++ b/packages/core/src/session/runner/llm.ts @@ -173,12 +173,11 @@ const layer = Layer.effect( let needsContinuation = false let currentStep = step if (promotion) { - const cutoff = yield* EventV2.latestSequence(db, session.id) let promoted = 0 - if (promotion === "steer") promoted = yield* SessionInput.promoteSteers(db, events, session.id, cutoff) + if (promotion === "steer") promoted = yield* SessionInput.promoteSteers(db, events, session.id) if (promotion === "queue") { promoted += Number(yield* SessionInput.promoteNextQueued(db, events, session.id)) - promoted += yield* SessionInput.promoteSteers(db, events, session.id, cutoff) + promoted += yield* SessionInput.promoteSteers(db, events, session.id) } if (promoted > 0) currentStep = 1 } diff --git a/packages/core/test/session-compact.test.ts b/packages/core/test/session-compact.test.ts index 41190d470b..3de83577cd 100644 --- a/packages/core/test/session-compact.test.ts +++ b/packages/core/test/session-compact.test.ts @@ -81,11 +81,20 @@ describe("SessionV2.compact", () => { const events = yield* EventV2.Service const created = yield* session.create({ location }) + const messageID = SessionMessage.ID.create() + const prompt = Prompt.make({ text: "Please compact this session history." }) + yield* events.publish(SessionEvent.PromptAdmitted, { + sessionID: created.id, + messageID, + timestamp: DateTime.makeUnsafe(0), + prompt, + delivery: "steer", + }) yield* events.publish(SessionEvent.Prompted, { sessionID: created.id, - messageID: SessionMessage.ID.create(), + messageID, timestamp: DateTime.makeUnsafe(0), - prompt: Prompt.make({ text: "Please compact this session history." }), + prompt, delivery: "steer", }) diff --git a/packages/core/test/session-projector.test.ts b/packages/core/test/session-projector.test.ts index c78abd17d5..64e037987b 100644 --- a/packages/core/test/session-projector.test.ts +++ b/packages/core/test/session-projector.test.ts @@ -118,6 +118,13 @@ describe("SessionProjector", () => { .pipe(Effect.orDie) const events = yield* EventV2.Service + yield* events.publish(SessionEvent.PromptAdmitted, { + sessionID, + messageID: SessionMessage.ID.make("msg_first"), + timestamp: created, + prompt: Prompt.make({ text: "first" }), + delivery: "steer", + }) yield* events.publish( SessionEvent.Prompted, { @@ -129,6 +136,13 @@ describe("SessionProjector", () => { }, { id: EventV2.ID.make("evt_z") }, ) + yield* events.publish(SessionEvent.PromptAdmitted, { + sessionID, + messageID: SessionMessage.ID.make("msg_second"), + timestamp: created, + prompt: Prompt.make({ text: "second" }), + delivery: "steer", + }) yield* events.publish( SessionEvent.Prompted, { diff --git a/packages/core/test/session-prompt.test.ts b/packages/core/test/session-prompt.test.ts index 8eb8fb4a27..16d99148cc 100644 --- a/packages/core/test/session-prompt.test.ts +++ b/packages/core/test/session-prompt.test.ts @@ -197,7 +197,7 @@ describe("SessionV2.prompt", () => { prompt: Prompt.make({ text: "boundary" }), resume: false, }) - yield* SessionInput.promoteSteers(db, events, sessionID, Number.MAX_SAFE_INTEGER) + yield* SessionInput.promoteSteers(db, events, sessionID) const stale = SessionMessage.ID.make("msg_stale_assistant") yield* db.insert(SessionMessageTable).values(assistantRow(stale, 100)).run().pipe(Effect.orDie) yield* events.publish(SessionEvent.RevertEvent.Staged, { @@ -250,7 +250,7 @@ describe("SessionV2.prompt", () => { yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "First" }), resume: false }) yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Second" }), resume: false }) - yield* SessionInput.promoteSteers(db, events, sessionID, Number.MAX_SAFE_INTEGER) + yield* SessionInput.promoteSteers(db, events, sessionID) const streamed = Array.from(yield* Fiber.join(fiber)) expect(streamed.map((event) => [event.durable?.seq, event.type])).toEqual([ @@ -424,10 +424,7 @@ describe("SessionV2.prompt", () => { yield* session.prompt({ id: messageID, sessionID, prompt: Prompt.make({ text: "Promote once" }), resume: false }) yield* Effect.all( - [ - SessionInput.promoteSteers(db, events, sessionID, Number.MAX_SAFE_INTEGER), - SessionInput.promoteSteers(db, events, sessionID, Number.MAX_SAFE_INTEGER), - ], + [SessionInput.promoteSteers(db, events, sessionID), SessionInput.promoteSteers(db, events, sessionID)], { concurrency: "unbounded" }, ) @@ -439,23 +436,6 @@ describe("SessionV2.prompt", () => { }), ) - it.effect("promotes steers only through the captured inbox cutoff", () => - Effect.gen(function* () { - yield* setup - const { db } = yield* Database.Service - const session = yield* SessionV2.Service - const events = yield* EventV2.Service - const first = yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Before cutoff" }), resume: false }) - const cutoff = first.admittedSeq - const second = yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "After cutoff" }), resume: false }) - - yield* SessionInput.promoteSteers(db, events, sessionID, cutoff) - - expect(yield* admitted(first.id)).toHaveProperty("promotedSeq") - expect(yield* admitted(second.id)).not.toHaveProperty("promotedSeq") - }), - ) - it.effect("reprojects pending inbox input without scheduling execution", () => Effect.gen(function* () { yield* setup @@ -499,48 +479,6 @@ describe("SessionV2.prompt", () => { }), ) - it.effect("returns an exact retry of a legacy projected prompt", () => - Effect.gen(function* () { - yield* setup - const session = yield* SessionV2.Service - const events = yield* EventV2.Service - const prompt = Prompt.make({ text: "Historical prompt" }) - yield* events.publish(SessionEvent.Prompted, { - sessionID, - messageID, - timestamp: yield* DateTime.now, - prompt, - delivery: "steer", - }) - - const retried = yield* session.prompt({ id: messageID, sessionID, prompt, resume: false }) - - expect(retried).toMatchObject({ id: messageID, prompt: { text: "Historical prompt" } }) - expect(yield* admitted(messageID)).toHaveProperty("promotedSeq") - }), - ) - - it.effect("returns an exact retry of a legacy projected queued prompt", () => - Effect.gen(function* () { - yield* setup - const session = yield* SessionV2.Service - const events = yield* EventV2.Service - const prompt = Prompt.make({ text: "Historical queued prompt" }) - yield* events.publish(SessionEvent.Prompted, { - sessionID, - messageID, - timestamp: yield* DateTime.now, - prompt, - delivery: "queue", - }) - - const retried = yield* session.prompt({ id: messageID, sessionID, prompt, delivery: "queue", resume: false }) - - expect(retried).toMatchObject({ id: messageID, prompt: { text: "Historical queued prompt" } }) - expect(yield* admitted(messageID)).toMatchObject({ delivery: "queue" }) - }), - ) - it.effect("rejects reuse of one globally unique message ID across sessions", () => Effect.gen(function* () { yield* setup diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts index f6fde4e7ab..c07946c788 100644 --- a/packages/core/test/session-runner.test.ts +++ b/packages/core/test/session-runner.test.ts @@ -2300,7 +2300,7 @@ describe("SessionRunnerLLM", () => { const session = yield* SessionV2.Service const events = yield* EventV2.Service yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Recover interrupted tool" }), resume: false }) - yield* SessionInput.promoteSteers((yield* Database.Service).db, events, sessionID, Number.MAX_SAFE_INTEGER) + yield* SessionInput.promoteSteers((yield* Database.Service).db, events, sessionID) const assistantMessageID = SessionMessage.ID.create() yield* events.publish(SessionEvent.Step.Started, { sessionID, @@ -2364,7 +2364,7 @@ describe("SessionRunnerLLM", () => { prompt: Prompt.make({ text: "Recover interrupted hosted tool" }), resume: false, }) - yield* SessionInput.promoteSteers((yield* Database.Service).db, events, sessionID, Number.MAX_SAFE_INTEGER) + yield* SessionInput.promoteSteers((yield* Database.Service).db, events, sessionID) const assistantMessageID = SessionMessage.ID.create() yield* events.publish(SessionEvent.Step.Started, { sessionID, @@ -2424,7 +2424,7 @@ describe("SessionRunnerLLM", () => { prompt: Prompt.make({ text: "Recover interrupted tool input" }), resume: false, }) - yield* SessionInput.promoteSteers((yield* Database.Service).db, events, sessionID, Number.MAX_SAFE_INTEGER) + yield* SessionInput.promoteSteers((yield* Database.Service).db, events, sessionID) const assistantMessageID = SessionMessage.ID.create() yield* events.publish(SessionEvent.Step.Started, { sessionID, diff --git a/packages/core/test/session-title.test.ts b/packages/core/test/session-title.test.ts index 3e945e3eac..ef51b79e7b 100644 --- a/packages/core/test/session-title.test.ts +++ b/packages/core/test/session-title.test.ts @@ -39,7 +39,14 @@ const client = Layer.mock(LLMClient.Service)({ const models = Layer.mock(SessionRunnerModel.Service)({ resolve: () => Effect.succeed(model) }) const it = testEffect( AppNodeBuilder.build( - LayerNode.group([Database.node, EventV2.node, SessionProjector.node, SessionStore.node, AgentV2.node, SessionTitle.node]), + LayerNode.group([ + Database.node, + EventV2.node, + SessionProjector.node, + SessionStore.node, + AgentV2.node, + SessionTitle.node, + ]), [ [llmClient, client], [SessionRunnerModel.node, models], @@ -74,9 +81,17 @@ const insertSession = (id: SessionV2.ID) => const prompt = (sessionID: SessionV2.ID, text: string) => Effect.gen(function* () { const events = yield* EventV2.Service + const messageID = SessionMessage.ID.create() + yield* events.publish(SessionEvent.PromptAdmitted, { + sessionID, + messageID, + timestamp: DateTime.makeUnsafe(0), + prompt: Prompt.make({ text }), + delivery: "steer", + }) yield* events.publish(SessionEvent.Prompted, { sessionID, - messageID: SessionMessage.ID.create(), + messageID, timestamp: DateTime.makeUnsafe(0), prompt: Prompt.make({ text }), delivery: "steer", @@ -99,9 +114,9 @@ it.effect("generates a title from the sole user message and renames the session" yield* prompt(sessionID, "Help me debug the failing build") const store = yield* SessionStore.Service - const session = yield* store.get(sessionID).pipe( - Effect.flatMap((session) => (session ? Effect.succeed(session) : Effect.die("session missing"))), - ) + const session = yield* store + .get(sessionID) + .pipe(Effect.flatMap((session) => (session ? Effect.succeed(session) : Effect.die("session missing")))) const title = yield* SessionTitle.Service yield* title.generateForFirstPrompt(session) @@ -129,9 +144,9 @@ it.effect("does not generate once a second user message exists", () => yield* prompt(sessionID, "Second message") const store = yield* SessionStore.Service - const session = yield* store.get(sessionID).pipe( - Effect.flatMap((session) => (session ? Effect.succeed(session) : Effect.die("session missing"))), - ) + const session = yield* store + .get(sessionID) + .pipe(Effect.flatMap((session) => (session ? Effect.succeed(session) : Effect.die("session missing")))) const title = yield* SessionTitle.Service yield* title.generateForFirstPrompt(session) @@ -177,9 +192,9 @@ it.effect("does not generate for a child session", () => yield* prompt(sessionID, "Do this subtask") const store = yield* SessionStore.Service - const session = yield* store.get(sessionID).pipe( - Effect.flatMap((session) => (session ? Effect.succeed(session) : Effect.die("session missing"))), - ) + const session = yield* store + .get(sessionID) + .pipe(Effect.flatMap((session) => (session ? Effect.succeed(session) : Effect.die("session missing")))) const title = yield* SessionTitle.Service yield* title.generateForFirstPrompt(session) @@ -195,9 +210,9 @@ it.effect("does not generate when the title agent is removed", () => yield* prompt(sessionID, "Help me debug the failing build") const store = yield* SessionStore.Service - const session = yield* store.get(sessionID).pipe( - Effect.flatMap((session) => (session ? Effect.succeed(session) : Effect.die("session missing"))), - ) + const session = yield* store + .get(sessionID) + .pipe(Effect.flatMap((session) => (session ? Effect.succeed(session) : Effect.die("session missing")))) const title = yield* SessionTitle.Service yield* title.generateForFirstPrompt(session)