ceos/opencode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(core): simplify prompt inbox promotion and projection

Browse source 
Remove the steer promotion cutoff: a turn boundary now promotes every
steer pending at promotion time instead of snapshotting the inbox at a
captured sequence, so late-arriving steers ride along into the turn.

Delete legacy projected-prompt synthesis from projectPrompted. The
event-sourced session input migration wiped pre-inbox event history,
so every Prompted event follows an admitted inbox row and a missing or
divergent row on replay is an invariant violation.

Inline the prompt equivalence predicate and update tests that seeded
history with bare Prompted events to publish PromptAdmitted first.
This commit is contained in:
Kit Langton 2026-07-02 16:46:16 -04:00
commit 606de48d8b
7 changed files with 69 additions and 112 deletions
Download patch file Download diff file Expand all files Collapse all files

32
packages/core/src/session/input.ts
View file

@ -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))

5
packages/core/src/session/runner/llm.ts
View file

@ -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
}

13
packages/core/test/session-compact.test.ts
View file

@ -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",
})

14
packages/core/test/session-projector.test.ts
View file

@ -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,
{

68
packages/core/test/session-prompt.test.ts
View file

@ -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

6
packages/core/test/session-runner.test.ts
View file

@ -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,

43
packages/core/test/session-title.test.ts
View file

@ -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)