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

refactor(core): trim v2 context epoch scope

Browse source
This commit is contained in:
Kit Langton 2026-06-04 21:32:37 -04:00
commit 8cb02ba9cf
17 changed files with 118 additions and 412 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -381,10 +381,10 @@ export const layerWith = (options?: LayerOptions) =>
})
}
function publishEvent<D extends Definition>(event: Payload<D>, options?: PublishOptions) {
function publishEvent<D extends Definition>(event: Payload<D>, commit?: PublishOptions["commit"]) {
return Effect.gen(function* () {
const durable = registry.get(event.type)?.sync !== undefined
if (!durable && options?.commit)
if (!durable && commit)
return yield* Effect.die(
new InvalidSyncEventError({
type: event.type,
@ -392,7 +392,7 @@ export const layerWith = (options?: LayerOptions) =>
}),
)
if (durable) {
const committed = yield* commitSyncEvent(event as Payload, undefined, options?.commit)
const committed = yield* commitSyncEvent(event as Payload, undefined, commit)
if (committed) {
event = { ...event, seq: committed.seq }
yield* Effect.forEach(syncHandlers, (sync) => observe(event as Payload, "sync", sync), { discard: true })
@ -446,7 +446,7 @@ export const layerWith = (options?: LayerOptions) =>
...(location ? { location } : {}),
data,
} as Payload<D>,
options,
options?.commit,
)
})
}

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

@ -57,7 +57,7 @@ export class LocationServiceMap extends LayerMap.Service<LocationServiceMap>()("
Catalog.locationLayer,
CommandV2.locationLayer,
AgentV2.locationLayer,
PluginBoot.locationLayer.pipe(Layer.provide(systemContext)),
PluginBoot.locationLayer,
FileSystem.locationLayer,
Watcher.locationLayer,
Pty.locationLayer,

4
packages/core/src/plugin/boot.ts
View file

@ -25,7 +25,6 @@ import { EnvPlugin } from "./env"
import { ModelsDevPlugin } from "./models-dev"
import { ProviderPlugins } from "./provider"
import { SkillV2 } from "../skill"
import { SystemContextRegistry } from "../system-context-registry"
type Plugin = {
id: PluginV2.ID
@ -43,7 +42,6 @@ type Plugin = {
| Config.Service
| ModelsDev.Service
| SkillV2.Service
| SystemContextRegistry.Service
>
}
@ -69,7 +67,6 @@ export const layer = Layer.effect(
const fs = yield* FSUtil.Service
const global = yield* Global.Service
const skill = yield* SkillV2.Service
const systemContext = yield* SystemContextRegistry.Service
const done = yield* Deferred.make<void>()
const add = Effect.fn("PluginBoot.add")(function* (input: Plugin) {
@ -89,7 +86,6 @@ export const layer = Layer.effect(
Effect.provideService(Global.Service, global),
Effect.provideService(SkillV2.Service, skill),
Effect.provideService(PluginV2.Service, plugin),
Effect.provideService(SystemContextRegistry.Service, systemContext),
),
})
})

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

@ -98,13 +98,6 @@ 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,
baselineSeq: number,
) {
return yield* store.runnerContext(sessionID, baselineSeq)
})
const failInterruptedTools = Effect.fn("SessionRunner.failInterruptedTools")(function* (
sessionID: SessionSchema.ID,
) {
@ -152,7 +145,7 @@ export const layer = Layer.effect(
}
}
const system = initialized ?? (yield* SessionContextEpoch.prepare(db, events, systemContext, session.id))
const context = yield* getRunnerContext(session.id, system.baselineSeq)
const context = yield* store.runnerContext(session.id, system.baselineSeq)
const request = LLM.request({
model,
system: system.baseline.length > 0 ? [SystemPart.make(system.baseline)] : [],
@ -251,7 +244,6 @@ export const layer = Layer.effect(
readonly sessionID: SessionSchema.ID
readonly force?: boolean
}) {
const session = yield* getSession(input.sessionID)
const hasSteer = yield* SessionInput.hasPending(db, input.sessionID, "steer")
const hasQueue = hasSteer ? false : yield* SessionInput.hasPending(db, input.sessionID, "queue")
if (input.force !== true && !hasSteer && !hasQueue) return
@ -261,7 +253,7 @@ export const layer = Layer.effect(
while (openActivity) {
let needsContinuation = true
for (let step = 0; step < MAX_STEPS; step++) {
needsContinuation = yield* runTurn(session.id, promotion)
needsContinuation = yield* runTurn(input.sessionID, promotion)
promotion = "steer"
if (!needsContinuation) needsContinuation = yield* SessionInput.hasPending(db, input.sessionID, "steer")
if (!needsContinuation) break

153
packages/core/test/database-migration.test.ts
View file

@ -1,4 +1,4 @@
import { describe, expect } from "bun:test"
import { describe, expect, test } from "bun:test"
import { $ } from "bun"
import { fileURLToPath } from "url"
import path from "path"
@ -18,45 +18,43 @@ import { AbsolutePath } from "@opencode-ai/core/schema"
import { SessionSchema } from "@opencode-ai/core/session/schema"
import { SessionTable } from "@opencode-ai/core/session/sql"
import sessionMetadataMigration from "@opencode-ai/core/database/migration/20260511173437_session-metadata"
import type { SqlClient as SqlClientService } from "effect/unstable/sql/SqlClient"
import { Database } from "@opencode-ai/core/database/database"
import { tmpdir } from "./fixture/tmpdir"
import { testEffect } from "./lib/effect"
const run = <A, E>(effect: Effect.Effect<A, E, SqlClientService>) =>
Effect.runPromise(
effect.pipe(Effect.provide(SqliteClient.layer({ filename: ":memory:", disableWAL: true })), Effect.scoped),
)
const makeDb = EffectDrizzleSqlite.makeWithDefaults()
const it = testEffect(SqliteClient.layer({ filename: ":memory:", disableWAL: true }))
describe("DatabaseMigration", () => {
it.effect("serializes concurrent embedded initialization for one database path", () =>
Effect.promise(async () => {
await using tmp = await tmpdir()
const filename = path.join(tmp.path, "embedded.sqlite")
const layers = [Database.layerFromPath(filename), Database.layerFromPath(filename)]
test("serializes concurrent embedded initialization for one database path", async () => {
await using tmp = await tmpdir()
const filename = path.join(tmp.path, "embedded.sqlite")
const layers = [Database.layerFromPath(filename), Database.layerFromPath(filename)]
await Effect.runPromise(
Effect.all(
layers.map((layer) => Effect.scoped(Layer.build(layer))),
{ concurrency: "unbounded" },
),
)
}),
)
if (process.platform === "linux") {
it.effect(
"declared schema has no ungenerated migrations",
() =>
Effect.promise(async () => {
const result = await $`bun ${fileURLToPath(new URL("../script/migration.ts", import.meta.url))} --check`
.quiet()
.nothrow()
expect(result.exitCode, result.stderr.toString()).toBe(0)
expect(result.stdout.toString()).toContain("No schema changes, nothing to migrate")
}),
30_000,
await Effect.runPromise(
Effect.all(
layers.map((layer) => Effect.scoped(Layer.build(layer))),
{ concurrency: "unbounded" },
),
)
})
if (process.platform === "linux") {
test("declared schema has no ungenerated migrations", async () => {
const result = await $`bun ${fileURLToPath(new URL("../script/migration.ts", import.meta.url))} --check`
.quiet()
.nothrow()
expect(result.exitCode, result.stderr.toString()).toBe(0)
expect(result.stdout.toString()).toContain("No schema changes, nothing to migrate")
}, 30_000)
}
it.effect("applies tracked migrations to an empty database", () =>
Effect.gen(function* () {
test("applies tracked migrations to an empty database", async () => {
await run(
Effect.gen(function* () {
const db = yield* makeDb
yield* DatabaseMigration.apply(db)
@ -84,11 +82,13 @@ describe("DatabaseMigration", () => {
{ name: "session_message_session_time_created_id_idx" },
{ name: "session_message_session_type_seq_idx" },
])
}),
)
}),
)
})
it.effect("resets beta history and rebuilds event-sourced Session input storage", () =>
Effect.gen(function* () {
test("resets beta history and rebuilds event-sourced Session input storage", async () => {
await run(
Effect.gen(function* () {
const db = yield* makeDb
yield* db.run(sql`CREATE TABLE session (id text PRIMARY KEY, workspace_id text)`)
yield* db.run(sql`CREATE TABLE workspace (id text PRIMARY KEY)`)
@ -158,11 +158,13 @@ describe("DatabaseMigration", () => {
expect.objectContaining({ name: "session_input_session_promoted_seq_idx", unique: 1 }),
expect.objectContaining({ name: "session_input_session_admitted_seq_idx", unique: 1 }),
])
}),
)
}),
)
})
it.effect("resets incompatible projected Session messages before adding sequence order", () =>
Effect.gen(function* () {
test("resets incompatible projected Session messages before adding sequence order", async () => {
await run(
Effect.gen(function* () {
const db = yield* makeDb
yield* db.run(sql`CREATE TABLE session (id text PRIMARY KEY)`)
yield* db.run(
@ -211,11 +213,13 @@ describe("DatabaseMigration", () => {
sql`INSERT INTO session_message (id, session_id, type, seq, time_created, time_updated, data) VALUES ('fresh_projection', 'session', 'user', 7, 2, 2, '{}')`,
)
expect(yield* db.get(sql`SELECT id, seq FROM session_message`)).toEqual({ id: "fresh_projection", seq: 7 })
}),
)
}),
)
})
it.effect("runs session usage backfill in order with schema changes", () =>
Effect.gen(function* () {
test("runs session usage backfill in order with schema changes", async () => {
await run(
Effect.gen(function* () {
const db = yield* makeDb
yield* db.run(sql`CREATE TABLE session (id text PRIMARY KEY, time_updated integer NOT NULL)`)
yield* db.run(sql`CREATE TABLE message (id text PRIMARY KEY, session_id text NOT NULL, data text NOT NULL)`)
@ -238,11 +242,13 @@ describe("DatabaseMigration", () => {
tokens_cache_read: 5,
tokens_cache_write: 6,
})
}),
)
}),
)
})
it.effect("normalizes Windows storage paths and leaves POSIX paths untouched", () =>
Effect.gen(function* () {
test("normalizes Windows storage paths and leaves POSIX paths untouched", async () => {
await run(
Effect.gen(function* () {
const db = yield* makeDb
yield* db.run(sql`CREATE TABLE project (id text PRIMARY KEY, worktree text NOT NULL, sandboxes text NOT NULL)`)
yield* db.run(sql`CREATE TABLE session (id text PRIMARY KEY, directory text NOT NULL, path text)`)
@ -287,12 +293,14 @@ describe("DatabaseMigration", () => {
directory: "/home/me/we\\ird",
path: "src\\weird",
})
}),
)
}),
)
})
it.effect("maps native Windows paths through database columns", () => {
if (process.platform !== "win32") return Effect.void
return Effect.gen(function* () {
test("maps native Windows paths through database columns", async () => {
if (process.platform !== "win32") return
await run(
Effect.gen(function* () {
const db = yield* makeDb
yield* DatabaseMigration.apply(db)
const projectID = ProjectV2.ID.make("codec_project")
@ -395,11 +403,13 @@ describe("DatabaseMigration", () => {
expect(() =>
Effect.runSync(db.select().from(ProjectTable).where(eq(ProjectTable.id, projectID)).get()),
).toThrow()
})
}),
)
})
it.effect("imports existing drizzle migration state", () =>
Effect.gen(function* () {
test("imports existing drizzle migration state", async () => {
await run(
Effect.gen(function* () {
const db = yield* makeDb
yield* db.run(
sql`CREATE TABLE __drizzle_migrations (id INTEGER PRIMARY KEY, hash text NOT NULL, created_at numeric, name text, applied_at TEXT)`,
@ -412,11 +422,13 @@ describe("DatabaseMigration", () => {
yield* DatabaseMigration.applyOnly(db, [])
expect(yield* db.get(sql`SELECT id FROM migration`)).toEqual({ id: "20260127222353_familiar_lady_ursula" })
}),
)
}),
)
})
it.effect("does not replay a migrated session metadata column", () =>
Effect.gen(function* () {
test("does not replay a migrated session metadata column", async () => {
await run(
Effect.gen(function* () {
const db = yield* makeDb
yield* db.run(sql`CREATE TABLE session (id text PRIMARY KEY, metadata text)`)
yield* db.run(
@ -430,11 +442,13 @@ describe("DatabaseMigration", () => {
yield* DatabaseMigration.applyOnly(db, [sessionMetadataMigration])
expect(yield* db.all(sql`SELECT id FROM migration`)).toEqual([{ id: "20260511173437_session-metadata" }])
}),
)
}),
)
})
it.effect("accepts the temporary replacement session metadata migration id", () =>
Effect.gen(function* () {
test("accepts the temporary replacement session metadata migration id", async () => {
await run(
Effect.gen(function* () {
const db = yield* makeDb
yield* db.run(sql`CREATE TABLE session (id text PRIMARY KEY, metadata text)`)
yield* db.run(sql`CREATE TABLE migration (id TEXT PRIMARY KEY, time_completed INTEGER NOT NULL)`)
@ -446,11 +460,13 @@ describe("DatabaseMigration", () => {
{ id: "20260511173437_session-metadata" },
{ id: "20260530232709_lovely_romulus" },
])
}),
)
}),
)
})
it.effect("skips drizzle import when migration table already has state", () =>
Effect.gen(function* () {
test("skips drizzle import when migration table already has state", async () => {
await run(
Effect.gen(function* () {
const db = yield* makeDb
yield* db.run(sql`CREATE TABLE migration (id TEXT PRIMARY KEY, time_completed INTEGER NOT NULL)`)
yield* db.run(sql`INSERT INTO migration (id, time_completed) VALUES ('existing', 1)`)
@ -465,6 +481,7 @@ describe("DatabaseMigration", () => {
yield* DatabaseMigration.applyOnly(db, [])
expect(yield* db.all(sql`SELECT id FROM migration ORDER BY id`)).toEqual([{ id: "existing" }])
}),
)
}),
)
})
})

2
packages/core/test/fixtures/recordings/session-runner/openai-chat-streams-text.json vendored
View file

@ -13,7 +13,7 @@
"headers": {
"content-type": "application/json"
},
"body": "{\"model\":\"gpt-4o-mini\",\"messages\":[{\"role\":\"system\",\"content\":\"Recorded context\"},{\"role\":\"user\",\"content\":\"Say hello in one short sentence.\"}],\"stream\":true,\"stream_options\":{\"include_usage\":true},\"max_tokens\":20,\"temperature\":0}"
"body": "{\"model\":\"gpt-4o-mini\",\"messages\":[{\"role\":\"user\",\"content\":\"Say hello in one short sentence.\"}],\"stream\":true,\"stream_options\":{\"include_usage\":true},\"max_tokens\":20,\"temperature\":0}"
},
"response": {
"status": 200,

60
packages/core/test/session-runner-message.test.ts
View file

@ -1,4 +1,4 @@
import { describe, expect } from "bun:test"
import { describe, expect, test } from "bun:test"
import { Message, Model } from "@opencode-ai/llm"
import * as OpenAIChat from "@opencode-ai/llm/protocols/openai-chat"
import { ModelV2 } from "@opencode-ai/core/model"
@ -8,16 +8,14 @@ import { AgentAttachment, FileAttachment, ReferenceAttachment } from "@opencode-
import { toLLMMessages } from "@opencode-ai/core/session/runner/to-llm-message"
import { SessionV2 } from "@opencode-ai/core/session"
import { ToolOutput } from "@opencode-ai/core/tool-output"
import { DateTime, Effect } from "effect"
import { it } from "./lib/effect"
import { DateTime } from "effect"
const created = DateTime.makeUnsafe(0)
const id = (value: string) => SessionMessage.ID.make(`msg_${value}`)
const model = Model.make({ id: "model", provider: "provider", route: OpenAIChat.route })
describe("toLLMMessages", () => {
it.effect("maps every top-level V2 Session message type", () =>
Effect.sync(() => {
test("maps every top-level V2 Session message type", () => {
const file = new FileAttachment({ uri: "data:image/png;base64,aGVsbG8=", mime: "image/png", name: "hello.png" })
const reference = new ReferenceAttachment({ name: "docs", kind: "local", uri: "file:///docs" })
const messages = toLLMMessages(
@ -34,6 +32,12 @@ describe("toLLMMessages", () => {
model: { id: ModelV2.ID.make("model"), providerID: ProviderV2.ID.make("provider") },
time: { created },
}),
new SessionMessage.System({
id: id("system"),
type: "system",
text: "Updated context\n\nOther context",
time: { created },
}),
new SessionMessage.User({
id: id("user"),
type: "user",
@ -69,8 +73,9 @@ describe("toLLMMessages", () => {
model,
)
expect(messages.map((message) => message.role)).toEqual(["user", "user", "user", "user"])
expect(messages[0]).toEqual(
expect(messages.map((message) => message.role)).toEqual(["system", "user", "user", "user", "user"])
expect(messages[0]).toEqual(Message.system("Updated context\n\nOther context"))
expect(messages[1]).toEqual(
Message.make({
id: id("user"),
role: "user",
@ -81,34 +86,14 @@ describe("toLLMMessages", () => {
metadata: { agents: [{ name: "build" }], references: [reference] },
}),
)
expect(messages.slice(1).map((message) => message.content)).toEqual([
expect(messages.slice(2).map((message) => message.content)).toEqual([
[{ type: "text", text: "Synthetic context" }],
[{ type: "text", text: "Shell command: pwd\n\n/project" }],
[{ type: "text", text: "Summary of earlier conversation:\nEarlier work" }],
])
}),
)
})
it.effect("maps durable Session system messages into chronological system messages", () =>
Effect.sync(() => {
expect(
toLLMMessages(
[
new SessionMessage.System({
id: id("system"),
type: "system",
text: "Updated context\n\nOther context",
time: { created },
}),
],
model,
),
).toEqual([Message.system("Updated context\n\nOther context")])
}),
)
it.effect("expands assistant tool calls and settled outcomes into canonical tool messages", () =>
Effect.sync(() => {
test("expands assistant tool calls and settled outcomes into canonical tool messages", () => {
const messages = toLLMMessages(
[
new SessionMessage.Assistant({
@ -263,11 +248,9 @@ describe("toLLMMessages", () => {
},
},
])
}),
)
})
it.effect("restores OpenAI encrypted reasoning metadata", () =>
Effect.sync(() => {
test("restores OpenAI encrypted reasoning metadata", () => {
const messages = toLLMMessages(
[
new SessionMessage.Assistant({
@ -296,11 +279,9 @@ describe("toLLMMessages", () => {
providerMetadata: { openai: { itemId: "rs_1", reasoningEncryptedContent: "encrypted-state" } },
},
])
}),
)
})
it.effect("drops provider-native continuation metadata after a model switch", () =>
Effect.sync(() => {
test("drops provider-native continuation metadata after a model switch", () => {
const messages = toLLMMessages(
[
new SessionMessage.Assistant({
@ -398,6 +379,5 @@ describe("toLLMMessages", () => {
providerMetadata: undefined,
},
])
}),
)
})
})

24
packages/core/test/session-runner-recorded.test.ts
View file

@ -19,11 +19,10 @@ import { SessionRunnerModel } from "@opencode-ai/core/session/runner/model"
import { ToolRegistry } from "@opencode-ai/core/tool/registry"
import { SessionTable } from "@opencode-ai/core/session/sql"
import { SessionStore } from "@opencode-ai/core/session/store"
import { SystemContext } from "@opencode-ai/core/system-context"
import { SystemContextRegistry } from "@opencode-ai/core/system-context-registry"
import { describe, expect } from "bun:test"
import { eq } from "drizzle-orm"
import { Effect, Layer, Schema } from "effect"
import { Effect, Layer } from "effect"
import path from "node:path"
import { testEffect } from "./lib/effect"
@ -57,26 +56,7 @@ const model = OpenAIChat.route
})
.model({ id: "gpt-4o-mini" })
const models = SessionRunnerModel.layerWith(() => Effect.succeed(model))
const systemContextKey = SystemContext.Key.make("test/context")
const systemContext = Layer.effectDiscard(
SystemContextRegistry.Service.pipe(
Effect.flatMap((registry) =>
registry.contribute({
key: systemContextKey,
load: Effect.succeed(
SystemContext.make({
key: systemContextKey,
codec: Schema.toCodecJson(Schema.String),
load: Effect.succeed("Recorded context"),
baseline: String,
update: (_previous, current) => current,
removed: () => "Recorded context removed",
}),
),
}),
),
),
).pipe(Layer.provideMerge(SystemContextRegistry.layer))
const systemContext = SystemContextRegistry.layer
const runner = SessionRunnerLLM.defaultLayer.pipe(
Layer.provide(database),
Layer.provide(store),

34
packages/core/test/session-runner.test.ts
View file

@ -823,40 +823,6 @@ describe("SessionRunnerLLM", () => {
}),
)
it.effect("replays retained context projections after multiple replacements", () =>
Effect.gen(function* () {
yield* setup
const session = yield* SessionV2.Service
const events = yield* EventV2.Service
yield* session.prompt({ sessionID, prompt: new Prompt({ text: "First" }), resume: false })
requests.length = 0
response = []
yield* session.resume(sessionID)
yield* events.publish(SessionEvent.ModelSwitched, {
sessionID,
messageID: SessionMessage.ID.create(),
timestamp: DateTime.makeUnsafe(1),
model: { id: ModelV2.ID.make("replacement-1"), providerID: ProviderV2.ID.make("fake") },
})
systemBaseline = "Replacement context 1"
yield* session.prompt({ sessionID, prompt: new Prompt({ text: "Second" }), resume: false })
yield* session.resume(sessionID)
yield* events.publish(SessionEvent.ModelSwitched, {
sessionID,
messageID: SessionMessage.ID.create(),
timestamp: DateTime.makeUnsafe(2),
model: { id: ModelV2.ID.make("replacement-2"), providerID: ProviderV2.ID.make("fake") },
})
systemBaseline = "Replacement context 2"
yield* session.prompt({ sessionID, prompt: new Prompt({ text: "Third" }), resume: false })
yield* session.resume(sessionID)
yield* replaySessionProjection(sessionID)
expect(requests.at(-1)?.system.map((part) => part.text)).toEqual(["Replacement context 2"])
}),
)
it.effect("replaces the baseline lazily after completed compaction without reopening replacement on replay", () =>
Effect.gen(function* () {
yield* setup