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

feat(core): replace instruction checkpoints with value-delta sync (#36254)

Browse source
This commit is contained in:
Kit Langton 2026-07-10 13:26:25 -04:00 • committed by GitHub
commit 96a9731947
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
57 changed files with 2053 additions and 1278 deletions
Download patch file Download diff file Expand all files Collapse all files

2
packages/core/src/session/compaction.ts
View file

@ -159,7 +159,7 @@ const select = (
tokens: number,
): { readonly head: string; readonly recent: string } | undefined => {
const conversation = messages
.filter((message) => message.type !== "compaction")
.filter((message) => message.type !== "compaction" && message.type !== "system")
.flatMap((message) => {
const text = serialize(message)
return text ? [{ message, text }] : []

58
packages/core/src/session/history.ts
View file

@ -1,10 +1,12 @@
import { and, asc, desc, eq, gt, gte, ne, or, sql } from "drizzle-orm"
import { and, asc, desc, eq, gte, sql } from "drizzle-orm"
import { Effect, Schema } from "effect"
import { Database } from "../database/database"
import { MessageDecodeError } from "./error"
import { SessionMessage } from "./message"
import { SessionSchema } from "./schema"
import { InstructionCheckpointTable, SessionMessageTable } from "./sql"
import { Instructions } from "../instructions/index"
import { InstructionState } from "./instruction-state"
import { SessionMessageTable } from "./sql"
type DatabaseService = Database.Interface["db"]
@ -31,7 +33,6 @@ const messageRows = Effect.fnUntraced(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
compaction: { readonly seq: number } | undefined,
baselineSeq?: number,
) {
const rows = yield* db
.select()
@ -39,20 +40,7 @@ const messageRows = Effect.fnUntraced(function* (
.where(
and(
eq(SessionMessageTable.session_id, sessionID),
// Keep system updates visible in the gap between a completed compaction
// and the next prepared step's rebaseline, when their content is not yet
// folded into a new baseline.
compaction
? or(
gte(SessionMessageTable.seq, compaction.seq),
baselineSeq === undefined
? undefined
: and(eq(SessionMessageTable.type, "system"), gt(SessionMessageTable.seq, baselineSeq)),
)
: undefined,
baselineSeq === undefined
? undefined
: or(ne(SessionMessageTable.type, "system"), gt(SessionMessageTable.seq, baselineSeq)),
compaction ? gte(SessionMessageTable.seq, compaction.seq) : undefined,
),
)
.orderBy(asc(SessionMessageTable.seq))
@ -73,30 +61,32 @@ const decodeMessageRow = (row: typeof SessionMessageTable.$inferSelect) =>
)
export const load = Effect.fn("SessionHistory.load")(function* (db: DatabaseService, sessionID: SessionSchema.ID) {
const [epoch, compaction] = yield* Effect.all(
[
db
.select({ baselineSeq: InstructionCheckpointTable.baseline_seq })
.from(InstructionCheckpointTable)
.where(eq(InstructionCheckpointTable.session_id, sessionID))
.get()
.pipe(Effect.orDie),
latestCompaction(db, sessionID),
],
{ concurrency: "unbounded" },
return yield* Effect.forEach(
yield* messageRows(db, sessionID, yield* latestCompaction(db, sessionID)),
decodeMessageRow,
)
return yield* Effect.forEach(yield* messageRows(db, sessionID, compaction, epoch?.baselineSeq), decodeMessageRow)
})
export const entriesForRunner = Effect.fn("SessionHistory.entriesForRunner")(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
baselineSeq: number,
instructions: Instructions.Instructions,
) {
const rows = yield* messageRows(db, sessionID, yield* latestCompaction(db, sessionID), baselineSeq)
return yield* Effect.forEach(rows, (row) =>
decodeMessageRow(row).pipe(Effect.map((message) => ({ seq: row.seq, message }))),
)
return yield* db
.transaction(() =>
Effect.gen(function* () {
const rows = yield* messageRows(db, sessionID, yield* latestCompaction(db, sessionID))
const messages = yield* Effect.forEach(rows, (row) =>
decodeMessageRow(row).pipe(Effect.map((message) => ({ seq: row.seq, message }))),
)
const assembled = yield* InstructionState.assemble(db, sessionID, instructions)
return {
initial: assembled.initial,
entries: [...messages, ...assembled.updates].toSorted((a, b) => a.seq - b.seq),
}
}),
)
.pipe(Effect.orDie)
})
/** Returns the session's sole user message, or `undefined` once a second one exists. */

130
packages/core/src/session/instruction-checkpoint.ts
View file

@ -1,130 +0,0 @@
export * as InstructionCheckpoint from "./instruction-checkpoint"
import { eq } from "drizzle-orm"
import { Effect, Option, Schema } from "effect"
import type { Database } from "../database/database"
import { EventV2 } from "../event"
import { Instructions } from "../instructions/index"
import { SessionEvent } from "./event"
import { SessionHistory } from "./history"
import { SessionSchema } from "./schema"
import { InstructionCheckpointTable } from "./sql"
type DatabaseService = Database.Interface["db"]
const decodeApplied = Schema.decodeUnknownOption(Instructions.Applied)
/**
* Loads or creates the session's durable instruction checkpoint, narrating any
* drift since the model was last told as a chronological update. Completed
* compaction rebaselines; nothing else rewrites the baseline. Runs before
* input promotion so a blocked first step leaves pending inputs untouched.
*/
export const prepare = Effect.fn("InstructionCheckpoint.prepare")(function* (
db: DatabaseService,
events: EventV2.Interface,
instructions: Effect.Effect<Instructions.Instructions>,
sessionID: SessionSchema.ID,
) {
const [value, stored, compaction] = yield* Effect.all(
[instructions, find(db, sessionID), SessionHistory.latestCompaction(db, sessionID)],
{ concurrency: "unbounded" },
)
if (!stored) {
const baseline = yield* Instructions.initialize(value)
const baselineSeq = yield* insert(db, sessionID, baseline)
return { baseline: baseline.text, baselineSeq }
}
// The applied record is comparison state only; an undecodable one heals by
// treating every source as new, re-announcing baselines as updates.
const applied = Option.getOrElse(decodeApplied(stored.snapshot), () => ({}))
if (compaction !== undefined && compaction.seq > stored.baseline_seq) {
const baseline = yield* Instructions.rebaseline(value, applied)
yield* rewrite(db, sessionID, compaction.seq, baseline)
return { baseline: baseline.text, baselineSeq: compaction.seq }
}
const result = yield* Instructions.reconcile(value, applied)
if (result._tag === "Unchanged") return { baseline: stored.baseline, baselineSeq: stored.baseline_seq }
yield* events.publish(
SessionEvent.InstructionsUpdated,
{ sessionID, text: result.text },
{ commit: () => advance(db, sessionID, result.applied).pipe(Effect.orDie) },
)
return { baseline: stored.baseline, baselineSeq: stored.baseline_seq }
})
export const reset = Effect.fn("InstructionCheckpoint.reset")(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
) {
yield* db
.delete(InstructionCheckpointTable)
.where(eq(InstructionCheckpointTable.session_id, sessionID))
.run()
.pipe(Effect.orDie)
})
const find = Effect.fnUntraced(function* (db: DatabaseService, sessionID: SessionSchema.ID) {
return yield* db
.select()
.from(InstructionCheckpointTable)
.where(eq(InstructionCheckpointTable.session_id, sessionID))
.get()
.pipe(Effect.orDie)
})
const insert = Effect.fnUntraced(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
baseline: Instructions.Baseline,
) {
const baselineSeq = yield* EventV2.latestSequence(db, sessionID)
yield* db
.insert(InstructionCheckpointTable)
.values({
session_id: sessionID,
baseline: baseline.text,
snapshot: baseline.applied,
baseline_seq: baselineSeq,
})
.run()
.pipe(Effect.orDie)
return baselineSeq
})
const rewrite = Effect.fnUntraced(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
baselineSeq: number,
baseline: Instructions.Baseline,
) {
const updated = yield* db
.update(InstructionCheckpointTable)
.set({
baseline: baseline.text,
snapshot: baseline.applied,
baseline_seq: baselineSeq,
})
.where(eq(InstructionCheckpointTable.session_id, sessionID))
.returning({ sessionID: InstructionCheckpointTable.session_id })
.get()
.pipe(Effect.orDie)
if (!updated) return yield* Effect.die(new Error("Instruction checkpoint not found"))
})
const advance = Effect.fnUntraced(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
applied: Instructions.Applied,
) {
const updated = yield* db
.update(InstructionCheckpointTable)
.set({ snapshot: applied })
.where(eq(InstructionCheckpointTable.session_id, sessionID))
.returning({ sessionID: InstructionCheckpointTable.session_id })
.get()
.pipe(Effect.orDie)
if (!updated) return yield* Effect.die(new Error("Instruction checkpoint not found"))
})

80
packages/core/src/session/instruction-entry.ts
View file

@ -1,6 +1,6 @@
export * as InstructionEntry from "./instruction-entry"
import { and, asc, eq } from "drizzle-orm"
import { and, asc, eq, isNotNull, isNull, ne, or } from "drizzle-orm"
import { Context, Effect, Layer, Schema } from "effect"
import { InstructionEntry } from "@opencode-ai/schema/instruction-entry"
import { Database } from "../database/database"
@ -13,6 +13,8 @@ export const Key = InstructionEntry.Key
export type Key = typeof Key.Type
export const Info = InstructionEntry.Info
export type Info = typeof Info.Type
export const MaxValueBytes = InstructionEntry.MaxValueBytes
export const ValueTooLargeError = InstructionEntry.ValueTooLargeError
export interface Interface {
readonly list: (sessionID: SessionSchema.ID) => Effect.Effect<ReadonlyArray<Info>>
@ -20,7 +22,7 @@ export interface Interface {
readonly sessionID: SessionSchema.ID
readonly key: Key
readonly value: Schema.Json
}) => Effect.Effect<void>
}) => Effect.Effect<void, InstructionEntry.ValueTooLargeError>
readonly remove: (input: { readonly sessionID: SessionSchema.ID; readonly key: Key }) => Effect.Effect<void>
/** Produces one Instructions source per stored entry, keyed `api/<key>`. */
readonly load: (sessionID: SessionSchema.ID) => Effect.Effect<Instructions.Instructions>
@ -35,18 +37,20 @@ const renderBlock = (key: Key, value: Schema.Json) =>
// Rendering stays mechanism-neutral: the model sees session context, not how
// it was attached. Only chronological updates and removals carry narration.
const source = (entry: Info) =>
Instructions.make({
const source = (entry: Info & { readonly removed: boolean }) =>
Instructions.make<Schema.Json>({
key: Instructions.Key.make(`api/${entry.key}`),
codec: Schema.toCodecJson(Schema.Json),
load: Effect.succeed(entry.value),
baseline: (value) => renderBlock(entry.key, value),
update: (_previous, value) =>
[
`The context under "${entry.key}" changed and supersedes the previous value:`,
renderBlock(entry.key, value),
].join("\n"),
removed: () => `The context under "${entry.key}" no longer applies. Disregard it.`,
read: Effect.succeed(entry.removed ? Instructions.removed : entry.value),
render: {
initial: (value) => renderBlock(entry.key, value),
changed: (_previous, value) =>
[
`The context under "${entry.key}" changed and supersedes the previous value:`,
renderBlock(entry.key, value),
].join("\n"),
removed: () => `The context under "${entry.key}" no longer applies. Disregard it.`,
},
})
const layer = Layer.effect(
@ -54,15 +58,27 @@ const layer = Layer.effect(
Effect.gen(function* () {
const { db } = yield* Database.Service
const list = Effect.fn("InstructionEntry.list")(function* (sessionID: SessionSchema.ID) {
const rows = yield* db
.select()
const rows = Effect.fnUntraced(function* (sessionID: SessionSchema.ID, includeRemoved: boolean) {
return yield* db
.select({
key: InstructionEntryTable.key,
value: InstructionEntryTable.value,
removed: InstructionEntryTable.removed,
})
.from(InstructionEntryTable)
.where(eq(InstructionEntryTable.session_id, sessionID))
.where(
and(
eq(InstructionEntryTable.session_id, sessionID),
includeRemoved ? undefined : eq(InstructionEntryTable.removed, false),
),
)
.orderBy(asc(InstructionEntryTable.key))
.all()
.pipe(Effect.orDie)
return rows.map((row) => ({ key: row.key, value: row.value }))
})
const list = Effect.fn("InstructionEntry.list")(function* (sessionID: SessionSchema.ID) {
return (yield* rows(sessionID, false)).map((row) => ({ key: row.key, value: row.value }))
})
const put = Effect.fn("InstructionEntry.put")(function* (input: {
@ -70,12 +86,24 @@ const layer = Layer.effect(
readonly key: Key
readonly value: Schema.Json
}) {
const actualBytes = Buffer.byteLength(JSON.stringify(input.value), "utf8")
if (actualBytes > MaxValueBytes)
yield* new ValueTooLargeError({
actualBytes,
maxBytes: MaxValueBytes,
message: `Instruction entry value is ${actualBytes} bytes; the limit is ${MaxValueBytes} bytes`,
})
const changed =
input.value === null
? isNotNull(InstructionEntryTable.value)
: or(isNull(InstructionEntryTable.value), ne(InstructionEntryTable.value, input.value))
yield* db
.insert(InstructionEntryTable)
.values({ session_id: input.sessionID, key: input.key, value: input.value })
.values({ session_id: input.sessionID, key: input.key, value: input.value, removed: false })
.onConflictDoUpdate({
target: [InstructionEntryTable.session_id, InstructionEntryTable.key],
set: { value: input.value, time_updated: Date.now() },
set: { value: input.value, removed: false, time_updated: Date.now() },
setWhere: or(eq(InstructionEntryTable.removed, true), changed),
})
.run()
.pipe(Effect.orDie)
@ -86,15 +114,21 @@ const layer = Layer.effect(
readonly key: Key
}) {
yield* db
.delete(InstructionEntryTable)
.where(and(eq(InstructionEntryTable.session_id, input.sessionID), eq(InstructionEntryTable.key, input.key)))
.update(InstructionEntryTable)
.set({ value: null, removed: true, time_updated: Date.now() })
.where(
and(
eq(InstructionEntryTable.session_id, input.sessionID),
eq(InstructionEntryTable.key, input.key),
eq(InstructionEntryTable.removed, false),
),
)
.run()
.pipe(Effect.orDie)
})
const load = Effect.fn("InstructionEntry.load")(function* (sessionID: SessionSchema.ID) {
const entries = yield* list(sessionID)
return Instructions.combine(entries.map(source))
return Instructions.combine((yield* rows(sessionID, true)).map(source))
})
return Service.of({ list, put, remove, load })

338
packages/core/src/session/instruction-state.ts Normal file
View file

@ -0,0 +1,338 @@
export * as InstructionState from "./instruction-state"
import { and, asc, desc, eq, gt, inArray, lte, sql } from "drizzle-orm"
import { DateTime, Effect, Option, Schema } from "effect"
import type { Database } from "../database/database"
import { EventV2 } from "../event"
import { EventTable } from "../event/sql"
import { Instructions } from "../instructions/index"
import { SessionEvent } from "./event"
import { SessionMessage } from "./message"
import { SessionSchema } from "./schema"
import { InstructionBlobTable, InstructionStateTable, SessionTable } from "./sql"
type DatabaseService = Database.Interface["db"]
const decodeInstructionsUpdated = Schema.decodeUnknownSync(SessionEvent.InstructionsUpdated.data)
export const prepare = Effect.fn("InstructionState.prepare")(function* (
db: DatabaseService,
events: EventV2.Interface,
instructions: Instructions.Instructions,
sessionID: SessionSchema.ID,
) {
const [observed, stored] = yield* Effect.all([Instructions.read(instructions), ensure(db, sessionID)], {
concurrency: "unbounded",
})
const admission = yield* Instructions.diff(observed, stored?.current_values)
if (!stored || Object.keys(admission.delta).length > 0) {
yield* events.publish(
SessionEvent.InstructionsUpdated,
{ sessionID, delta: admission.delta },
{
commit: () => insertBlobs(db, admission.blobs),
},
)
}
})
export const apply = Effect.fn("InstructionState.apply")(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
seq: number,
delta: Instructions.Delta,
) {
const stored = yield* find(db, sessionID)
const current = Instructions.applyHashDelta(stored?.current_values ?? {}, delta)
if (!stored) {
yield* db
.insert(InstructionStateTable)
.values({
session_id: sessionID,
epoch_start: seq,
through_seq: seq,
initial_values: current,
current_values: current,
})
.run()
.pipe(Effect.orDie)
return
}
yield* db
.update(InstructionStateTable)
.set({ through_seq: seq, current_values: current })
.where(eq(InstructionStateTable.session_id, sessionID))
.run()
.pipe(Effect.orDie)
})
export const advanceEpoch = Effect.fn("InstructionState.advanceEpoch")(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
epochStart: number,
) {
yield* db
.update(InstructionStateTable)
.set({
epoch_start: epochStart,
through_seq: epochStart,
initial_values: sql`${InstructionStateTable.current_values}`,
})
.where(eq(InstructionStateTable.session_id, sessionID))
.run()
.pipe(Effect.orDie)
})
export const reset = Effect.fn("InstructionState.reset")(function* (db: DatabaseService, sessionID: SessionSchema.ID) {
yield* db
.delete(InstructionStateTable)
.where(eq(InstructionStateTable.session_id, sessionID))
.run()
.pipe(Effect.orDie)
})
export const rebuild = Effect.fn("InstructionState.rebuild")(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
) {
const folded = fold(yield* instructionEvents(db, sessionID))
if (!folded) {
yield* reset(db, sessionID)
return undefined
}
const state = {
session_id: sessionID,
epoch_start: folded.epochStart,
through_seq: folded.throughSeq,
initial_values: folded.initial,
current_values: folded.current,
}
yield* db
.insert(InstructionStateTable)
.values(state)
.onConflictDoUpdate({
target: InstructionStateTable.session_id,
set: {
epoch_start: folded.epochStart,
through_seq: folded.throughSeq,
initial_values: folded.initial,
current_values: folded.current,
},
})
.run()
.pipe(Effect.orDie)
return state
})
export const assemble = Effect.fn("InstructionState.assemble")(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
instructions: Instructions.Instructions,
) {
const state = yield* find(db, sessionID)
if (!state) return yield* Effect.die(new Error(`Instruction state not found during assembly: ${sessionID}`))
const rows = yield* instructionUpdatesAfter(db, sessionID, state.epoch_start)
const updates = rows.map((row) => ({
row,
delta: decodeInstructionsUpdated(row.data).delta,
}))
const blobs = yield* loadBlobs(db, [
...Object.values(state.initial_values),
...updates.flatMap((update) =>
Object.values(update.delta).filter((hash): hash is Instructions.Hash => hash !== "removed"),
),
])
const valuesAtStart = dereference(state.initial_values, blobs)
let values = valuesAtStart
const result: Array<{ readonly seq: number; readonly message: SessionMessage.System }> = []
for (const update of updates) {
const delta = dereferenceDelta(update.delta, blobs)
const text = Instructions.renderUpdate(instructions, values, delta)
if (text.length > 0)
result.push({
seq: update.row.seq,
message: SessionMessage.System.make({
id: SessionMessage.ID.fromEvent(EventV2.ID.make(update.row.id)),
type: "system",
text,
time: { created: DateTime.makeUnsafe(update.row.created) },
}),
})
values = Instructions.applyDelta(values, delta)
}
return { initial: Instructions.renderInitial(instructions, valuesAtStart), updates: result }
})
const find = Effect.fnUntraced(function* (db: DatabaseService, sessionID: SessionSchema.ID) {
return yield* db
.select()
.from(InstructionStateTable)
.where(eq(InstructionStateTable.session_id, sessionID))
.get()
.pipe(Effect.orDie)
})
const ensure = Effect.fnUntraced(function* (db: DatabaseService, sessionID: SessionSchema.ID) {
const stored = yield* find(db, sessionID)
if (!stored) return yield* rebuild(db, sessionID)
const latest = yield* db
.select({ seq: EventTable.seq })
.from(EventTable)
.where(and(eq(EventTable.aggregate_id, sessionID), inArray(EventTable.type, relevantEventTypes)))
.orderBy(desc(EventTable.seq))
.limit(1)
.get()
.pipe(Effect.orDie)
if (!latest || latest.seq <= stored.through_seq) return stored
return yield* rebuild(db, sessionID)
})
const insertBlobs = Effect.fnUntraced(function* (db: DatabaseService, blobs: Readonly<Record<string, Schema.Json>>) {
const rows = Object.entries(blobs).map(([hash, value]) => ({ hash: Instructions.Hash.make(hash), value }))
if (rows.length === 0) return
yield* db.insert(InstructionBlobTable).values(rows).onConflictDoNothing().run().pipe(Effect.orDie)
})
const loadBlobs = Effect.fnUntraced(function* (db: DatabaseService, values: ReadonlyArray<Instructions.Hash>) {
const hashes = [...new Set(values)]
const batches = Array.from({ length: Math.ceil(hashes.length / 500) }, (_, index) =>
hashes.slice(index * 500, (index + 1) * 500),
)
const rows = (yield* Effect.forEach(
batches,
(batch) =>
db.select().from(InstructionBlobTable).where(inArray(InstructionBlobTable.hash, batch)).all().pipe(Effect.orDie),
{ concurrency: 4 },
)).flat()
const blobs = new Map(rows.map((row) => [row.hash, row.value]))
for (const hash of hashes) {
if (!blobs.has(hash)) return yield* Effect.die(new Error(`Instruction blob not found: ${hash}`))
}
return blobs
})
function dereference(values: Instructions.Values, blobs: ReadonlyMap<Instructions.Hash, Schema.Json>) {
return Object.fromEntries(Object.entries(values).map(([key, hash]) => [key, requireBlob(blobs, hash)])) as Readonly<
Record<string, Schema.Json>
>
}
function dereferenceDelta(delta: Instructions.Delta, blobs: ReadonlyMap<Instructions.Hash, Schema.Json>) {
return Object.fromEntries(
Object.entries(delta).map(([key, hash]) => [
key,
hash === "removed" ? Option.none() : Option.some(requireBlob(blobs, hash)),
]),
) as Readonly<Record<string, Option.Option<Schema.Json>>>
}
function requireBlob(blobs: ReadonlyMap<Instructions.Hash, Schema.Json>, hash: Instructions.Hash) {
const value = blobs.get(hash)
if (value === undefined) throw new Error(`Instruction blob not found: ${hash}`)
return value
}
const instructionEventType = EventV2.versionedType(
SessionEvent.InstructionsUpdated.type,
SessionEvent.InstructionsUpdated.durable.version,
)
const compactionEventType = EventV2.versionedType(
SessionEvent.Compaction.Ended.type,
SessionEvent.Compaction.Ended.durable.version,
)
const movedEventType = EventV2.versionedType(SessionEvent.Moved.type, SessionEvent.Moved.durable.version)
const revertedEventType = EventV2.versionedType(
SessionEvent.RevertEvent.Committed.type,
SessionEvent.RevertEvent.Committed.durable.version,
)
const relevantEventTypes = [instructionEventType, compactionEventType, movedEventType, revertedEventType]
type InstructionEventRow = typeof EventTable.$inferSelect
const instructionEvents = Effect.fnUntraced(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
): Effect.fn.Return<ReadonlyArray<InstructionEventRow>> {
return yield* eventRows(db, sessionID, relevantEventTypes)
})
const instructionUpdatesAfter = Effect.fnUntraced(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
after: number,
) {
return yield* eventRows(db, sessionID, [instructionEventType], after)
})
const eventRows = Effect.fnUntraced(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
types: ReadonlyArray<string>,
after?: number,
): Effect.fn.Return<ReadonlyArray<InstructionEventRow>> {
const segments = (yield* lineage(db, sessionID)).filter(
(segment) => after === undefined || segment.through === undefined || segment.through > after,
)
return (yield* Effect.forEach(segments, (segment) =>
db
.select()
.from(EventTable)
.where(
and(
eq(EventTable.aggregate_id, segment.sessionID),
inArray(EventTable.type, types),
segment.through === undefined ? undefined : lte(EventTable.seq, segment.through),
after === undefined ? undefined : gt(EventTable.seq, after),
),
)
.orderBy(asc(EventTable.seq))
.all()
.pipe(Effect.orDie),
)).flat()
})
const lineage = Effect.fnUntraced(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
through?: number,
): Effect.fn.Return<ReadonlyArray<{ readonly sessionID: SessionSchema.ID; readonly through?: number }>> {
const session = yield* db
.select({ parentID: SessionTable.fork_session_id, forkSeq: SessionTable.fork_seq })
.from(SessionTable)
.where(eq(SessionTable.id, sessionID))
.get()
.pipe(Effect.orDie)
const inherited =
session?.parentID && session.forkSeq !== null
? yield* lineage(
db,
session.parentID,
through === undefined ? session.forkSeq : Math.min(session.forkSeq, through),
)
: []
return [...inherited, { sessionID, ...(through === undefined ? {} : { through }) }]
})
function fold(rows: ReadonlyArray<InstructionEventRow>) {
return rows.reduce<
| {
readonly epochStart: number
readonly throughSeq: number
readonly initial: Instructions.Values
readonly current: Instructions.Values
}
| undefined
>((state, row) => {
if (row.type === movedEventType || row.type === revertedEventType) return undefined
if (row.type === compactionEventType)
return state
? { epochStart: row.seq, throughSeq: row.seq, initial: state.current, current: state.current }
: undefined
if (row.type !== instructionEventType) return state
const delta = decodeInstructionsUpdated(row.data).delta
const current = Instructions.applyHashDelta(state?.current ?? {}, delta)
return state
? { ...state, throughSeq: row.seq, current }
: { epochStart: row.seq, throughSeq: row.seq, initial: current, current }
}, undefined)
}

11
packages/core/src/session/message-updater.ts
View file

@ -178,16 +178,7 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
"session.execution.succeeded": () => clearCurrentRetry,
"session.execution.failed": () => clearCurrentRetry,
"session.execution.interrupted": () => clearCurrentRetry,
"session.instructions.updated": (event) =>
adapter.appendMessage(
SessionMessage.System.make({
id: SessionMessage.ID.fromEvent(event.id),
type: "system",
text: event.data.text,
metadata: event.metadata,
time: { created: event.created },
}),
),
"session.instructions.updated": () => Effect.void,
"session.synthetic": (event) => {
return adapter.appendMessage(
SessionMessage.Synthetic.make({

46
packages/core/src/session/projector.ts
View file

@ -13,22 +13,18 @@ import { SessionMessage } from "./message"
import { SessionMessageUpdater } from "./message-updater"
import { SessionPending } from "./pending"
import { WorkspaceV2 } from "../workspace"
import { InstructionCheckpoint } from "./instruction-checkpoint"
import {
MessageTable,
PartTable,
InstructionCheckpointTable,
SessionPendingTable,
SessionMessageTable,
SessionTable,
} from "./sql"
import { InstructionState } from "./instruction-state"
import { MessageTable, PartTable, SessionPendingTable, SessionMessageTable, SessionTable } from "./sql"
import type { DeepMutable } from "../schema"
import { Slug } from "../util/slug"
import { Money } from "@opencode-ai/schema/money"
type DatabaseService = Database.Interface["db"]
type CurrentDurableEvent = Extract<SessionEvent.Event, { readonly durable: object }>
type MessageEvent = Exclude<CurrentDurableEvent, typeof SessionEvent.Forked.Type | typeof SessionEvent.Deleted.Type>
type MessageEvent = Exclude<
CurrentDurableEvent,
typeof SessionEvent.Forked.Type | typeof SessionEvent.Deleted.Type | typeof SessionEvent.InstructionsUpdated.Type
>
const decodeMessage = Schema.decodeUnknownSync(SessionMessage.Info)
const encodeMessage = Schema.encodeSync(SessionMessage.Info)
@ -212,6 +208,7 @@ const projectFork = Effect.fn("SessionProjector.projectFork")(function* (
parent_id: null,
fork_session_id: event.data.parentID,
fork_message_id: event.data.from,
fork_seq: event.data.parentSeq,
project_id: parent.project_id,
workspace_id: parent.workspace_id,
slug: Slug.create(),
@ -236,23 +233,6 @@ const projectFork = Effect.fn("SessionProjector.projectFork")(function* (
.pipe(Effect.orDie)
if (!stored) return yield* Effect.die(new SessionAlreadyProjected())
// The fork inherits the parent's transcript, so it inherits the context
// checkpoint that transcript was built against: copied message seqs keep
// folding at the same baseline horizon.
const checkpoint = yield* db
.select()
.from(InstructionCheckpointTable)
.where(eq(InstructionCheckpointTable.session_id, event.data.parentID))
.get()
.pipe(Effect.orDie)
if (checkpoint) {
yield* db
.insert(InstructionCheckpointTable)
.values({ ...checkpoint, session_id: event.data.sessionID })
.run()
.pipe(Effect.orDie)
}
let cursor = -1
while (copiedSeq !== undefined) {
const rows = yield* db
@ -334,7 +314,8 @@ const projectFork = Effect.fn("SessionProjector.projectFork")(function* (
cursor = rows.at(-1)!.seq
}
if (copiedSeq !== undefined) yield* EventV2.reserveSequence(db, event.data.sessionID, copiedSeq)
yield* EventV2.reserveSequence(db, event.data.sessionID, event.data.parentSeq)
yield* InstructionState.rebuild(db, event.data.sessionID)
})
function run(db: DatabaseService, event: MessageEvent) {
@ -522,7 +503,7 @@ const layer = Layer.effectDiscard(
.where(eq(SessionTable.id, event.data.sessionID))
.run()
.pipe(Effect.orDie)
yield* InstructionCheckpoint.reset(db, event.data.sessionID)
yield* InstructionState.reset(db, event.data.sessionID)
}),
)
yield* events.project(SessionV1.Event.Deleted, (event) =>
@ -688,7 +669,9 @@ const layer = Layer.effectDiscard(
yield* events.project(SessionEvent.Execution.Succeeded, (event) => run(db, event))
yield* events.project(SessionEvent.Execution.Failed, (event) => run(db, event))
yield* events.project(SessionEvent.Execution.Interrupted, (event) => run(db, event))
yield* events.project(SessionEvent.InstructionsUpdated, (event) => run(db, event))
yield* events.project(SessionEvent.InstructionsUpdated, (event) =>
InstructionState.apply(db, event.data.sessionID, event.durable.seq, event.data.delta),
)
yield* events.project(SessionEvent.Synthetic, (event) => run(db, event))
yield* events.project(SessionEvent.Skill.Activated, (event) => run(db, event))
yield* events.project(SessionEvent.Shell.Started, (event) => run(db, event))
@ -722,6 +705,7 @@ const layer = Layer.effectDiscard(
yield* events.project(SessionEvent.Compaction.Ended, (event) =>
Effect.gen(function* () {
yield* run(db, event)
yield* InstructionState.advanceEpoch(db, event.data.sessionID, event.durable.seq)
if (event.durable === undefined)
return yield* Effect.die(new Error("Durable Session event is missing aggregate sequence"))
if (event.data.reason === "manual")
@ -793,7 +777,7 @@ const layer = Layer.effectDiscard(
.where(eq(SessionTable.id, event.data.sessionID))
.run()
.pipe(Effect.orDie)
yield* InstructionCheckpoint.reset(db, event.data.sessionID)
yield* InstructionState.reset(db, event.data.sessionID)
}),
)
yield* events.subscribe([SessionEvent.Step.Ended, SessionEvent.Step.Failed]).pipe(

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

@ -29,7 +29,7 @@ import { InstructionEntry } from "../instruction-entry"
import { QuestionTool } from "../../tool/question"
import { ToolRegistry } from "../../tool/registry"
import { ToolOutputStore } from "../../tool-output-store"
import { InstructionCheckpoint } from "../instruction-checkpoint"
import { InstructionState } from "../instruction-state"
import { SessionCompaction } from "../compaction"
import { SessionEvent } from "../event"
import { SessionHistory } from "../history"
@ -112,6 +112,8 @@ const layer = Layer.effect(
if (!session) return yield* Effect.die(new Error(`Session not found: ${sessionID}`))
return session
})
const isCurrentLocation = (session: SessionSchema.Info) =>
session.location.directory === location.directory && session.location.workspaceID === location.workspaceID
const failInterruptedTools = Effect.fn("SessionRunner.failInterruptedTools")(function* (
sessionID: SessionSchema.ID,
@ -160,20 +162,15 @@ const layer = Layer.effect(
assistantMessageID?: SessionMessage.ID,
) {
const session = yield* getSession(sessionID)
if (session.location.directory !== location.directory || session.location.workspaceID !== location.workspaceID)
return yield* Effect.interrupt
if (!isCurrentLocation(session)) return yield* Effect.interrupt
yield* plugins.flush
const agent = yield* agents.select(session.agent)
const agentInfo = agent.info
if (!agentInfo) return yield* new AgentNotFoundError({ sessionID: session.id, agent: session.agent ?? agent.id })
// Establish what the model knows before admitting what the user said, so
// a blocked first step leaves pending inputs untouched.
const checkpoint = yield* InstructionCheckpoint.prepare(
db,
events,
loadInstructions(agent, session.id),
session.id,
)
const instructions = yield* loadInstructions(agent, session.id)
yield* InstructionState.prepare(db, events, instructions, session.id)
let currentStep = step
if (promotion) {
let promoted = 0
@ -187,8 +184,8 @@ const layer = Layer.effect(
const resolved = yield* models.resolve(session)
const model = resolved.model
const providerMetadataKey = model.route.providerMetadataKey ?? model.provider
const entries = yield* SessionHistory.entriesForRunner(db, session.id, checkpoint.baselineSeq)
const context = entries.map((entry) => entry.message)
const history = yield* SessionHistory.entriesForRunner(db, session.id, instructions)
const context = history.entries.map((entry) => entry.message)
const compactionInput = { sessionID: session.id, messages: context, model }
if (compaction.required(compactionInput) && !(yield* SessionPending.compaction(db, session.id))) {
const compacted = yield* compaction.compact(compactionInput)
@ -201,7 +198,7 @@ const layer = Layer.effect(
const request = LLM.request({
model,
providerOptions: { openai: { promptCacheKey } },
system: [agentInfo.system ? agentInfo.system : SessionRunnerSystemPrompt.provider(model), checkpoint.baseline]
system: [agentInfo.system ? agentInfo.system : SessionRunnerSystemPrompt.provider(model), history.initial]
.filter((part): part is string => part !== undefined && part.length > 0)
.map(SystemPart.make),
messages: [

20
packages/core/src/session/sql.ts
View file

@ -11,7 +11,7 @@ import type { SessionSchema } from "./schema"
import type { MessageID, PartID, SessionV1 } from "../v1/session"
import { WorkspaceV2 } from "../workspace"
import { Timestamps } from "../database/schema.sql"
import type { Instructions } from "../instructions/index"
import type { Instruction } from "@opencode-ai/schema/instruction"
import type { Session } from "@opencode-ai/schema/session"
import type { SyntheticData, UserData } from "@opencode-ai/schema/session-pending"
import type { RevertV1 } from "@opencode-ai/schema/session-revert"
@ -33,6 +33,7 @@ export const SessionTable = sqliteTable(
parent_id: text().$type<SessionSchema.ID>(),
fork_session_id: text().$type<SessionSchema.ID>(),
fork_message_id: text().$type<SessionMessage.ID>(),
fork_seq: integer(),
slug: text().notNull(),
directory: directoryColumn().notNull(),
path: pathColumn(),
@ -159,18 +160,25 @@ export const InstructionEntryTable = sqliteTable(
.notNull()
.references(() => SessionTable.id, { onDelete: "cascade" }),
key: text().notNull(),
value: text({ mode: "json" }).notNull().$type<Schema.Json>(),
value: text({ mode: "json" }).$type<Schema.Json>(),
removed: integer({ mode: "boolean" }).notNull().default(false),
...Timestamps,
},
(table) => [primaryKey({ columns: [table.session_id, table.key] })],
)
export const InstructionCheckpointTable = sqliteTable("instruction_checkpoint", {
export const InstructionBlobTable = sqliteTable("instruction_blob", {
hash: text().$type<Instruction.Hash>().primaryKey(),
value: text({ mode: "json" }).$type<Schema.Json>(),
})
export const InstructionStateTable = sqliteTable("instruction_state", {
session_id: text()
.$type<SessionSchema.ID>()
.primaryKey()
.references(() => SessionTable.id, { onDelete: "cascade" }),
baseline: text().notNull(),
snapshot: text({ mode: "json" }).notNull().$type<Instructions.Applied>(),
baseline_seq: integer().notNull(),
epoch_start: integer().notNull(),
through_seq: integer().notNull(),
initial_values: text({ mode: "json" }).notNull().$type<Instruction.Values>(),
current_values: text({ mode: "json" }).notNull().$type<Instruction.Values>(),
})