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

feat(core): add transient session generation (#37330)

Browse source
This commit is contained in:
Kit Langton 2026-07-16 14:28:54 -04:00 • committed by GitHub
commit e4a16830f1
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
9 changed files with 687 additions and 79 deletions
Download patch file Download diff file Expand all files Collapse all files

69
packages/core/src/session/generate-node.ts Normal file
View file

@ -0,0 +1,69 @@
export * as SessionGenerateNode from "./generate-node"
import { LLM, LLMClient, Message, SystemPart } from "@opencode-ai/ai"
import { Effect, Layer } from "effect"
import { Database } from "../database/database"
import { makeLocationNode } from "../effect/app-node"
import { llmClient } from "../effect/app-node-platform"
import { PluginHooks } from "../plugin/hooks"
import { SessionContext } from "./context"
import { SessionGenerate } from "./generate"
import { SessionHistory } from "./history"
import { SessionModelHeaders } from "./model-headers"
import { SessionRunnerModel } from "./runner/model"
import PROMPT_DEFAULT from "./runner/prompt/base.txt"
import { toLLMMessages } from "./runner/to-llm-message"
const layer = Layer.effect(
SessionGenerate.Service,
Effect.gen(function* () {
const context = yield* SessionContext.Service
const database = yield* Database.Service
const hooks = yield* PluginHooks.Service
const llm = yield* LLMClient.Service
const models = yield* SessionRunnerModel.Service
return SessionGenerate.Service.of({
generate: Effect.fn("SessionGenerate.generate")(function* (input) {
const selection = yield* context.select(input.sessionID)
const model = yield* models.resolve(selection.session)
const history = yield* SessionHistory.preview(database.db, selection.session.id, selection.instructions)
const providerMetadataKey = model.model.route.providerMetadataKey ?? model.model.provider
const promptCacheKey = /^ses_[0-9a-f]{64}$/.test(selection.session.id)
? selection.session.id.slice(4)
: selection.session.id
const contextEvent = yield* hooks.trigger("session", "context", {
sessionID: selection.session.id,
agent: selection.agent.id,
model: model.ref,
system: [selection.agent.info.system ? selection.agent.info.system : PROMPT_DEFAULT, history.initial]
.filter((part) => part.length > 0)
.map(SystemPart.make),
messages: [
...toLLMMessages(history.messages, model.ref, providerMetadataKey),
...(history.instructionUpdate ? [Message.system(history.instructionUpdate)] : []),
Message.user(input.prompt),
],
tools: {},
})
return (yield* llm.generate(
LLM.request({
model: model.model,
http: { headers: SessionModelHeaders.make(selection.session) },
providerOptions: { openai: { promptCacheKey } },
system: contextEvent.system,
messages: contextEvent.messages,
tools: [],
toolChoice: "none",
}),
)).text
}),
})
}),
)
export const node = makeLocationNode({
service: SessionGenerate.Service,
layer,
deps: [SessionContext.node, Database.node, PluginHooks.node, SessionRunnerModel.node, llmClient],
})

21
packages/core/src/session/generate.ts Normal file
View file

@ -0,0 +1,21 @@
export * as SessionGenerate from "./generate"
import type { LLMError } from "@opencode-ai/ai"
import { Context, type Effect } from "effect"
import type { Instructions } from "../instructions"
import type { AgentNotFoundError } from "./error"
import type { SessionRunnerModel } from "./runner/model"
import type { SessionSchema } from "./schema"
export type Error = AgentNotFoundError | Instructions.InitializationBlocked | SessionRunnerModel.Error | LLMError
export interface Interface {
/** Generates text from current Session context without mutating the Session. */
readonly generate: (input: {
readonly sessionID: SessionSchema.ID
readonly prompt: string
}) => Effect.Effect<string, Error>
}
/** Location-scoped transient generation from Session context. */
export class Service extends Context.Service<Service, Interface>()("@opencode/v2/SessionGenerate") {}

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

@ -60,13 +60,17 @@ const decodeMessageRow = (row: typeof SessionMessageTable.$inferSelect) =>
),
)
export const load = Effect.fn("SessionHistory.load")(function* (db: DatabaseService, sessionID: SessionSchema.ID) {
return yield* Effect.forEach(
yield* messageRows(db, sessionID, yield* latestCompaction(db, sessionID)),
decodeMessageRow,
const messageEntries = Effect.fnUntraced(function* (db: DatabaseService, sessionID: SessionSchema.ID) {
const rows = yield* messageRows(db, sessionID, yield* latestCompaction(db, sessionID))
return yield* Effect.forEach(rows, (row) =>
decodeMessageRow(row).pipe(Effect.map((message) => ({ seq: row.seq, message }))),
)
})
export const load = Effect.fn("SessionHistory.load")(function* (db: DatabaseService, sessionID: SessionSchema.ID) {
return (yield* messageEntries(db, sessionID)).map((entry) => entry.message)
})
export const entriesForRunner = Effect.fn("SessionHistory.entriesForRunner")(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
@ -75,10 +79,7 @@ export const entriesForRunner = Effect.fn("SessionHistory.entriesForRunner")(fun
return yield* db
.transaction(() =>
Effect.gen(function* () {
const rows = yield* messageRows(db, sessionID, yield* latestCompaction(db, sessionID))
const messages = yield* Effect.forEach(rows, (row) =>
decodeMessageRow(row).pipe(Effect.map((message) => ({ seq: row.seq, message }))),
)
const messages = yield* messageEntries(db, sessionID)
const assembled = yield* InstructionState.assemble(db, sessionID, instructions)
return {
initial: assembled.initial,
@ -89,6 +90,33 @@ export const entriesForRunner = Effect.fn("SessionHistory.entriesForRunner")(fun
.pipe(Effect.orDie)
})
export const preview = Effect.fn("SessionHistory.preview")(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
instructions: Instructions.Instructions,
) {
const observed = yield* Instructions.read(instructions)
return yield* db
.transaction(() =>
Effect.gen(function* () {
const messages = yield* messageEntries(db, sessionID)
// An active assistant may contain an unresolved tool call, so only preview the settled prefix.
const unsettled = messages.findIndex(
(entry) => entry.message.type === "assistant" && entry.message.time.completed === undefined,
)
const settled = unsettled === -1 ? messages : messages.slice(0, unsettled)
const assembled = yield* InstructionState.preview(db, sessionID, instructions, observed)
const entries = [...settled, ...assembled.updates].toSorted((a, b) => a.seq - b.seq)
return {
initial: assembled.initial,
messages: entries.map((entry) => entry.message),
instructionUpdate: assembled.update,
}
}),
)
.pipe(Effect.catch((error) => (error instanceof Instructions.InitializationBlocked ? error : Effect.die(error))))
})
/** Returns the session's sole user message, or `undefined` once a second one exists. */
export const firstUserMessageIfOnly = Effect.fn("SessionHistory.firstUserMessageIfOnly")(function* (
db: DatabaseService,

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

@ -29,12 +29,11 @@ export const observe = Effect.fn("InstructionState.observe")(function* (
const [observed, stored] = yield* Effect.all([Instructions.read(instructions), ensure(db, sessionID)], {
concurrency: "unbounded",
})
const admission = yield* Instructions.diff(observed, stored?.current_values)
const result = yield* observeAgainst(observed, stored?.current_values)
return {
sessionID,
initial: !stored,
current: Instructions.applyHashDelta(stored?.current_values ?? {}, admission.delta),
...admission,
...result,
}
})
@ -123,28 +122,21 @@ export const rebuild = Effect.fn("InstructionState.rebuild")(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
) {
const folded = fold(yield* instructionEvents(db, sessionID))
if (!folded) {
const state = yield* stateFromEvents(db, sessionID)
if (!state) {
yield* reset(db, sessionID)
return undefined
}
const state = {
session_id: sessionID,
epoch_start: folded.epochStart,
through_seq: folded.throughSeq,
initial_values: folded.initial,
current_values: folded.current,
}
yield* db
.insert(InstructionStateTable)
.values(state)
.onConflictDoUpdate({
target: InstructionStateTable.session_id,
set: {
epoch_start: folded.epochStart,
through_seq: folded.throughSeq,
initial_values: folded.initial,
current_values: folded.current,
epoch_start: state.epoch_start,
through_seq: state.through_seq,
initial_values: state.initial_values,
current_values: state.current_values,
},
})
.run()
@ -152,13 +144,12 @@ export const rebuild = Effect.fn("InstructionState.rebuild")(function* (
return state
})
export const assemble = Effect.fn("InstructionState.assemble")(function* (
const assembleState = Effect.fnUntraced(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
instructions: Instructions.Instructions,
state: typeof InstructionStateTable.$inferSelect,
) {
const state = yield* find(db, sessionID)
if (!state) return yield* Effect.die(new Error(`Instruction state not found during assembly: ${sessionID}`))
const rows = yield* instructionUpdatesAfter(db, sessionID, state.epoch_start)
const updates = rows.map((row) => ({
row,
@ -188,7 +179,49 @@ export const assemble = Effect.fn("InstructionState.assemble")(function* (
})
values = Instructions.applyDelta(values, delta)
}
return { initial: Instructions.renderInitial(instructions, valuesAtStart), updates: result }
return { initial: Instructions.renderInitial(instructions, valuesAtStart), updates: result, current: values }
})
export const assemble = Effect.fn("InstructionState.assemble")(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
instructions: Instructions.Instructions,
) {
const state = yield* find(db, sessionID)
if (!state) return yield* Effect.die(new Error(`Instruction state not found during assembly: ${sessionID}`))
const assembled = yield* assembleState(db, sessionID, instructions, state)
return { initial: assembled.initial, updates: assembled.updates }
})
export const preview = Effect.fn("InstructionState.preview")(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
instructions: Instructions.Instructions,
observed: Instructions.ReadResult,
) {
const state = yield* readState(db, sessionID)
const result = yield* observeAgainst(observed, state?.current_values)
const blobs = new Map<Instructions.Hash, Schema.Json>(
Object.entries(result.blobs).map(([hash, value]) => [Instructions.Hash.make(hash), value]),
)
if (!state) {
const values = dereference(result.current, blobs)
return { initial: Instructions.renderInitial(instructions, values), updates: [], update: "" }
}
const assembled = yield* assembleState(db, sessionID, instructions, state)
return {
initial: assembled.initial,
updates: assembled.updates,
update: Instructions.renderUpdate(instructions, assembled.current, dereferenceDelta(result.delta, blobs)),
}
})
const observeAgainst = Effect.fnUntraced(function* (observed: Instructions.ReadResult, previous?: Instructions.Values) {
const admission = yield* Instructions.diff(observed, previous)
return {
current: Instructions.applyHashDelta(previous ?? {}, admission.delta),
...admission,
}
})
const find = Effect.fnUntraced(function* (db: DatabaseService, sessionID: SessionSchema.ID) {
@ -203,7 +236,26 @@ const find = Effect.fnUntraced(function* (db: DatabaseService, sessionID: Sessio
const ensure = Effect.fnUntraced(function* (db: DatabaseService, sessionID: SessionSchema.ID) {
const stored = yield* find(db, sessionID)
if (!stored) return yield* rebuild(db, sessionID)
const latest = yield* db
const latest = yield* latestRelevantSequence(db, sessionID)
if (!latest || latest.seq <= stored.through_seq) return stored
return yield* rebuild(db, sessionID)
})
const readState = Effect.fnUntraced(function* (db: DatabaseService, sessionID: SessionSchema.ID) {
const stored = yield* find(db, sessionID)
if (!stored) return yield* stateFromEvents(db, sessionID)
const latest = yield* latestRelevantSequence(db, sessionID)
if (!latest || latest.seq <= stored.through_seq) return stored
return yield* stateFromEvents(db, sessionID)
})
const stateFromEvents = Effect.fnUntraced(function* (db: DatabaseService, sessionID: SessionSchema.ID) {
const folded = fold(yield* instructionEvents(db, sessionID))
return folded ? foldedState(sessionID, folded) : undefined
})
const latestRelevantSequence = Effect.fnUntraced(function* (db: DatabaseService, sessionID: SessionSchema.ID) {
return yield* db
.select({ seq: EventTable.seq })
.from(EventTable)
.where(and(eq(EventTable.aggregate_id, sessionID), inArray(EventTable.type, relevantEventTypes)))
@ -211,8 +263,6 @@ const ensure = Effect.fnUntraced(function* (db: DatabaseService, sessionID: Sess
.limit(1)
.get()
.pipe(Effect.orDie)
if (!latest || latest.seq <= stored.through_seq) return stored
return yield* rebuild(db, sessionID)
})
const insertBlobs = Effect.fnUntraced(function* (db: DatabaseService, blobs: Readonly<Record<string, Schema.Json>>) {
@ -364,3 +414,13 @@ function fold(rows: ReadonlyArray<InstructionEventRow>) {
: { epochStart: row.seq, throughSeq: row.seq, initial: current, current }
}, undefined)
}
function foldedState(sessionID: SessionSchema.ID, folded: NonNullable<ReturnType<typeof fold>>) {
return {
session_id: sessionID,
epoch_start: folded.epochStart,
through_seq: folded.throughSeq,
initial_values: folded.initial,
current_values: folded.current,
}
}