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

refactor(core): dedupe context epoch initialization

Browse source 
Share one insertInitial path between initialize and prepare, and
document the runner's ordering constraint: epoch initialization runs
before promotion so a blocked System Context leaves prompts pending,
while reconciliation runs after promotion so ContextUpdated messages
sequence after promoted input.
This commit is contained in:
Kit Langton 2026-07-02 20:56:56 -04:00
commit fae51d853d
2 changed files with 15 additions and 8 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -8,7 +8,6 @@ import { SystemContext } from "../system-context/index"
import { ContextSnapshotDecodeError } from "./error"
import { SessionEvent } from "./event"
import { SessionHistory } from "./history"
import { SessionInput } from "./input"
import { SessionMessage } from "./message"
import { SessionSchema } from "./schema"
import { SessionContextEpochTable } from "./sql"
@ -47,11 +46,7 @@ const prepareOnce = Effect.fnUntraced(function* (
[context, find(db, sessionID), SessionHistory.latestCompaction(db, sessionID)],
{ concurrency: "unbounded" },
)
if (!stored) {
const generation = yield* SystemContext.initialize(value)
const baselineSeq = yield* insert(db, sessionID, generation)
return { baseline: generation.baseline, baselineSeq }
}
if (!stored) return yield* insertInitial(db, sessionID, value)
const snapshot = yield* Schema.decodeUnknownEffect(SystemContext.Snapshot)(stored.snapshot).pipe(
Effect.mapError((error) => new ContextSnapshotDecodeError({ sessionID, details: String(error) })),
@ -82,8 +77,16 @@ const initializeOnce = Effect.fnUntraced(function* (
context: Effect.Effect<SystemContext.SystemContext>,
sessionID: SessionSchema.ID,
) {
if (yield* exists(db, sessionID)) return
const generation = yield* context.pipe(Effect.flatMap(SystemContext.initialize))
if (yield* exists(db, sessionID)) return undefined
return yield* insertInitial(db, sessionID, yield* context)
})
const insertInitial = Effect.fnUntraced(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
value: SystemContext.SystemContext,
) {
const generation = yield* SystemContext.initialize(value)
const baselineSeq = yield* insert(db, sessionID, generation)
return { baseline: generation.baseline, baselineSeq }
})

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

@ -168,6 +168,10 @@ const layer = Layer.effect(
if (session.location.directory !== location.directory || session.location.workspaceID !== location.workspaceID)
return yield* Effect.interrupt
const agent = yield* agents.select(session.agent)
// Epoch initialization runs before promotion so a blocked System Context fails the
// turn while prompts are still pending in the durable inbox; a retry re-drains them.
// Reconciliation (`prepare` below) runs after promotion so ContextUpdated messages
// sequence after the promoted user input.
const initialized = yield* SessionContextEpoch.initialize(db, loadSystemContext(agent), session.id)
const toolFibers = yield* FiberSet.make<void, ToolOutputStore.Error>()
let needsContinuation = false