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

feat(core): add durable compaction barrier (#35371)

Browse source
This commit is contained in:
Kit Langton 2026-07-06 21:26:05 -04:00 • committed by GitHub
commit 04b673432c
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
39 changed files with 1486 additions and 396 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -16,6 +16,7 @@ import contextEpochAgentMigration from "@opencode-ai/core/database/migration/202
import simplifyIntegrationCredentialsMigration from "@opencode-ai/core/database/migration/20260611192811_lush_chimera"
import simplifySessionInputMigration from "@opencode-ai/core/database/migration/20260622202450_simplify_session_input"
import resetSessionEventsMigration from "@opencode-ai/core/database/migration/20260703200000_reset_v2_session_events"
import durableSessionInboxMigration from "@opencode-ai/core/database/migration/20260707010146_durable_session_inbox"
import renameInstructionsMigration from "@opencode-ai/core/database/migration/20260705180000_rename_instructions"
import addSessionForkMigration from "@opencode-ai/core/database/migration/20260706223930_add-session-fork"
import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
@ -109,13 +110,14 @@ describe("DatabaseMigration", () => {
expect(yield* db.get(sql`SELECT count(*) as count FROM migration`)).toEqual({ count: migrations.length })
expect(
yield* db.all(
sql`SELECT name FROM sqlite_master WHERE type = 'index' AND name IN ('event_aggregate_seq_idx', 'event_aggregate_type_seq_idx', 'session_input_session_pending_seq_idx', 'session_input_session_pending_delivery_seq_idx', 'session_input_session_admitted_seq_idx', 'session_input_session_promoted_seq_idx', 'session_message_session_idx', 'session_message_session_type_idx', 'session_message_session_seq_idx', 'session_message_session_type_seq_idx', 'session_message_session_time_created_id_idx') ORDER BY name`,
sql`SELECT name FROM sqlite_master WHERE type = 'index' AND name IN ('event_aggregate_seq_idx', 'event_aggregate_type_seq_idx', 'session_input_session_pending_seq_idx', 'session_input_session_pending_delivery_seq_idx', 'session_input_session_pending_type_delivery_seq_idx', 'session_input_session_pending_compaction_idx', 'session_input_session_admitted_seq_idx', 'session_input_session_promoted_seq_idx', 'session_message_session_idx', 'session_message_session_type_idx', 'session_message_session_seq_idx', 'session_message_session_type_seq_idx', 'session_message_session_time_created_id_idx') ORDER BY name`,
),
).toEqual([
{ name: "event_aggregate_seq_idx" },
{ name: "event_aggregate_type_seq_idx" },
{ name: "session_input_session_admitted_seq_idx" },
{ name: "session_input_session_pending_delivery_seq_idx" },
{ name: "session_input_session_pending_compaction_idx" },
{ name: "session_input_session_pending_type_delivery_seq_idx" },
{ name: "session_input_session_promoted_seq_idx" },
{ name: "session_message_session_seq_idx" },
{ name: "session_message_session_time_created_id_idx" },
@ -353,7 +355,7 @@ describe("DatabaseMigration", () => {
sql`INSERT INTO event (id, aggregate_id, seq, type, data, created) VALUES ('event', 'session', 9, 'session.updated.1', '{}', 1)`,
)
yield* db.run(
sql`INSERT INTO session_input (id, session_id, prompt, delivery, admitted_seq, time_created) VALUES ('input', 'session', '{}', 'steer', 9, 1)`,
sql`INSERT INTO session_input (id, session_id, type, prompt, delivery, admitted_seq, time_created) VALUES ('input', 'session', 'prompt', '{}', 'steer', 9, 1)`,
)
yield* db.run(
sql`INSERT INTO session_message (id, session_id, type, seq, time_created, time_updated, data) VALUES ('projected', 'session', 'user', 9, 1, 1, '{}')`,
@ -416,6 +418,37 @@ describe("DatabaseMigration", () => {
)
})
test("preserves admitted prompts while generalizing the durable inbox", async () => {
await run(
Effect.gen(function* () {
const db = yield* makeDb
yield* db.run(
sql`CREATE TABLE session_input (id text PRIMARY KEY, session_id text NOT NULL, prompt text NOT NULL, delivery text NOT NULL, admitted_seq integer NOT NULL, promoted_seq integer, time_created integer NOT NULL)`,
)
yield* db.run(
sql`INSERT INTO session_input (id, session_id, prompt, delivery, admitted_seq, promoted_seq, time_created) VALUES ('input', 'session', '{"text":"hello"}', 'steer', 4, NULL, 1)`,
)
yield* DatabaseMigration.applyOnly(db, [durableSessionInboxMigration])
expect(
yield* db.all(
sql`SELECT id, type, prompt, delivery, admitted_seq, promoted_seq FROM session_input ORDER BY admitted_seq`,
),
).toEqual([
{
id: "input",
type: "prompt",
prompt: '{"text":"hello"}',
delivery: "steer",
admitted_seq: 4,
promoted_seq: null,
},
])
}),
)
})
test("resets incompatible projected Session messages before adding sequence order", async () => {
await run(
Effect.gen(function* () {

23
packages/core/test/session-compact.test.ts
View file

@ -75,7 +75,7 @@ const it = testEffect(
)
describe("SessionV2.compact", () => {
it.effect("manually compacts the active session context", () =>
it.effect("durably admits and coalesces manual compaction", () =>
Effect.gen(function* () {
requests = []
const session = yield* SessionV2.Service
@ -95,13 +95,22 @@ describe("SessionV2.compact", () => {
inputID: messageID,
})
yield* session.compact({ sessionID: created.id })
expect(yield* session.compact({ id: messageID, sessionID: created.id }).pipe(Effect.flip)).toMatchObject({
_tag: "Session.CompactionConflictError",
inputID: messageID,
})
const first = yield* session.compact({ sessionID: created.id })
const second = yield* session.compact({ sessionID: created.id })
expect(requests).toHaveLength(1)
expect(JSON.stringify(requests[0]?.messages)).toContain("Please compact this session history.")
expect(yield* session.context(created.id)).toMatchObject([
{ type: "compaction", reason: "manual", summary: "manual session summary", recent: "" },
])
expect(second.id).toBe(first.id)
expect(requests).toHaveLength(0)
expect((yield* session.context(created.id)).find((message) => message.id === first.id)).toMatchObject({
type: "compaction",
status: "queued",
reason: "manual",
summary: "",
recent: "",
})
}),
)
})

7
packages/core/test/session-prompt.test.ts
View file

@ -246,7 +246,9 @@ describe("SessionV2.prompt", () => {
mention: { start: 8, end: 17, text: "[Image 1]" },
},
])
expect((yield* admitted(message.id))?.prompt.files).toEqual(message.prompt.files)
const stored = yield* admitted(message.id)
expect(stored?.type).toBe("prompt")
if (stored?.type === "prompt") expect(stored.prompt.files).toEqual(message.prompt.files)
}),
)
@ -336,7 +338,8 @@ describe("SessionV2.prompt", () => {
name: "image.png",
},
])
expect((yield* admitted(message.id))?.prompt.files).toEqual(message.prompt.files)
const stored = yield* admitted(message.id)
expect(stored?.type === "prompt" ? stored.prompt.files : undefined).toEqual(message.prompt.files)
}),
)

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

@ -106,6 +106,7 @@ describe("toLLMMessages", () => {
SessionMessage.Compaction.make({
id: id("compaction"),
type: "compaction",
status: "completed",
reason: "auto",
summary: "Earlier work",
recent: "Recent work",

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

@ -1324,6 +1324,101 @@ describe("SessionRunnerLLM", () => {
}),
)
it.effect("runs one durable compaction barrier before later steer and queued prompts", () =>
Effect.gen(function* () {
yield* setup
requests.length = 0
currentModel = recoveryModel
const session = yield* SessionV2.Service
streamGate = yield* Deferred.make<void>()
streamStarted = yield* Deferred.make<void>()
responses = [
fragmentFixture("text", "text-active", ["Active complete"]).completeEvents,
[LLMEvent.textDelta({ id: "summary", text: "durable summary" })],
fragmentFixture("text", "text-steer", ["Steer complete"]).completeEvents,
fragmentFixture("text", "text-queue", ["Queue complete"]).completeEvents,
]
yield* session.prompt({ sessionID, prompt: PromptInput.Prompt.make({ text: "Active work" }), resume: false })
const active = yield* session.resume(sessionID).pipe(Effect.forkChild)
yield* Deferred.await(streamStarted)
const first = yield* session.compact({ sessionID })
const second = yield* session.compact({ sessionID })
expect(second.id).toBe(first.id)
expect(yield* SessionInput.pendingCompaction((yield* Database.Service).db, sessionID)).toMatchObject({
id: first.id,
})
expect((yield* session.messages({ sessionID })).find((message) => message.id === first.id)).toMatchObject({
type: "compaction",
status: "queued",
})
yield* session.prompt({
sessionID,
prompt: PromptInput.Prompt.make({ text: "Steer after compaction" }),
resume: false,
})
yield* session.prompt({
sessionID,
prompt: PromptInput.Prompt.make({ text: "Queue after compaction" }),
delivery: "queue",
resume: false,
})
expect(yield* SessionInput.hasPending((yield* Database.Service).db, sessionID, "steer")).toBe(false)
yield* Deferred.succeed(streamGate, undefined)
yield* Fiber.join(active)
expect(requests).toHaveLength(4)
expect(userTexts(requests[1])[0]).toContain("Create a new anchored summary")
expect(userTexts(requests[2])).toContain("Steer after compaction")
expect(userTexts(requests[3])).toContain("Queue after compaction")
expect(yield* SessionInput.pendingCompaction((yield* Database.Service).db, sessionID)).toBeUndefined()
expect((yield* session.messages({ sessionID })).find((message) => message.id === first.id)).toMatchObject({
type: "compaction",
status: "completed",
summary: "durable summary",
})
}),
)
it.effect("releases queued prompts when durable compaction fails", () =>
Effect.gen(function* () {
yield* setup
requests.length = 0
currentModel = recoveryModel
const session = yield* SessionV2.Service
streamGate = yield* Deferred.make<void>()
streamStarted = yield* Deferred.make<void>()
responses = [
fragmentFixture("text", "text-active-failure", ["Active complete"]).completeEvents,
[],
fragmentFixture("text", "text-after-failure", ["Continued"]).completeEvents,
]
yield* session.prompt({ sessionID, prompt: PromptInput.Prompt.make({ text: "Active work" }), resume: false })
const active = yield* session.resume(sessionID).pipe(Effect.forkChild)
yield* Deferred.await(streamStarted)
const compaction = yield* session.compact({ sessionID })
yield* session.prompt({
sessionID,
prompt: PromptInput.Prompt.make({ text: "Continue after failure" }),
delivery: "queue",
resume: false,
})
yield* Deferred.succeed(streamGate, undefined)
yield* Fiber.join(active)
expect(requests).toHaveLength(3)
expect(userTexts(requests[2])).toContain("Continue after failure")
expect(yield* SessionInput.pendingCompaction((yield* Database.Service).db, sessionID)).toBeUndefined()
expect((yield* session.messages({ sessionID })).find((message) => message.id === compaction.id)).toMatchObject({
type: "compaction",
status: "failed",
})
}),
)
it.effect("automatically compacts into a completed summary and retained recent turn", () =>
Effect.gen(function* () {
yield* setup