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

refactor(core): simplify v2 system context epochs

Browse source
This commit is contained in:
Kit Langton 2026-06-04 19:49:19 -04:00
commit 00c4114911
30 changed files with 997 additions and 828 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/20260604184448_add_session_context_epoch"),
import("./migration/20260604234609_add_session_context_snapshot"),
])
).map((module) => module.default) satisfies DatabaseMigration.Migration[]

13
packages/core/src/database/migration/20260604184448_add_session_context_epoch.ts → packages/core/src/database/migration/20260604234609_add_session_context_snapshot.ts
View file

@ -2,29 +2,20 @@ import { Effect } from "effect"
import type { DatabaseMigration } from "../migration"
export default {
id: "20260604184448_add_session_context_epoch",
id: "20260604234609_add_session_context_snapshot",
up(tx) {
return Effect.gen(function* () {
yield* tx.run(`
CREATE TABLE \`session_context_epoch\` (
\`session_id\` text PRIMARY KEY,
\`baseline\` text NOT NULL,
\`checkpoint\` text NOT NULL,
\`snapshot\` text NOT NULL,
\`baseline_seq\` integer NOT NULL,
\`replacement_seq\` integer,
\`revision\` integer DEFAULT 0 NOT NULL,
CONSTRAINT \`fk_session_context_epoch_session_id_session_id_fk\` FOREIGN KEY (\`session_id\`) REFERENCES \`session\`(\`id\`) ON DELETE CASCADE
);
`)
yield* tx.run(`
CREATE TABLE \`session_context_message\` (
\`session_id\` text NOT NULL,
\`seq\` integer NOT NULL,
\`parts\` text NOT NULL,
CONSTRAINT \`session_context_message_pk\` PRIMARY KEY(\`session_id\`, \`seq\`),
CONSTRAINT \`fk_session_context_message_session_id_session_id_fk\` FOREIGN KEY (\`session_id\`) REFERENCES \`session\`(\`id\`) ON DELETE CASCADE
);
`)
})
},
} satisfies DatabaseMigration.Migration

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

@ -45,6 +45,8 @@ export type Payload<D extends Definition = Definition> = {
readonly version?: number
readonly location?: Location.Ref
readonly metadata?: Record<string, unknown>
/** Internal replay marker for projectors that own non-replicated operational state. */
readonly replay?: boolean
}
export type Projector<D extends Definition = Definition> = (event: Payload<D>) => Effect.Effect<void>
@ -137,6 +139,8 @@ export interface PublishOptions {
readonly id?: ID
readonly metadata?: Record<string, unknown>
readonly location?: Location.Ref
/** Local operational projection committed atomically with a new synchronized event. Not replayed or serialized. */
readonly commit?: (seq: number) => Effect.Effect<void>
}
export interface Interface {
@ -151,6 +155,7 @@ 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>
@ -215,6 +220,7 @@ export const layerWith = (options?: LayerOptions) =>
readonly ownerID?: string
readonly strictOwner?: boolean
},
commit?: (seq: number) => Effect.Effect<void>,
) {
return Effect.gen(function* () {
const definition = registry.get(event.type)
@ -330,6 +336,10 @@ export const layerWith = (options?: LayerOptions) =>
for (const projector of list) {
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 }])
@ -375,11 +385,15 @@ export const layerWith = (options?: LayerOptions) =>
})
}
function publishEvent<D extends Definition>(event: Payload<D>) {
function publishEvent<D extends Definition>(event: Payload<D>, options?: PublishOptions) {
return Effect.gen(function* () {
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" }),
)
if (durable) {
const committed = yield* commitSyncEvent(event as Payload)
const committed = yield* commitSyncEvent(event as Payload, undefined, options?.commit)
if (committed) {
event = { ...event, seq: committed.seq }
yield* Effect.forEach(syncHandlers, (sync) => observe(event as Payload, "sync", sync), { discard: true })
@ -431,7 +445,7 @@ export const layerWith = (options?: LayerOptions) =>
...(definition.sync === undefined ? {} : { version: definition.sync.version }),
...(location ? { location } : {}),
data,
} as Payload<D>)
} as Payload<D>, options)
})
}
@ -451,6 +465,7 @@ export const layerWith = (options?: LayerOptions) =>
type: definition.type,
version: definition.sync.version,
data: definition.decode(event.data),
replay: true,
} as Payload
const committed = yield* commitSyncEvent(payload, {
seq: event.seq,
@ -519,6 +534,14 @@ 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>),
@ -646,6 +669,7 @@ export const layerWith = (options?: LayerOptions) =>
subscribe,
all: streamAll,
aggregateEvents: streamEvents,
sequence,
sync,
listen,
beforeCommit,

39
packages/core/src/session-system-context.ts
View file

@ -1,11 +1,11 @@
export * as SessionSystemContext from "./session-system-context"
import { Context, DateTime, Effect, Layer } from "effect"
import { Context, DateTime, Effect, Layer, Schema } from "effect"
import { Location } from "./location"
import { SystemContext } from "./system-context"
export interface Interface {
readonly load: () => Effect.Effect<SystemContext.Snapshot>
readonly load: () => Effect.Effect<SystemContext.SystemContext>
}
export class Service extends Context.Service<Service, Interface>()("@opencode/v2/SessionSystemContext") {}
@ -22,32 +22,25 @@ export const layer = Layer.effect(
` Platform: ${process.platform}`,
"</env>",
].join("\n")
const context = SystemContext.struct({
environment: SystemContext.value({
const context = SystemContext.combine([
SystemContext.make({
key: SystemContext.Key.make("core/environment"),
load: Effect.succeed({
baseline: ["Here is some useful information about the environment you are running in:", environment].join(
"\n",
),
update: ["The environment you are running in is now:", environment].join("\n"),
}),
codec: Schema.toCodecJson(Schema.String),
load: Effect.succeed(environment),
baseline: (environment) =>
["Here is some useful information about the environment you are running in:", environment].join("\n"),
update: (_previous, environment) => ["The environment you are running in is now:", environment].join("\n"),
}),
date: SystemContext.value({
SystemContext.make({
key: SystemContext.Key.make("core/date"),
load: DateTime.nowAsDate.pipe(
Effect.map((date) => ({
baseline: `Today's date: ${date.toDateString()}`,
update: `Today's date is now: ${date.toDateString()}`,
})),
),
codec: Schema.toCodecJson(Schema.String),
load: DateTime.nowAsDate.pipe(Effect.map((date) => date.toDateString())),
baseline: (date) => `Today's date: ${date}`,
update: (_previous, date) => `Today's date is now: ${date}`,
}),
})
])
return Service.of({
load: Effect.fn("SessionSystemContext.load")(function* () {
return yield* SystemContext.load(context)
}),
})
return Service.of({ load: () => Effect.succeed(context) })
}),
)

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

@ -1,18 +1,17 @@
export * as SessionContextEpoch from "./context-epoch"
import { and, eq, isNull, lt, sql } from "drizzle-orm"
import { and, eq, isNull, lt, or, sql } from "drizzle-orm"
import { DateTime, Effect, Schema } from "effect"
import type { Database } from "../database/database"
import { EventV2 } from "../event"
import { SessionSystemContext } from "../session-system-context"
import { SystemContext } from "../system-context"
import { SessionEvent } from "./event"
import { SessionMessageID } from "./message-id"
import { SessionSchema } from "./schema"
import { SessionContextEpochTable, SessionContextMessageTable } from "./sql"
import { SessionContextEpochTable } from "./sql"
type DatabaseService = Database.Interface["db"]
const sameBaseline = Schema.toEquivalence(SystemContext.PartsSchema)
const sameCheckpoint = Schema.toEquivalence(SystemContext.CheckpointSchema)
export const prepare = Effect.fn("SessionContextEpoch.prepare")(function* (
db: DatabaseService,
@ -20,48 +19,33 @@ export const prepare = Effect.fn("SessionContextEpoch.prepare")(function* (
context: SessionSystemContext.Interface,
sessionID: SessionSchema.ID,
) {
const snapshot = yield* context.load()
const stored = yield* find(db, sessionID)
const [value, stored] = yield* Effect.all([context.load(), find(db, sessionID)], { concurrency: "unbounded" })
if (!stored) {
const initialized = SystemContext.initialize(snapshot)
const event = yield* events.publish(SessionEvent.ContextInitialized, {
sessionID,
timestamp: yield* DateTime.now,
baseline: initialized.baseline,
checkpoint: initialized.checkpoint,
})
if (event.seq === undefined) return yield* Effect.die("Synchronized Session event is missing aggregate sequence")
return { baseline: initialized.baseline, baselineSeq: event.seq }
}
if (stored.replacement_seq !== null) {
if (SystemContext.replacementBlocked(snapshot, stored.checkpoint))
return { baseline: stored.baseline, baselineSeq: stored.baseline_seq }
const initialized = SystemContext.initialize(snapshot)
const event = yield* events.publish(SessionEvent.ContextReplaced, {
sessionID,
timestamp: yield* DateTime.now,
expectedRevision: stored.revision,
baseline: initialized.baseline,
checkpoint: initialized.checkpoint,
})
if (event.seq === undefined) return yield* Effect.die("Synchronized Session event is missing aggregate sequence")
return { baseline: initialized.baseline, baselineSeq: event.seq }
const generation = yield* SystemContext.initialize(value)
const baselineSeq = yield* initialize(db, events, sessionID, generation)
return { baseline: generation.baseline, baselineSeq }
}
const refreshed = SystemContext.refresh(snapshot, stored.checkpoint)
if (sameCheckpoint(refreshed.checkpoint, stored.checkpoint))
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)
if (result._tag === "Unchanged" || result._tag === "ReplacementBlocked")
return { baseline: stored.baseline, baselineSeq: stored.baseline_seq }
yield* events.publish(SessionEvent.ContextUpdated, {
sessionID,
timestamp: yield* DateTime.now,
expectedRevision: stored.revision,
parts: refreshed.changes,
checkpoint: refreshed.checkpoint,
})
if (result._tag === "Replaced") {
const replacementSeq = stored.replacement_seq ?? (yield* events.sequence(sessionID))
yield* replace(db, sessionID, stored.revision, replacementSeq, result.generation)
return { baseline: result.generation.baseline, baselineSeq: replacementSeq }
}
yield* events.publish(
SessionEvent.ContextUpdated,
{ sessionID, messageID: SessionMessageID.ID.create(), timestamp: yield* DateTime.now, text: result.text },
{ commit: () => advance(db, sessionID, stored.revision, result.snapshot).pipe(Effect.orDie) },
)
return { baseline: stored.baseline, baselineSeq: stored.baseline_seq }
})
export const find = Effect.fn("SessionContextEpoch.find")(function* (db: DatabaseService, sessionID: SessionSchema.ID) {
const find = Effect.fn("SessionContextEpoch.find")(function* (db: DatabaseService, sessionID: SessionSchema.ID) {
return yield* db
.select()
.from(SessionContextEpochTable)
@ -70,110 +54,6 @@ export const find = Effect.fn("SessionContextEpoch.find")(function* (db: Databas
.pipe(Effect.orDie)
})
export const projectInitialized = Effect.fn("SessionContextEpoch.projectInitialized")(function* (
db: DatabaseService,
event: SessionEvent.ContextInitialized,
seq: number,
) {
const stored = yield* find(db, event.data.sessionID)
if (stored) {
if (stored.baseline_seq > seq) return yield* Effect.void
if (stored.baseline_seq !== seq || !sameBaseline(stored.baseline, event.data.baseline))
return yield* Effect.die("Session context epoch initialization conflicts with stored baseline")
return yield* Effect.void
}
return yield* db
.insert(SessionContextEpochTable)
.values({
session_id: event.data.sessionID,
baseline: event.data.baseline,
checkpoint: event.data.checkpoint,
baseline_seq: seq,
replacement_seq: null,
revision: 0,
})
.run()
.pipe(Effect.orDie)
})
export const projectUpdated = Effect.fn("SessionContextEpoch.projectUpdated")(function* (
db: DatabaseService,
event: SessionEvent.ContextUpdated,
seq: number,
) {
const stored = yield* find(db, event.data.sessionID)
if (!stored) return yield* Effect.die("Session context epoch is not initialized")
if (stored.baseline_seq > seq) return yield* Effect.void
if (stored.replacement_seq !== null && seq >= stored.replacement_seq)
return yield* Effect.die("Session context epoch replacement is pending")
if (stored.revision > event.data.expectedRevision) {
if (event.data.parts.length === 0) return yield* Effect.void
const projected = yield* db
.select({ parts: SessionContextMessageTable.parts })
.from(SessionContextMessageTable)
.where(
and(eq(SessionContextMessageTable.session_id, event.data.sessionID), eq(SessionContextMessageTable.seq, seq)),
)
.get()
.pipe(Effect.orDie)
if (projected && sameBaseline(projected.parts, event.data.parts)) return yield* Effect.void
return yield* Effect.die("Session context update conflicts with stored projection")
}
const updated = yield* db
.update(SessionContextEpochTable)
.set({ checkpoint: event.data.checkpoint, revision: event.data.expectedRevision + 1 })
.where(
and(
eq(SessionContextEpochTable.session_id, event.data.sessionID),
eq(SessionContextEpochTable.revision, event.data.expectedRevision),
),
)
.returning({ revision: SessionContextEpochTable.revision })
.get()
.pipe(Effect.orDie)
if (!updated) return yield* Effect.die("Session context epoch revision mismatch")
if (event.data.parts.length === 0) return yield* Effect.void
return yield* db
.insert(SessionContextMessageTable)
.values({ session_id: event.data.sessionID, seq, parts: event.data.parts })
.run()
.pipe(Effect.orDie)
})
export const projectReplaced = Effect.fn("SessionContextEpoch.projectReplaced")(function* (
db: DatabaseService,
event: SessionEvent.ContextReplaced,
seq: number,
) {
const stored = yield* find(db, event.data.sessionID)
if (!stored) return yield* Effect.die("Session context epoch is not initialized")
if (stored.baseline_seq > seq) return yield* Effect.void
if (stored.baseline_seq === seq && sameBaseline(stored.baseline, event.data.baseline)) return yield* Effect.void
if (stored.replacement_seq === null) {
return yield* Effect.die("Session context epoch replacement was not requested")
}
const updated = yield* db
.update(SessionContextEpochTable)
.set({
baseline: event.data.baseline,
checkpoint: event.data.checkpoint,
baseline_seq: seq,
replacement_seq: null,
revision: event.data.expectedRevision + 1,
})
.where(
and(
eq(SessionContextEpochTable.session_id, event.data.sessionID),
eq(SessionContextEpochTable.revision, event.data.expectedRevision),
),
)
.returning({ revision: SessionContextEpochTable.revision })
.get()
.pipe(Effect.orDie)
if (!updated) return yield* Effect.die("Session context epoch revision mismatch")
return yield* Effect.void
})
export const requestReplacement = Effect.fn("SessionContextEpoch.requestReplacement")(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
@ -185,10 +65,97 @@ export const requestReplacement = Effect.fn("SessionContextEpoch.requestReplacem
.where(
and(
eq(SessionContextEpochTable.session_id, sessionID),
isNull(SessionContextEpochTable.replacement_seq),
lt(SessionContextEpochTable.baseline_seq, seq),
or(isNull(SessionContextEpochTable.replacement_seq), lt(SessionContextEpochTable.replacement_seq, seq)),
),
)
.run()
.pipe(Effect.orDie)
})
const initialize = Effect.fnUntraced(function* (
db: DatabaseService,
events: EventV2.Interface,
sessionID: SessionSchema.ID,
generation: SystemContext.Generation,
) {
return yield* db
.transaction(
() =>
Effect.gen(function* () {
const baselineSeq = yield* events.sequence(sessionID)
yield* db
.insert(SessionContextEpochTable)
.values({
session_id: sessionID,
baseline: generation.baseline,
snapshot: generation.snapshot,
baseline_seq: baselineSeq,
revision: 0,
})
.run()
.pipe(Effect.orDie)
return baselineSeq
}),
{ behavior: "immediate" },
)
.pipe(Effect.orDie)
})
const replace = Effect.fnUntraced(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
expectedRevision: number,
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" },
)
.pipe(Effect.orDie)
})
const advance = Effect.fnUntraced(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
expectedRevision: number,
snapshot: SystemContext.Snapshot,
) {
const updated = yield* db
.update(SessionContextEpochTable)
.set({ snapshot, revision: expectedRevision + 1 })
.where(
and(
eq(SessionContextEpochTable.session_id, sessionID),
eq(SessionContextEpochTable.revision, expectedRevision),
isNull(SessionContextEpochTable.replacement_seq),
),
)
.returning({ revision: SessionContextEpochTable.revision })
.get()
.pipe(Effect.orDie)
if (!updated) return yield* Effect.die("Session context epoch revision mismatch")
})

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

@ -1,22 +1,18 @@
import { and, asc, desc, eq, gt, gte, or } from "drizzle-orm"
import { and, asc, desc, eq, gt, gte, ne, or } 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 { SessionContextMessageTable, SessionMessageTable } from "./sql"
import type { SystemContext } from "../system-context"
import { SessionContextEpochTable, SessionMessageTable } from "./sql"
type DatabaseService = Database.Interface["db"]
const decode = Schema.decodeUnknownEffect(SessionMessage.Message)
export type RunnerMessage =
| SessionMessage.Message
| { readonly type: "system-context"; readonly parts: ReadonlyArray<SystemContext.Part> }
const latestCompaction = Effect.fnUntraced(function* (db: DatabaseService, sessionID: SessionSchema.ID) {
return yield* db
.select()
.select({ seq: SessionMessageTable.seq })
.from(SessionMessageTable)
.where(and(eq(SessionMessageTable.session_id, sessionID), eq(SessionMessageTable.type, "compaction")))
.orderBy(desc(SessionMessageTable.seq))
@ -28,7 +24,8 @@ const latestCompaction = Effect.fnUntraced(function* (db: DatabaseService, sessi
const messageRows = Effect.fnUntraced(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
compaction: typeof SessionMessageTable.$inferSelect | undefined,
compaction: { readonly seq: number } | undefined,
baselineSeq?: number,
) {
return yield* db
.select()
@ -36,7 +33,17 @@ const messageRows = Effect.fnUntraced(function* (
.where(
and(
eq(SessionMessageTable.session_id, sessionID),
compaction ? or(gte(SessionMessageTable.seq, compaction.seq)) : undefined,
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)),
),
)
.orderBy(asc(SessionMessageTable.seq))
@ -56,8 +63,20 @@ const decodeMessageRow = (row: typeof SessionMessageTable.$inferSelect) =>
)
export const load = Effect.fn("SessionContext.load")(function* (db: DatabaseService, sessionID: SessionSchema.ID) {
const [epoch, compaction] = yield* Effect.all(
[
db
.select({ baselineSeq: SessionContextEpochTable.baseline_seq })
.from(SessionContextEpochTable)
.where(eq(SessionContextEpochTable.session_id, sessionID))
.get()
.pipe(Effect.orDie),
latestCompaction(db, sessionID),
],
{ concurrency: "unbounded" },
)
return yield* Effect.forEach(
yield* messageRows(db, sessionID, yield* latestCompaction(db, sessionID)),
yield* messageRows(db, sessionID, compaction, epoch?.baselineSeq),
decodeMessageRow,
)
})
@ -67,44 +86,7 @@ export const loadForRunner = Effect.fn("SessionContext.loadForRunner")(function*
sessionID: SessionSchema.ID,
baselineSeq: number,
) {
const compaction = yield* latestCompaction(db, sessionID)
const messages = yield* messageRows(db, sessionID, compaction)
const updates = yield* db
.select()
.from(SessionContextMessageTable)
.where(and(eq(SessionContextMessageTable.session_id, sessionID), gt(SessionContextMessageTable.seq, baselineSeq)))
.orderBy(asc(SessionContextMessageTable.seq))
.all()
.pipe(Effect.orDie)
return yield* Effect.forEach(
merge(
messages.map((row) => ({ type: "message" as const, seq: row.seq, row })),
updates.map((row) => ({ type: "system-context" as const, seq: row.seq, row })),
),
(item): Effect.Effect<RunnerMessage, MessageDecodeError> =>
item.type === "message"
? decodeMessageRow(item.row)
: Effect.succeed({ type: "system-context", parts: item.row.parts }),
)
return yield* Effect.forEach(yield* messageRows(db, sessionID, yield* latestCompaction(db, sessionID), baselineSeq), decodeMessageRow)
})
function merge<Left extends { readonly seq: number }, Right extends { readonly seq: number }>(
left: ReadonlyArray<Left>,
right: ReadonlyArray<Right>,
): Array<Left | Right> {
const result: Array<Left | Right> = []
let leftIndex = 0
let rightIndex = 0
while (leftIndex < left.length || rightIndex < right.length) {
if (rightIndex >= right.length || (leftIndex < left.length && left[leftIndex].seq < right[rightIndex].seq)) {
result.push(left[leftIndex])
leftIndex++
continue
}
result.push(right[rightIndex])
rightIndex++
}
return result
}
export * as SessionContext from "./context"

31
packages/core/src/session/event.ts
View file

@ -10,7 +10,6 @@ import { SessionSchema } from "./schema"
import { Location } from "../location"
import { RelativePath } from "../schema"
import { SessionMessageID } from "./message-id"
import { SystemContext } from "../system-context"
export { FileAttachment }
@ -120,41 +119,17 @@ export namespace PromptLifecycle {
export type Promoted = typeof Promoted.Type
}
export const ContextInitialized = EventV2.define({
type: "session.next.context.initialized",
...options,
schema: {
...Base,
baseline: SystemContext.PartsSchema,
checkpoint: SystemContext.CheckpointSchema,
},
})
export type ContextInitialized = typeof ContextInitialized.Type
export const ContextUpdated = EventV2.define({
type: "session.next.context.updated",
...options,
schema: {
...Base,
expectedRevision: NonNegativeInt,
parts: SystemContext.PartsSchema,
checkpoint: SystemContext.CheckpointSchema,
messageID: SessionMessageID.ID,
text: Schema.String,
},
})
export type ContextUpdated = typeof ContextUpdated.Type
export const ContextReplaced = EventV2.define({
type: "session.next.context.replaced",
...options,
schema: {
...Base,
expectedRevision: NonNegativeInt,
baseline: SystemContext.PartsSchema,
checkpoint: SystemContext.CheckpointSchema,
},
})
export type ContextReplaced = typeof ContextReplaced.Type
export const Synthetic = EventV2.define({
type: "session.next.synthetic",
...options,
@ -480,9 +455,7 @@ const DurableDefinitions = [
Prompted,
PromptLifecycle.Admitted,
PromptLifecycle.Promoted,
ContextInitialized,
ContextUpdated,
ContextReplaced,
Synthetic,
Shell.Started,
Shell.Ended,

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

@ -159,9 +159,15 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
},
"session.next.prompt.admitted": () => Effect.void,
"session.next.prompt.promoted": () => Effect.void,
"session.next.context.initialized": () => Effect.void,
"session.next.context.updated": () => Effect.void,
"session.next.context.replaced": () => Effect.void,
"session.next.context.updated": (event) =>
adapter.appendMessage(
new SessionMessage.System({
id: event.data.messageID,
type: "system",
text: event.data.text,
time: { created: event.data.timestamp },
}),
),
"session.next.synthetic": (event) => {
return adapter.appendMessage(
new SessionMessage.Synthetic({

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

@ -51,6 +51,12 @@ export class Synthetic extends Schema.Class<Synthetic>("Session.Message.Syntheti
type: Schema.Literal("synthetic"),
}) {}
export class System extends Schema.Class<System>("Session.Message.System")({
...Base,
type: Schema.Literal("system"),
text: SessionEvent.ContextUpdated.data.fields.text,
}) {}
export class Shell extends Schema.Class<Shell>("Session.Message.Shell")({
...Base,
type: Schema.Literal("shell"),
@ -170,7 +176,7 @@ export class Compaction extends Schema.Class<Compaction>("Session.Message.Compac
...Base,
}) {}
export const Message = Schema.Union([AgentSwitched, ModelSwitched, User, Synthetic, Shell, Assistant, Compaction])
export const Message = Schema.Union([AgentSwitched, ModelSwitched, User, Synthetic, System, Shell, Assistant, Compaction])
.pipe(Schema.toTaggedUnion("type"))
.annotate({ identifier: "Session.Message" })

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

@ -420,17 +420,9 @@ export const layer = Layer.effectDiscard(
)
}),
)
yield* events.project(SessionEvent.ContextInitialized, (event) => {
if (event.seq === undefined) return Effect.die("Synchronized Session event is missing aggregate sequence")
return SessionContextEpoch.projectInitialized(db, event, event.seq)
})
yield* events.project(SessionEvent.ContextUpdated, (event) => {
if (event.seq === undefined) return Effect.die("Synchronized Session event is missing aggregate sequence")
return SessionContextEpoch.projectUpdated(db, event, event.seq)
})
yield* events.project(SessionEvent.ContextReplaced, (event) => {
if (event.seq === undefined) return Effect.die("Synchronized Session event is missing aggregate sequence")
return SessionContextEpoch.projectReplaced(db, event, event.seq)
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)))
})
yield* events.project(SessionEvent.Synthetic, (event) => run(db, event))
yield* events.project(SessionEvent.Shell.Started, (event) => run(db, event))

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

@ -154,7 +154,7 @@ export const layer = Layer.effect(
const context = yield* getRunnerContext(session.id, system.baselineSeq)
const request = LLM.request({
model,
system: system.baseline.map((part) => SystemPart.make(part.text)),
system: system.baseline.length > 0 ? [SystemPart.make(system.baseline)] : [],
messages: toLLMMessages(context, model),
tools: yield* tools.definitions(),
})

9
packages/core/src/session/runner/to-llm-message.ts
View file

@ -9,7 +9,6 @@ import {
} from "@opencode-ai/llm"
import { SessionMessage } from "../message"
import type { FileAttachment } from "../prompt"
import { SessionContext } from "../context"
const media = (file: FileAttachment): ContentPart => ({
type: "media",
@ -92,7 +91,7 @@ const assistant = (message: SessionMessage.Assistant, model: Model) => {
return [Message.make({ id: message.id, role: "assistant", content, metadata: message.metadata }), ...results]
}
function toLLMMessage(message: SessionContext.RunnerMessage, model: Model): Message[] {
function toLLMMessage(message: SessionMessage.Message, model: Model): Message[] {
switch (message.type) {
case "agent-switched":
case "model-switched":
@ -112,6 +111,8 @@ function toLLMMessage(message: SessionContext.RunnerMessage, model: Model): Mess
]
case "synthetic":
return [Message.make({ id: message.id, role: "user", content: message.text, metadata: message.metadata })]
case "system":
return [Message.system(message.text)]
case "shell":
return [
Message.make({
@ -132,11 +133,9 @@ function toLLMMessage(message: SessionContext.RunnerMessage, model: Model): Mess
metadata: message.metadata,
}),
]
case "system-context":
return [Message.system(message.parts.map((part) => ({ type: "text", text: part.text })))]
}
}
/** Translate projected V2 Session history into canonical @opencode-ai/llm context. */
export const toLLMMessages = (messages: readonly SessionContext.RunnerMessage[], model: Model) =>
export const toLLMMessages = (messages: readonly SessionMessage.Message[], model: Model) =>
messages.flatMap((message) => toLLMMessage(message, model))

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

@ -168,22 +168,9 @@ export const SessionContextEpochTable = sqliteTable("session_context_epoch", {
.$type<SessionSchema.ID>()
.primaryKey()
.references(() => SessionTable.id, { onDelete: "cascade" }),
baseline: text({ mode: "json" }).notNull().$type<ReadonlyArray<SystemContext.Part>>(),
checkpoint: text({ mode: "json" }).notNull().$type<SystemContext.Checkpoint>(),
baseline: text().notNull(),
snapshot: text({ mode: "json" }).notNull().$type<SystemContext.Snapshot>(),
baseline_seq: integer().notNull(),
replacement_seq: integer(),
revision: integer().notNull().default(0),
})
export const SessionContextMessageTable = sqliteTable(
"session_context_message",
{
session_id: text()
.$type<SessionSchema.ID>()
.notNull()
.references(() => SessionTable.id, { onDelete: "cascade" }),
seq: integer().notNull(),
parts: text({ mode: "json" }).notNull().$type<ReadonlyArray<SystemContext.Part>>(),
},
(table) => [primaryKey({ columns: [table.session_id, table.seq] })],
)

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

@ -16,7 +16,7 @@ export interface Interface {
readonly runnerContext: (
sessionID: SessionSchema.ID,
baselineSeq: number,
) => Effect.Effect<SessionContext.RunnerMessage[], MessageDecodeError>
) => Effect.Effect<SessionMessage.Message[], MessageDecodeError>
readonly message: (
messageID: SessionMessage.ID,
) => Effect.Effect<{ readonly sessionID: SessionSchema.ID; readonly message: SessionMessage.Message } | undefined>

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

@ -1,73 +1,84 @@
export * as SystemContext from "./system-context"
import { Effect, Schema } from "effect"
import { Hash } from "./util/hash"
import { Effect, Option, Schema } from "effect"
/**
* Models privileged system context as independently refreshable typed sources.
*
* `Source<A>` describes how to observe, compare, and render one value. `make`
* closes over `A`, producing an opaque `SystemContext` that composes uniformly
* with contexts built from other value types. Interpreters observe the composed
* context once, then produce a durable structured
* `Snapshot` alongside the exact model-visible baseline or update text.
*
* Returning `unavailable` means observation failed temporarily. It differs from
* removing a source from the context: refresh preserves the admitted snapshot,
* and replacement waits rather than silently constructing an incomplete baseline.
*
* @module
*/
/** Stable namespaced identity for one independently refreshable context source. */
export const Key = Schema.String.check(Schema.isPattern(/^[a-z0-9][a-z0-9._-]*\/[a-z0-9][a-z0-9._/-]*$/)).pipe(
Schema.brand("SystemContext.Key"),
)
export type Key = typeof Key.Type
/** Indicates that a source could not be observed without treating it as removed. */
export const unavailable = Symbol.for("@opencode/SystemContext.Unavailable")
export type Unavailable = typeof unavailable
export interface Value {
/** Full component text rendered into a new epoch baseline. */
readonly baseline: string
/** Absolute current-state text emitted when this component changes. */
readonly update: string
}
export interface Component<out E = never, out R = never> {
/** Defines one typed source before its value type is hidden by `make`. */
export interface Source<A> {
readonly key: Key
readonly load: Effect.Effect<Value | Unavailable, E, R>
readonly codec: Schema.Codec<A, Schema.Json, never, never>
readonly load: Effect.Effect<A | Unavailable>
readonly baseline: (current: A) => string
readonly update: (previous: A, current: A) => string
readonly removed?: (previous: A) => string
}
export interface SystemContext<out E = never, out R = never> {
readonly components: ReadonlyArray<Component<E, R>>
const ContextTypeId: unique symbol = Symbol.for("@opencode/SystemContext")
/** Opaque carrier for composable system context sources. */
export interface SystemContext {
readonly [ContextTypeId]: ReadonlyArray<PackedSource>
}
export interface AvailableEntry extends Value {
readonly _tag: "Available"
readonly key: Key
readonly hash: string
}
export interface UnavailableEntry {
readonly _tag: "Unavailable"
readonly key: Key
}
export type Entry = AvailableEntry | UnavailableEntry
export interface Snapshot {
readonly entries: ReadonlyArray<Entry>
}
export interface Part {
readonly key: Key
readonly text: string
}
export const PartSchema = Schema.Struct({
key: Key,
text: Schema.String,
/** Durable comparison state for one admitted source. */
export const SourceSnapshot = Schema.Struct({
value: Schema.Json,
removed: Schema.optional(Schema.NonEmptyString),
})
export const PartsSchema = Schema.Array(PartSchema)
export const CheckpointSchema = Schema.Record(Key, Schema.String)
export type SourceSnapshot = typeof SourceSnapshot.Type
export type Checkpoint = Readonly<Record<string, string>>
/** Durable structured comparison state for one active context generation. */
export const Snapshot = Schema.Record(Key, SourceSnapshot)
export type Snapshot = Readonly<Record<string, SourceSnapshot>>
export interface Initialized {
readonly baseline: ReadonlyArray<Part>
readonly checkpoint: Checkpoint
export interface Generation {
readonly baseline: string
readonly snapshot: Snapshot
}
export interface Refreshed {
readonly changes: ReadonlyArray<Part>
readonly checkpoint: Checkpoint
export interface Updated {
readonly _tag: "Updated"
readonly text: string
readonly snapshot: Snapshot
}
export interface Replaced {
readonly _tag: "Replaced"
readonly generation: Generation
}
export interface ReplacementBlocked {
readonly _tag: "ReplacementBlocked"
}
export type ReplacementResult = Replaced | ReplacementBlocked
export type ReconcileResult = { readonly _tag: "Unchanged" } | Updated | ReplacementResult
export class DuplicateKeyError extends Schema.TaggedErrorClass<DuplicateKeyError>()("SystemContext.DuplicateKeyError", {
key: Key,
}) {
@ -76,78 +87,219 @@ export class DuplicateKeyError extends Schema.TaggedErrorClass<DuplicateKeyError
}
}
export const value = <E, R>(component: Component<E, R>): Component<E, R> => component
export function struct<E, R>(components: Readonly<Record<string, Component<E, R>>>): SystemContext<E, R> {
const values = Object.values(components)
assertUniqueKeys(values)
return { components: values }
interface PackedSource {
readonly key: Key
readonly load: Effect.Effect<Loaded | Unavailable>
}
export const load = <E, R>(context: SystemContext<E, R>) =>
Effect.sync(() => assertUniqueKeys(context.components)).pipe(
Effect.andThen(
Effect.forEach(context.components, (component) =>
component.load.pipe(
Effect.map(
(result): Entry =>
result === unavailable
? { _tag: "Unavailable", key: component.key }
: { _tag: "Available", key: component.key, ...result, hash: Hash.sha256(result.update) },
),
interface Loaded {
readonly baseline: () => Rendered
readonly compare: (previous: Schema.Json) => Compared
}
interface Rendered {
readonly text: string
readonly snapshot: SourceSnapshot
}
type Compared =
| { readonly _tag: "Incompatible" }
| { readonly _tag: "Unchanged" }
| { readonly _tag: "Updated"; readonly render: () => Rendered }
interface AvailableEntry extends Loaded {
readonly _tag: "Available"
readonly key: Key
}
interface UnavailableEntry {
readonly _tag: "Unavailable"
readonly key: Key
}
type Entry = AvailableEntry | UnavailableEntry
/** The identity context. */
export const empty = context([])
/** Closes a typed source into a context that composes with differently typed sources. */
export function make<A>(source: Source<A>): SystemContext {
const decode = Schema.decodeUnknownOption(source.codec)
const encode = Schema.encodeSync(source.codec)
const equivalent = Schema.toEquivalence(source.codec)
return context([
{
key: source.key,
load: source.load.pipe(
Effect.map((value) => {
if (isUnavailable(value)) return value
const snapshot = (): SourceSnapshot => ({
value: encode(value),
...(source.removed ? { removed: requireText(source.key, "removal", source.removed(value)) } : {}),
})
return {
baseline: (): Rendered => ({
text: requireText(source.key, "baseline", source.baseline(value)),
snapshot: snapshot(),
}),
compare: (previous): Compared =>
Option.match(decode(previous), {
onNone: (): Compared => ({ _tag: "Incompatible" }),
onSome: (decoded): Compared =>
equivalent(decoded, value)
? { _tag: "Unchanged" }
: {
_tag: "Updated",
render: () => ({
text: requireText(source.key, "update", source.update(decoded, value)),
snapshot: snapshot(),
}),
},
}),
}
}),
),
},
])
}
/** Combines contexts in order and rejects duplicate source keys immediately. */
export function combine(values: ReadonlyArray<SystemContext>): SystemContext {
const sources = values.flatMap((value) => value[ContextTypeId])
assertUniqueKeys(sources)
return context(sources)
}
const observe = (value: SystemContext) =>
Effect.forEach(
value[ContextTypeId],
(source) =>
source.load.pipe(
Effect.map(
(result): Entry =>
result === unavailable
? { _tag: "Unavailable", key: source.key }
: { _tag: "Available", key: source.key, ...result },
),
),
),
Effect.map((entries): Snapshot => ({ entries })),
{ concurrency: "unbounded" },
)
export function initialize(snapshot: Snapshot): Initialized {
/** 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))
}
function initializeObservation(entries: ReadonlyArray<Entry>): Generation {
const available = entries.filter((entry): entry is AvailableEntry => entry._tag === "Available")
const rendered = available.map((entry) => [entry.key, entry.baseline()] as const)
return {
baseline: snapshot.entries.flatMap((entry) =>
entry._tag === "Available" ? [{ key: entry.key, text: entry.baseline }] : [],
),
checkpoint: nextCheckpoint(snapshot, {}),
baseline: render(rendered.map(([, result]) => result.text)),
snapshot: Object.fromEntries(rendered.map(([key, result]) => [key, result.snapshot])),
}
}
export function refresh(snapshot: Snapshot, previous: Checkpoint): Refreshed {
const keys = new Set(snapshot.entries.map((entry) => entry.key))
return {
changes: [
...snapshot.entries.flatMap((entry) =>
entry._tag === "Available" && getCheckpoint(previous, entry.key) !== entry.hash
? [{ key: entry.key, text: entry.update }]
: [],
),
...Object.keys(previous).flatMap((key) =>
keys.has(Key.make(key)) ? [] : [{ key: Key.make(key), text: `System context component removed: ${key}` }],
),
],
checkpoint: nextCheckpoint(snapshot, previous),
}
}
export const replacementBlocked = (snapshot: Snapshot, previous: Checkpoint) =>
snapshot.entries.some((entry) => entry._tag === "Unavailable" && getCheckpoint(previous, entry.key) !== undefined)
function nextCheckpoint(snapshot: Snapshot, previous: Checkpoint) {
return Object.fromEntries(
snapshot.entries.flatMap((entry) => {
if (entry._tag === "Available") return [[entry.key, entry.hash]]
const hash = getCheckpoint(previous, entry.key)
return hash === undefined ? [] : [[entry.key, hash]]
/** Reconciles current source values with one active generation. */
export function reconcile(value: SystemContext, previous: Snapshot): Effect.Effect<ReconcileResult> {
return observe(value).pipe(
Effect.map((entries): ReconcileResult => {
const result = reconcileObservation(entries, previous)
if (result._tag === "Unchanged" || result._tag === "Updated") return result
return replaceObservation(entries, previous)
}),
)
}
function getCheckpoint(checkpoint: Checkpoint, key: Key) {
return Object.hasOwn(checkpoint, key) ? checkpoint[key] : undefined
function reconcileObservation(
entries: ReadonlyArray<Entry>,
previous: Snapshot,
): { readonly _tag: "Unchanged" } | Updated | { readonly _tag: "Replace" } {
const keys = new Set(entries.map((entry) => entry.key))
const comparisons = new Map<Key, Compared>()
for (const entry of entries) {
if (entry._tag === "Unavailable") continue
const stored = getSnapshot(previous, entry.key)
if (!stored) continue
const compared = entry.compare(stored.value)
if (compared._tag === "Incompatible") return { _tag: "Replace" }
comparisons.set(entry.key, compared)
}
for (const key of Object.keys(previous).sort()) {
if (keys.has(Key.make(key))) continue
if (previous[key].removed === undefined) return { _tag: "Replace" }
}
const snapshot: Record<string, SourceSnapshot> = {}
const updates: string[] = []
for (const entry of entries) {
const stored = getSnapshot(previous, entry.key)
if (entry._tag === "Unavailable") {
if (stored) snapshot[entry.key] = stored
continue
}
if (!stored) {
const rendered = entry.baseline()
updates.push(rendered.text)
snapshot[entry.key] = rendered.snapshot
continue
}
const compared = comparisons.get(entry.key)
if (!compared || compared._tag === "Incompatible")
throw new Error(`Missing comparison for system context source ${entry.key}`)
if (compared._tag === "Unchanged") {
snapshot[entry.key] = stored
continue
}
const rendered = compared.render()
updates.push(rendered.text)
snapshot[entry.key] = rendered.snapshot
}
for (const key of Object.keys(previous).sort()) {
if (keys.has(Key.make(key))) continue
const removed = previous[key].removed
if (removed === undefined) throw new Error(`Missing removal rendering for system context source ${key}`)
updates.push(removed)
}
if (updates.length === 0) return { _tag: "Unchanged" }
return { _tag: "Updated", text: render(updates), snapshot }
}
function assertUniqueKeys(components: ReadonlyArray<Component<unknown, unknown>>) {
/** Creates a complete replacement generation or blocks while admitted context is unavailable. */
export function replace(value: SystemContext, previous: Snapshot): Effect.Effect<ReplacementResult> {
return observe(value).pipe(Effect.map((entries) => replaceObservation(entries, previous)))
}
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) }
}
function context(sources: ReadonlyArray<PackedSource>): SystemContext {
return { [ContextTypeId]: sources }
}
function render(parts: ReadonlyArray<string>) {
return parts.join("\n\n")
}
function getSnapshot(snapshot: Snapshot, key: Key) {
return Object.hasOwn(snapshot, key) ? snapshot[key] : undefined
}
function isUnavailable(value: unknown): value is Unavailable {
return value === unavailable
}
function requireText(key: Key, kind: string, text: string) {
if (text.length === 0) throw new Error(`System context source ${key} rendered an empty ${kind}`)
return text
}
function assertUniqueKeys(sources: ReadonlyArray<PackedSource>) {
const keys = new Set<Key>()
for (const component of components) {
if (keys.has(component.key)) throw new DuplicateKeyError({ key: component.key })
keys.add(component.key)
for (const source of sources) {
if (keys.has(source.key)) throw new DuplicateKeyError({ key: source.key })
keys.add(source.key)
}
}