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

fix(core): finalize v2 session context epochs

Browse source
This commit is contained in:
Kit Langton 2026-06-04 21:17:05 -04:00
commit cd812e2045
33 changed files with 1245 additions and 791 deletions
Download patch file Download diff file Expand all files Collapse all files

2
packages/core/src/database/migration.gen.ts generated
View file

@ -32,6 +32,6 @@ export const migrations = (
import("./migration/20260603141458_session_input_inbox"),
import("./migration/20260603160727_jittery_ezekiel_stane"),
import("./migration/20260604172448_event_sourced_session_input"),
import("./migration/20260604234609_add_session_context_snapshot"),
import("./migration/20260605003541_add_session_context_snapshot"),
])
).map((module) => module.default) satisfies DatabaseMigration.Migration[]

2
packages/core/src/database/migration/20260604234609_add_session_context_snapshot.ts → packages/core/src/database/migration/20260605003541_add_session_context_snapshot.ts
View file

@ -2,7 +2,7 @@ import { Effect } from "effect"
import type { DatabaseMigration } from "../migration"
export default {
id: "20260604234609_add_session_context_snapshot",
id: "20260605003541_add_session_context_snapshot",
up(tx) {
return Effect.gen(function* () {
yield* tx.run(`

37
packages/core/src/event.ts
View file

@ -155,7 +155,6 @@ export interface Interface {
readonly aggregateID: string
readonly after?: Cursor
}) => Stream.Stream<CursorEvent>
readonly sequence: (aggregateID: string) => Effect.Effect<number>
readonly sync: (handler: Sync) => Effect.Effect<Unsubscribe>
readonly listen: (listener: Listener) => Effect.Effect<Unsubscribe>
readonly beforeCommit: (guard: CommitGuard) => Effect.Effect<void>
@ -337,9 +336,6 @@ export const layerWith = (options?: LayerOptions) =>
yield* projector({ ...event, seq } as Payload)
}
if (commit) yield* commit(seq)
const encoded = syncRegistry
.get(versionedType(definition.type, sync.version))!
.encode(event.data)
yield* db
.insert(EventSequenceTable)
.values([{ aggregate_id: aggregateID, seq, owner_id: input?.ownerID }])
@ -390,7 +386,10 @@ export const layerWith = (options?: LayerOptions) =>
const durable = registry.get(event.type)?.sync !== undefined
if (!durable && options?.commit)
return yield* Effect.die(
new InvalidSyncEventError({ type: event.type, message: "Local commit hooks require a synchronized event" }),
new InvalidSyncEventError({
type: event.type,
message: "Local commit hooks require a synchronized event",
}),
)
if (durable) {
const committed = yield* commitSyncEvent(event as Payload, undefined, options?.commit)
@ -438,14 +437,17 @@ export const layerWith = (options?: LayerOptions) =>
(serviceLocation
? { directory: serviceLocation.directory, workspaceID: serviceLocation.workspaceID }
: undefined)
return yield* publishEvent({
id: options?.id ?? ID.create(),
...(options?.metadata ? { metadata: options.metadata } : {}),
type: definition.type,
...(definition.sync === undefined ? {} : { version: definition.sync.version }),
...(location ? { location } : {}),
data,
} as Payload<D>, options)
return yield* publishEvent(
{
id: options?.id ?? ID.create(),
...(options?.metadata ? { metadata: options.metadata } : {}),
type: definition.type,
...(definition.sync === undefined ? {} : { version: definition.sync.version }),
...(location ? { location } : {}),
data,
} as Payload<D>,
options,
)
})
}
@ -534,14 +536,6 @@ export const layerWith = (options?: LayerOptions) =>
.pipe(Effect.orDie)
}
const sequence = (aggregateID: string) =>
db
.select({ seq: EventSequenceTable.seq })
.from(EventSequenceTable)
.where(eq(EventSequenceTable.aggregate_id, aggregateID))
.get()
.pipe(Effect.orDie, Effect.map((row) => row?.seq ?? -1))
const subscribe = <D extends Definition>(definition: D): Stream.Stream<Payload<D>> =>
Stream.unwrap(getOrCreate(definition).pipe(Effect.map((pubsub) => Stream.fromPubSub(pubsub)))).pipe(
Stream.map((event) => event as Payload<D>),
@ -669,7 +663,6 @@ export const layerWith = (options?: LayerOptions) =>
subscribe,
all: streamAll,
aggregateEvents: streamEvents,
sequence,
sync,
listen,
beforeCommit,

27
packages/core/src/instruction-context.ts
View file

@ -15,6 +15,7 @@ class File extends Schema.Class<File>("InstructionContext.File")({
}) {}
const Files = Schema.Array(File)
const key = SystemContext.Key.make("core/instructions")
export const layer = Layer.effectDiscard(
Effect.gen(function* () {
@ -25,7 +26,7 @@ export const layer = Layer.effectDiscard(
const source = (value: ReadonlyArray<File> | SystemContext.Unavailable) =>
SystemContext.make({
key: SystemContext.Key.make("core/instructions"),
key,
codec: Schema.toCodecJson(Files),
load: Effect.succeed(value),
baseline: render,
@ -43,29 +44,37 @@ export const layer = Layer.effectDiscard(
const files = yield* Effect.forEach(
paths,
(path) =>
fs.readFileStringSafe(path).pipe(
Effect.map((content) => (content === undefined ? undefined : new File({ path: AbsolutePath.make(path), content }))),
),
fs
.readFileStringSafe(path)
.pipe(
Effect.map((content) =>
content === undefined ? undefined : new File({ path: AbsolutePath.make(path), content }),
),
),
{ concurrency: "unbounded" },
)
if (files.some((file, index) => file === undefined && discovered.has(paths[index]))) return SystemContext.unavailable
if (files.some((file, index) => file === undefined && discovered.has(paths[index])))
return SystemContext.unavailable
return files.filter((file): file is File => file !== undefined)
})
yield* registry.contribute({
key: "core/instructions",
key,
load: observe().pipe(
Effect.map((files) =>
files === SystemContext.unavailable ? source(files) : files.length === 0 ? SystemContext.empty : source(files),
files === SystemContext.unavailable
? source(files)
: files.length === 0
? SystemContext.empty
: source(files),
),
Effect.catch(() => Effect.succeed(source(SystemContext.unavailable))),
Effect.catchDefect(() => Effect.succeed(source(SystemContext.unavailable))),
),
})
}),
)
export const locationLayer = layer
function render(files: ReadonlyArray<File>) {
return files.map((file) => `Instructions from: ${file.path}\n${file.content}`).join("\n\n")
}

127
packages/core/src/session/context-epoch.ts
View file

@ -7,13 +7,51 @@ import { EventV2 } from "../event"
import { SystemContext } from "../system-context"
import { SystemContextRegistry } from "../system-context-registry"
import { SessionEvent } from "./event"
import { SessionInput } from "./input"
import { SessionMessageID } from "./message-id"
import { SessionSchema } from "./schema"
import { SessionContextEpochTable } from "./sql"
type DatabaseService = Database.Interface["db"]
export const prepare = Effect.fn("SessionContextEpoch.prepare")(function* (
class RevisionMismatch extends Error {}
const retryRevisionMismatch = <A, E>(attempt: () => Effect.Effect<A, E>): Effect.Effect<A, E> =>
attempt().pipe(
Effect.catchDefect((defect) =>
defect instanceof RevisionMismatch
? Effect.yieldNow.pipe(Effect.andThen(retryRevisionMismatch(attempt)))
: Effect.die(defect),
),
)
interface Prepared {
readonly baseline: string
readonly baselineSeq: number
}
export function initialize(
db: DatabaseService,
context: SystemContextRegistry.Interface,
sessionID: SessionSchema.ID,
): Effect.Effect<Prepared | undefined, SystemContext.InitializationBlocked> {
return retryRevisionMismatch(() => initializeOnce(db, context, sessionID)).pipe(
Effect.withSpan("SessionContextEpoch.initialize"),
)
}
export function prepare(
db: DatabaseService,
events: EventV2.Interface,
context: SystemContextRegistry.Interface,
sessionID: SessionSchema.ID,
): Effect.Effect<Prepared, SystemContext.InitializationBlocked> {
return retryRevisionMismatch(() => prepareOnce(db, events, context, sessionID)).pipe(
Effect.withSpan("SessionContextEpoch.prepare"),
)
}
const prepareOnce = Effect.fnUntraced(function* (
db: DatabaseService,
events: EventV2.Interface,
context: SystemContextRegistry.Interface,
@ -22,17 +60,19 @@ export const prepare = Effect.fn("SessionContextEpoch.prepare")(function* (
const [value, stored] = yield* Effect.all([context.load(), find(db, sessionID)], { concurrency: "unbounded" })
if (!stored) {
const generation = yield* SystemContext.initialize(value)
const baselineSeq = yield* initialize(db, events, sessionID, generation)
const baselineSeq = yield* insert(db, sessionID, generation)
return { baseline: generation.baseline, baselineSeq }
}
const snapshot = yield* Schema.decodeUnknownEffect(SystemContext.Snapshot)(stored.snapshot).pipe(Effect.orDie)
const result =
stored.replacement_seq === null ? yield* SystemContext.reconcile(value, snapshot) : yield* SystemContext.replace(value, snapshot)
stored.replacement_seq === null
? yield* SystemContext.reconcile(value, snapshot)
: yield* SystemContext.replace(value, snapshot)
if (result._tag === "Unchanged" || result._tag === "ReplacementBlocked")
return { baseline: stored.baseline, baselineSeq: stored.baseline_seq }
if (result._tag === "Replaced") {
const replacementSeq = stored.replacement_seq ?? (yield* events.sequence(sessionID))
if (result._tag === "ReplacementReady") {
const replacementSeq = stored.replacement_seq ?? (yield* SessionInput.latestSeq(db, sessionID))
yield* replace(db, sessionID, stored.revision, replacementSeq, result.generation)
return { baseline: result.generation.baseline, baselineSeq: replacementSeq }
}
@ -45,6 +85,28 @@ export const prepare = Effect.fn("SessionContextEpoch.prepare")(function* (
return { baseline: stored.baseline, baselineSeq: stored.baseline_seq }
})
const initializeOnce = Effect.fnUntraced(function* (
db: DatabaseService,
context: SystemContextRegistry.Interface,
sessionID: SessionSchema.ID,
) {
if (yield* exists(db, sessionID)) return
const generation = yield* context.load().pipe(Effect.flatMap(SystemContext.initialize))
const baselineSeq = yield* insert(db, sessionID, generation)
return { baseline: generation.baseline, baselineSeq }
})
const exists = Effect.fn("SessionContextEpoch.exists")(function* (db: DatabaseService, sessionID: SessionSchema.ID) {
return (
(yield* db
.select({ sessionID: SessionContextEpochTable.session_id })
.from(SessionContextEpochTable)
.where(eq(SessionContextEpochTable.session_id, sessionID))
.get()
.pipe(Effect.orDie)) !== undefined
)
})
const find = Effect.fn("SessionContextEpoch.find")(function* (db: DatabaseService, sessionID: SessionSchema.ID) {
return yield* db
.select()
@ -73,9 +135,8 @@ export const requestReplacement = Effect.fn("SessionContextEpoch.requestReplacem
.pipe(Effect.orDie)
})
const initialize = Effect.fnUntraced(function* (
const insert = Effect.fnUntraced(function* (
db: DatabaseService,
events: EventV2.Interface,
sessionID: SessionSchema.ID,
generation: SystemContext.Generation,
) {
@ -83,7 +144,7 @@ const initialize = Effect.fnUntraced(function* (
.transaction(
() =>
Effect.gen(function* () {
const baselineSeq = yield* events.sequence(sessionID)
const baselineSeq = yield* SessionInput.latestSeq(db, sessionID)
yield* db
.insert(SessionContextEpochTable)
.values({
@ -93,8 +154,13 @@ const initialize = Effect.fnUntraced(function* (
baseline_seq: baselineSeq,
revision: 0,
})
.run()
.pipe(Effect.orDie)
.onConflictDoNothing()
.returning({ sessionID: SessionContextEpochTable.session_id })
.get()
.pipe(
Effect.orDie,
Effect.flatMap((inserted) => (inserted ? Effect.void : Effect.die(new RevisionMismatch()))),
)
return baselineSeq
}),
{ behavior: "immediate" },
@ -109,33 +175,22 @@ const replace = Effect.fnUntraced(function* (
baselineSeq: number,
generation: SystemContext.Generation,
) {
yield* db
.transaction(
() =>
Effect.gen(function* () {
const updated = yield* db
.update(SessionContextEpochTable)
.set({
baseline: generation.baseline,
snapshot: generation.snapshot,
baseline_seq: baselineSeq,
replacement_seq: null,
revision: expectedRevision + 1,
})
.where(
and(
eq(SessionContextEpochTable.session_id, sessionID),
eq(SessionContextEpochTable.revision, expectedRevision),
),
)
.returning({ revision: SessionContextEpochTable.revision })
.get()
.pipe(Effect.orDie)
if (!updated) return yield* Effect.die("Session context epoch revision mismatch")
}),
{ behavior: "immediate" },
const updated = yield* db
.update(SessionContextEpochTable)
.set({
baseline: generation.baseline,
snapshot: generation.snapshot,
baseline_seq: baselineSeq,
replacement_seq: null,
revision: expectedRevision + 1,
})
.where(
and(eq(SessionContextEpochTable.session_id, sessionID), eq(SessionContextEpochTable.revision, expectedRevision)),
)
.returning({ revision: SessionContextEpochTable.revision })
.get()
.pipe(Effect.orDie)
if (!updated) return yield* Effect.die(new RevisionMismatch())
})
const advance = Effect.fnUntraced(function* (
@ -157,5 +212,5 @@ const advance = Effect.fnUntraced(function* (
.returning({ revision: SessionContextEpochTable.revision })
.get()
.pipe(Effect.orDie)
if (!updated) return yield* Effect.die("Session context epoch revision mismatch")
if (!updated) return yield* Effect.die(new RevisionMismatch())
})

10
packages/core/src/session/context.ts
View file

@ -75,10 +75,7 @@ export const load = Effect.fn("SessionContext.load")(function* (db: DatabaseServ
],
{ concurrency: "unbounded" },
)
return yield* Effect.forEach(
yield* messageRows(db, sessionID, compaction, epoch?.baselineSeq),
decodeMessageRow,
)
return yield* Effect.forEach(yield* messageRows(db, sessionID, compaction, epoch?.baselineSeq), decodeMessageRow)
})
export const loadForRunner = Effect.fn("SessionContext.loadForRunner")(function* (
@ -86,7 +83,10 @@ export const loadForRunner = Effect.fn("SessionContext.loadForRunner")(function*
sessionID: SessionSchema.ID,
baselineSeq: number,
) {
return yield* Effect.forEach(yield* messageRows(db, sessionID, yield* latestCompaction(db, sessionID), baselineSeq), decodeMessageRow)
return yield* Effect.forEach(
yield* messageRows(db, sessionID, yield* latestCompaction(db, sessionID), baselineSeq),
decodeMessageRow,
)
})
export * as SessionContext from "./context"

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

@ -176,7 +176,16 @@ export class Compaction extends Schema.Class<Compaction>("Session.Message.Compac
...Base,
}) {}
export const Message = Schema.Union([AgentSwitched, ModelSwitched, User, Synthetic, System, Shell, Assistant, Compaction])
export const Message = Schema.Union([
AgentSwitched,
ModelSwitched,
User,
Synthetic,
System,
Shell,
Assistant,
Compaction,
])
.pipe(Schema.toTaggedUnion("type"))
.annotate({ identifier: "Session.Message" })

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

@ -422,7 +422,9 @@ export const layer = Layer.effectDiscard(
)
yield* events.project(SessionEvent.ContextUpdated, (event) => {
if (!event.replay || event.seq === undefined) return run(db, event)
return run(db, event).pipe(Effect.andThen(SessionContextEpoch.requestReplacement(db, event.data.sessionID, event.seq)))
return run(db, event).pipe(
Effect.andThen(SessionContextEpoch.requestReplacement(db, event.data.sessionID, event.seq)),
)
})
yield* events.project(SessionEvent.Synthetic, (event) => run(db, event))
yield* events.project(SessionEvent.Shell.Started, (event) => run(db, event))

8
packages/core/src/session/runner/index.ts
View file

@ -5,6 +5,7 @@ import { Context, Effect, Schema } from "effect"
import { SessionSchema } from "../schema"
import type { MessageDecodeError } from "../error"
import { SessionRunnerModel } from "./model"
import type { SystemContext } from "../../system-context"
export class StepLimitExceededError extends Schema.TaggedErrorClass<StepLimitExceededError>()(
"SessionRunner.StepLimitExceededError",
@ -14,7 +15,12 @@ export class StepLimitExceededError extends Schema.TaggedErrorClass<StepLimitExc
},
) {}
export type RunError = LLMError | SessionRunnerModel.Error | MessageDecodeError | StepLimitExceededError
export type RunError =
| LLMError
| SessionRunnerModel.Error
| MessageDecodeError
| StepLimitExceededError
| SystemContext.InitializationBlocked
/** Runs one local continuation from already-recorded Session history. */
export interface Interface {

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

@ -139,6 +139,7 @@ export const layer = Layer.effect(
promotion: "steer" | "queue" | undefined,
) {
const session = yield* getSession(sessionID)
const initialized = yield* SessionContextEpoch.initialize(db, systemContext, session.id)
const model = yield* models.resolve(session)
const toolFibers = yield* FiberSet.make<void, never>()
let needsContinuation = false
@ -150,7 +151,7 @@ export const layer = Layer.effect(
yield* SessionInput.promoteSteers(db, events, session.id, cutoff)
}
}
const system = yield* SessionContextEpoch.prepare(db, events, systemContext, session.id)
const system = initialized ?? (yield* SessionContextEpoch.prepare(db, events, systemContext, session.id))
const context = yield* getRunnerContext(session.id, system.baselineSeq)
const request = LLM.request({
model,

2
packages/core/src/system-context-builtins.ts
View file

@ -36,7 +36,7 @@ const builtIns = Layer.effectDiscard(
}),
])
yield* registry.contribute({ key: "core/builtins", load: Effect.succeed(context) })
yield* registry.contribute({ key: SystemContext.Key.make("core/builtins"), load: Effect.succeed(context) })
}),
)

6
packages/core/src/system-context-registry.ts
View file

@ -4,7 +4,7 @@ import { Context, Effect, Layer, Ref, Scope } from "effect"
import { SystemContext } from "./system-context"
export interface Contribution {
readonly key: string
readonly key: SystemContext.Key
readonly load: Effect.Effect<SystemContext.SystemContext>
}
@ -36,7 +36,7 @@ export const layer = Layer.effect(
)
}),
load: Effect.fn("SystemContextRegistry.load")(function* () {
const current = (yield* Ref.get(contributions)).toSorted((a, b) => a.key.localeCompare(b.key))
const current = (yield* Ref.get(contributions)).toSorted((a, b) => (a.key < b.key ? -1 : a.key > b.key ? 1 : 0))
return SystemContext.combine(
yield* Effect.forEach(current, (contribution) => contribution.load, { concurrency: "unbounded" }),
)
@ -44,5 +44,3 @@ export const layer = Layer.effect(
})
}),
)
export const locationLayer = layer

23
packages/core/src/system-context.ts
View file

@ -67,8 +67,8 @@ export interface Updated {
readonly snapshot: Snapshot
}
export interface Replaced {
readonly _tag: "Replaced"
export interface ReplacementReady {
readonly _tag: "ReplacementReady"
readonly generation: Generation
}
@ -76,9 +76,14 @@ export interface ReplacementBlocked {
readonly _tag: "ReplacementBlocked"
}
export type ReplacementResult = Replaced | ReplacementBlocked
export type ReplacementResult = ReplacementReady | ReplacementBlocked
export type ReconcileResult = { readonly _tag: "Unchanged" } | Updated | ReplacementResult
export class InitializationBlocked extends Schema.TaggedErrorClass<InitializationBlocked>()(
"SystemContext.InitializationBlocked",
{ keys: Schema.Array(Key) },
) {}
export class DuplicateKeyError extends Schema.TaggedErrorClass<DuplicateKeyError>()("SystemContext.DuplicateKeyError", {
key: Key,
}) {
@ -186,8 +191,14 @@ const observe = (value: SystemContext) =>
)
/** Creates the immutable baseline and durable snapshot for a new generation. */
export function initialize(value: SystemContext): Effect.Effect<Generation> {
return observe(value).pipe(Effect.map(initializeObservation))
export function initialize(value: SystemContext): Effect.Effect<Generation, InitializationBlocked> {
return observe(value).pipe(
Effect.flatMap((entries) => {
const unavailable = entries.flatMap((entry) => (entry._tag === "Unavailable" ? [entry.key] : []))
if (unavailable.length > 0) return new InitializationBlocked({ keys: unavailable })
return Effect.succeed(initializeObservation(entries))
}),
)
}
function initializeObservation(entries: ReadonlyArray<Entry>): Generation {
@ -272,7 +283,7 @@ export function replace(value: SystemContext, previous: Snapshot): Effect.Effect
function replaceObservation(entries: ReadonlyArray<Entry>, previous: Snapshot): ReplacementResult {
if (entries.some((entry) => entry._tag === "Unavailable" && getSnapshot(previous, entry.key) !== undefined))
return { _tag: "ReplacementBlocked" }
return { _tag: "Replaced", generation: initializeObservation(entries) }
return { _tag: "ReplacementReady", generation: initializeObservation(entries) }
}
function context(sources: ReadonlyArray<PackedSource>): SystemContext {