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

feat(core): persist v2 session context epochs

Browse source
This commit is contained in:
Kit Langton 2026-06-04 14:38:31 -04:00
commit 83916f667d
39 changed files with 9302 additions and 79 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -32,5 +32,9 @@ export const migrations = (
import("./migration/20260603141458_session_input_inbox"),
import("./migration/20260603160727_jittery_ezekiel_stane"),
import("./migration/20260604172448_event_sourced_session_input"),
import("./migration/20260604180746_add_session_context_epoch"),
import("./migration/20260604181329_add_session_context_updates"),
import("./migration/20260604181706_add_session_context_replacement"),
import("./migration/20260604181807_add_session_context_replacement_sequence"),
])
).map((module) => module.default) satisfies DatabaseMigration.Migration[]

19
packages/core/src/database/migration/20260604180746_add_session_context_epoch.ts Normal file
View file

@ -0,0 +1,19 @@
import { Effect } from "effect"
import type { DatabaseMigration } from "../migration"
export default {
id: "20260604180746_add_session_context_epoch",
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,
\`baseline_seq\` integer NOT NULL,
CONSTRAINT \`fk_session_context_epoch_session_id_session_id_fk\` FOREIGN KEY (\`session_id\`) REFERENCES \`session\`(\`id\`) ON DELETE CASCADE
);
`)
})
},
} satisfies DatabaseMigration.Migration

23
packages/core/src/database/migration/20260604181329_add_session_context_updates.ts Normal file
View file

@ -0,0 +1,23 @@
import { Effect } from "effect"
import type { DatabaseMigration } from "../migration"
export default {
id: "20260604181329_add_session_context_updates",
up(tx) {
return Effect.gen(function* () {
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
);
`)
yield* tx.run(`ALTER TABLE \`session_context_epoch\` ADD \`revision\` integer DEFAULT 0 NOT NULL;`)
yield* tx.run(
`CREATE INDEX \`session_context_message_session_seq_idx\` ON \`session_context_message\` (\`session_id\`,\`seq\`);`,
)
})
},
} satisfies DatabaseMigration.Migration

11
packages/core/src/database/migration/20260604181706_add_session_context_replacement.ts Normal file
View file

@ -0,0 +1,11 @@
import { Effect } from "effect"
import type { DatabaseMigration } from "../migration"
export default {
id: "20260604181706_add_session_context_replacement",
up(tx) {
return Effect.gen(function* () {
yield* tx.run(`ALTER TABLE \`session_context_epoch\` ADD \`replacement_pending\` integer DEFAULT false NOT NULL;`)
})
},
} satisfies DatabaseMigration.Migration

11
packages/core/src/database/migration/20260604181807_add_session_context_replacement_sequence.ts Normal file
View file

@ -0,0 +1,11 @@
import { Effect } from "effect"
import type { DatabaseMigration } from "../migration"
export default {
id: "20260604181807_add_session_context_replacement_sequence",
up(tx) {
return Effect.gen(function* () {
yield* tx.run(`ALTER TABLE \`session_context_epoch\` ADD \`replacement_seq\` integer;`)
})
},
} satisfies DatabaseMigration.Migration

2
packages/core/src/location-layer.ts
View file

@ -40,6 +40,7 @@ import { RequestExecutor } from "@opencode-ai/llm/route"
import * as SessionRunnerLLM from "./session/runner/llm"
import { SessionRunnerModel } from "./session/runner/model"
import { SessionRunCoordinator } from "./session/run-coordinator"
import { SessionSystemContext } from "./session-system-context"
import { FetchHttpClient } from "effect/unstable/http"
export class LocationServiceMap extends LayerMap.Service<LocationServiceMap>()("@opencode/example/LocationServiceMap", {
@ -60,6 +61,7 @@ export class LocationServiceMap extends LayerMap.Service<LocationServiceMap>()("
Watcher.locationLayer,
Pty.locationLayer,
SkillV2.locationLayer,
SessionSystemContext.locationLayer,
permissionsAndTools,
LocationMutation.locationLayer.pipe(Layer.orDie),
).pipe(Layer.provideMerge(location))

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

@ -0,0 +1,184 @@
export * as SessionContextEpoch from "./context-epoch"
import { and, eq, 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 { SessionSchema } from "./schema"
import { SessionContextEpochTable, SessionContextMessageTable } 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,
events: EventV2.Interface,
context: SessionSystemContext.Interface,
sessionID: SessionSchema.ID,
) {
const snapshot = yield* context.load()
const stored = yield* find(db, sessionID)
if (!stored) {
const initialized = SystemContext.initialize(snapshot)
yield* events.publish(SessionEvent.ContextInitialized, {
sessionID,
timestamp: yield* DateTime.now,
baseline: initialized.baseline,
checkpoint: initialized.checkpoint,
})
return initialized.baseline
}
if (stored.replacement_pending) {
const initialized = SystemContext.initialize(snapshot)
yield* events.publish(SessionEvent.ContextReplaced, {
sessionID,
timestamp: yield* DateTime.now,
expectedRevision: stored.revision,
baseline: initialized.baseline,
checkpoint: initialized.checkpoint,
})
return initialized.baseline
}
const refreshed = SystemContext.refresh(snapshot, stored.checkpoint)
if (sameCheckpoint(refreshed.checkpoint, stored.checkpoint)) return stored.baseline
yield* events.publish(SessionEvent.ContextUpdated, {
sessionID,
timestamp: yield* DateTime.now,
expectedRevision: stored.revision,
parts: refreshed.changes,
checkpoint: refreshed.checkpoint,
})
return stored.baseline
})
export const find = Effect.fn("SessionContextEpoch.find")(function* (db: DatabaseService, sessionID: SessionSchema.ID) {
return yield* db
.select()
.from(SessionContextEpochTable)
.where(eq(SessionContextEpochTable.session_id, sessionID))
.get()
.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_pending: false,
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.replacement_pending) 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.replacement_pending) {
if (stored.baseline_seq === seq && sameBaseline(stored.baseline, event.data.baseline)) return yield* Effect.void
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_pending: false,
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,
seq: number,
) {
const stored = yield* find(db, sessionID)
if (!stored || stored.baseline_seq >= seq || stored.replacement_seq === seq) return yield* Effect.void
return yield* db
.update(SessionContextEpochTable)
.set({ replacement_pending: true, replacement_seq: seq, revision: sql`${SessionContextEpochTable.revision} + 1` })
.where(eq(SessionContextEpochTable.session_id, sessionID))
.run()
.pipe(Effect.orDie)
})

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

@ -4,14 +4,18 @@ import { Database } from "../database/database"
import { MessageDecodeError } from "./error"
import { SessionMessage } from "./message"
import { SessionSchema } from "./schema"
import { SessionMessageTable } from "./sql"
import { SessionContextEpochTable, SessionContextMessageTable, SessionMessageTable } from "./sql"
import type { SystemContext } from "../system-context"
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> }
export const load = Effect.fn("SessionContext.load")(function* (db: DatabaseService, sessionID: SessionSchema.ID) {
const compaction = yield* db
const latestCompaction = Effect.fnUntraced(function* (db: DatabaseService, sessionID: SessionSchema.ID) {
return yield* db
.select()
.from(SessionMessageTable)
.where(and(eq(SessionMessageTable.session_id, sessionID), eq(SessionMessageTable.type, "compaction")))
@ -19,7 +23,14 @@ export const load = Effect.fn("SessionContext.load")(function* (db: DatabaseServ
.limit(1)
.get()
.pipe(Effect.orDie)
const rows = yield* db
})
const messageRows = Effect.fnUntraced(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
compaction: typeof SessionMessageTable.$inferSelect | undefined,
) {
return yield* db
.select()
.from(SessionMessageTable)
.where(
@ -31,17 +42,79 @@ export const load = Effect.fn("SessionContext.load")(function* (db: DatabaseServ
.orderBy(asc(SessionMessageTable.seq))
.all()
.pipe(Effect.orDie)
return yield* Effect.forEach(rows, (row) =>
decode({ ...row.data, id: row.id, type: row.type }).pipe(
Effect.mapError(
() =>
new MessageDecodeError({
sessionID: SessionSchema.ID.make(row.session_id),
messageID: SessionMessage.ID.make(row.id),
}),
),
})
const decodeMessageRow = (row: typeof SessionMessageTable.$inferSelect) =>
decode({ ...row.data, id: row.id, type: row.type }).pipe(
Effect.mapError(
() =>
new MessageDecodeError({
sessionID: SessionSchema.ID.make(row.session_id),
messageID: SessionMessage.ID.make(row.id),
}),
),
)
export const load = Effect.fn("SessionContext.load")(function* (db: DatabaseService, sessionID: SessionSchema.ID) {
return yield* Effect.forEach(
yield* messageRows(db, sessionID, yield* latestCompaction(db, sessionID)),
decodeMessageRow,
)
})
export const loadForRunner = Effect.fn("SessionContext.loadForRunner")(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
) {
const compaction = yield* latestCompaction(db, sessionID)
const messages = yield* messageRows(db, sessionID, compaction)
const epoch = yield* db
.select({ baselineSeq: SessionContextEpochTable.baseline_seq })
.from(SessionContextEpochTable)
.where(eq(SessionContextEpochTable.session_id, sessionID))
.get()
.pipe(Effect.orDie)
const updates = yield* db
.select()
.from(SessionContextMessageTable)
.where(
and(
eq(SessionContextMessageTable.session_id, sessionID),
epoch ? gt(SessionContextMessageTable.seq, epoch.baselineSeq) : undefined,
),
)
.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 }),
)
})
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"

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

@ -10,6 +10,7 @@ import { SessionSchema } from "./schema"
import { Location } from "../location"
import { RelativePath } from "../schema"
import { SessionMessageID } from "./message-id"
import { SystemContext } from "../system-context"
export { FileAttachment }
@ -119,6 +120,41 @@ 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,
},
})
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,
@ -444,6 +480,9 @@ const DurableDefinitions = [
Prompted,
PromptLifecycle.Admitted,
PromptLifecycle.Promoted,
ContextInitialized,
ContextUpdated,
ContextReplaced,
Synthetic,
Shell.Started,
Shell.Ended,

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

@ -159,6 +159,9 @@ 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.synthetic": (event) => {
return adapter.appendMessage(
new SessionMessage.Synthetic({

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

@ -11,6 +11,7 @@ import { SessionMessage } from "./message"
import { SessionMessageUpdater } from "./message-updater"
import { SessionInput } from "./input"
import { WorkspaceV2 } from "../workspace"
import { SessionContextEpoch } from "./context-epoch"
import { MessageTable, PartTable, SessionMessageTable, SessionTable } from "./sql"
import type { DeepMutable } from "../schema"
@ -352,12 +353,18 @@ export const layer = Layer.effectDiscard(
.pipe(Effect.orDie, Effect.andThen(run(db, event))),
)
yield* events.project(SessionEvent.ModelSwitched, (event) =>
db
.update(SessionTable)
.set({ model: event.data.model, time_updated: DateTime.toEpochMillis(event.data.timestamp) })
.where(eq(SessionTable.id, event.data.sessionID))
.run()
.pipe(Effect.orDie, Effect.andThen(run(db, event))),
Effect.gen(function* () {
yield* db
.update(SessionTable)
.set({ model: event.data.model, time_updated: DateTime.toEpochMillis(event.data.timestamp) })
.where(eq(SessionTable.id, event.data.sessionID))
.run()
.pipe(Effect.orDie)
yield* run(db, event)
if (event.seq === undefined)
return yield* Effect.die("Synchronized Session event is missing aggregate sequence")
yield* SessionContextEpoch.requestReplacement(db, event.data.sessionID, event.seq)
}),
)
yield* events.project(SessionEvent.Prompted, (event) =>
Effect.gen(function* () {
@ -413,6 +420,18 @@ 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)
})
yield* events.project(SessionEvent.Synthetic, (event) => run(db, event))
yield* events.project(SessionEvent.Shell.Started, (event) => run(db, event))
yield* events.project(SessionEvent.Shell.Ended, (event) => run(db, event))
@ -432,7 +451,12 @@ export const layer = Layer.effectDiscard(
// yield* events.project(SessionEvent.Retried, (event) => run(db, event))
yield* events.project(SessionEvent.Compaction.Started, (event) => run(db, event))
yield* events.project(SessionEvent.Compaction.Delta, (event) => run(db, event))
yield* events.project(SessionEvent.Compaction.Ended, (event) => run(db, event))
yield* events.project(SessionEvent.Compaction.Ended, (event) => {
if (event.seq === undefined) return Effect.die("Synchronized Session event is missing aggregate sequence")
return run(db, event).pipe(
Effect.andThen(SessionContextEpoch.requestReplacement(db, event.data.sessionID, event.seq)),
)
})
}),
)

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

@ -1,4 +1,4 @@
import { LLM, LLMClient, LLMError, LLMEvent } from "@opencode-ai/llm"
import { LLM, LLMClient, LLMError, LLMEvent, SystemPart } from "@opencode-ai/llm"
import { Cause, DateTime, Effect, FiberSet, Layer, Semaphore, Stream } from "effect"
import { EventV2 } from "../../event"
import { ModelV2 } from "../../model"
@ -14,6 +14,8 @@ import { SessionRunnerModel } from "./model"
import { Database } from "../../database/database"
import { SessionInput } from "../input"
import { QuestionV2 } from "../../question"
import { SessionSystemContext } from "../../session-system-context"
import { SessionContextEpoch } from "../context-epoch"
/**
* Runs one durable coding-agent Session until it settles.
@ -85,6 +87,7 @@ export const layer = Layer.effect(
const tools = yield* ToolRegistry.Service
const models = yield* SessionRunnerModel.Service
const store = yield* SessionStore.Service
const systemContext = yield* SessionSystemContext.Service
const db = (yield* Database.Service).db
const getSession = Effect.fn("SessionRunner.getSession")(function* (sessionID: SessionSchema.ID) {
const session = yield* store.get(sessionID)
@ -95,6 +98,9 @@ export const layer = Layer.effect(
const getContext = Effect.fn("SessionRunner.getContext")(function* (sessionID: SessionSchema.ID) {
return yield* store.context(sessionID)
})
const getRunnerContext = Effect.fn("SessionRunner.getRunnerContext")(function* (sessionID: SessionSchema.ID) {
return yield* store.runnerContext(sessionID)
})
const failInterruptedTools = Effect.fn("SessionRunner.failInterruptedTools")(function* (
sessionID: SessionSchema.ID,
@ -141,8 +147,14 @@ export const layer = Layer.effect(
}
}
yield* failInterruptedTools(session.id)
const context = yield* getContext(session.id)
const request = LLM.request({ model, messages: toLLMMessages(context, model), tools: yield* tools.definitions() })
const system = yield* SessionContextEpoch.prepare(db, events, systemContext, session.id)
const context = yield* getRunnerContext(session.id)
const request = LLM.request({
model,
system: system.map((part) => SystemPart.make(part.text)),
messages: toLLMMessages(context, model),
tools: yield* tools.definitions(),
})
const publisher = createLLMEventPublisher(events, {
sessionID: session.id,
agent: session.agent ?? "build",

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

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

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

@ -11,6 +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 { SystemContext } from "../system-context"
type SessionMessageData = Omit<(typeof SessionMessage.Message)["Encoded"], "type" | "id">
type V1MessageData = Omit<SessionV1.Info, "id" | "sessionID">
@ -161,3 +162,32 @@ export const SessionInputTable = sqliteTable(
uniqueIndex("session_input_session_promoted_seq_idx").on(table.session_id, table.promoted_seq),
],
)
export const SessionContextEpochTable = sqliteTable("session_context_epoch", {
session_id: text()
.$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_seq: integer().notNull(),
replacement_pending: integer({ mode: "boolean" }).notNull().default(false),
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] }),
index("session_context_message_session_seq_idx").on(table.session_id, table.seq),
],
)

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

@ -13,6 +13,9 @@ import { fromRow } from "./info"
export interface Interface {
readonly get: (sessionID: SessionSchema.ID) => Effect.Effect<SessionSchema.Info | undefined>
readonly context: (sessionID: SessionSchema.ID) => Effect.Effect<SessionMessage.Message[], MessageDecodeError>
readonly runnerContext: (
sessionID: SessionSchema.ID,
) => Effect.Effect<SessionContext.RunnerMessage[], MessageDecodeError>
readonly message: (
messageID: SessionMessage.ID,
) => Effect.Effect<{ readonly sessionID: SessionSchema.ID; readonly message: SessionMessage.Message } | undefined>
@ -34,6 +37,9 @@ export const layer = Layer.effect(
context: Effect.fn("SessionStore.context")(function* (sessionID) {
return yield* SessionContext.load(db, sessionID)
}),
runnerContext: Effect.fn("SessionStore.runnerContext")(function* (sessionID) {
return yield* SessionContext.loadForRunner(db, sessionID)
}),
message: Effect.fn("SessionStore.message")(function* (messageID) {
const row = yield* db
.select()

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

@ -49,6 +49,13 @@ export interface Part {
readonly text: string
}
export const PartSchema = Schema.Struct({
key: Key,
text: Schema.String,
})
export const PartsSchema = Schema.Array(PartSchema)
export const CheckpointSchema = Schema.Record(Schema.String, Schema.String)
export type Checkpoint = Readonly<Record<string, string>>
export interface Initialized {
@ -105,11 +112,18 @@ export function initialize(snapshot: Snapshot): Initialized {
export function refresh(snapshot: Snapshot, previous: Checkpoint): Refreshed {
return {
changes: snapshot.entries.flatMap((entry) =>
entry._tag === "Available" && getCheckpoint(previous, entry.key) !== entry.hash
? [{ key: entry.key, text: entry.update }]
: [],
),
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) =>
snapshot.entries.some((entry) => entry.key === key)
? []
: [{ key: Key.make(key), text: `System context component removed: ${key}` }],
),
],
checkpoint: nextCheckpoint(snapshot, previous),
}
}