Mirror_Repos/anomalyco_opencode
2
0
Fork 0
Code Issues Pull Requests Actions 66 Packages Projects Releases Wiki Activity

refactor(session): tighten persisted state (#43969)

Browse Source
This commit is contained in:
Kit Langton
2026-08-21 15:27:21 -04:00
committed by GitHub
parent 77c7a7def7
commit 238e1903df
8 changed files with 187 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/core/src/session.ts
+5 -12
View File
@@ -53,6 +53,7 @@ import { Shell as ShellSchema } from "@opencode-ai/schema/shell"
import { KeyedMutex } from "./effect/keyed-mutex.js"
import { fileURLToPath } from "url"
import { SessionEnvironment } from "./session/environment.js"
import { SessionHistory } from "./session/history.js"
// get project -> project.locations
//
@@ -324,19 +325,8 @@ const layer = Layer.effect(
Effect.provide(locations.get(location)),
)
})
const decodeMessage = Schema.decodeUnknownEffect(SessionMessage.Info)
const isDurableSessionEvent = Schema.is(SessionEvent.Durable)
const persistProject = (project: Project.Resolved) => upsertProject(db, project).pipe(Effect.orDie)
const decode = (row: typeof SessionMessageTable.$inferSelect) =>
decodeMessage({ ...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 pendingConflict = Effect.fn("Session.pendingConflict")(function* (input: InboxItemRef) {
yield* result.get(input.sessionID)
@@ -543,7 +533,10 @@ const layer = Layer.effect(
const rows = yield* (input.limit === undefined ? query.all() : query.limit(input.limit).all()).pipe(
Effect.orDie,
)
return yield* Effect.forEach(direction === "previous" ? rows.toReversed() : rows, decode)
return yield* Effect.forEach(
direction === "previous" ? rows.toReversed() : rows,
SessionHistory.decodeMessageRow,
)
}),
message: Effect.fn("Session.message")(function* (input) {
const stored = yield* store.message(input.messageID)
packages/core/src/session/history.ts
+1 -1
View File
@@ -29,7 +29,7 @@ export const latestCompaction = Effect.fnUntraced(function* (db: DatabaseService
.pipe(Effect.orDie)
})
const decodeMessageRow = (row: typeof SessionMessageTable.$inferSelect) =>
export const decodeMessageRow = (row: typeof SessionMessageTable.$inferSelect) =>
decode({ ...row.data, id: row.id, type: row.type }).pipe(
Effect.mapError(
() =>
packages/core/src/session/inbox.ts
+11 -14
View File
@@ -459,13 +459,7 @@ export const promote = Effect.fn("SessionInbox.promote")(function* (
return yield* serialized(
sessionID,
Effect.gen(function* () {
const steers = yield* db
.select()
.from(SessionInboxTable)
.where(and(eq(SessionInboxTable.session_id, sessionID), eq(SessionInboxTable.delivery, "steer")))
.orderBy(asc(SessionInboxTable.enqueued_seq))
.all()
.pipe(Effect.orDie)
const steers = yield* pendingSteers(db, sessionID)
if (steers.length > 0 || scope === "steer") {
const control = steers.findIndex((row) => row.type === "compaction" || row.type === "move")
return yield* publish(db, bus, sessionID, control === -1 ? steers : steers.slice(0, control))
@@ -481,13 +475,7 @@ export const promote = Effect.fn("SessionInbox.promote")(function* (
.pipe(Effect.orDie)
if (!queued) return 0
const promoted = yield* publish(db, bus, sessionID, [queued])
const arrivedSteers = yield* db
.select()
.from(SessionInboxTable)
.where(and(eq(SessionInboxTable.session_id, sessionID), eq(SessionInboxTable.delivery, "steer")))
.orderBy(asc(SessionInboxTable.enqueued_seq))
.all()
.pipe(Effect.orDie)
const arrivedSteers = yield* pendingSteers(db, sessionID)
const control = arrivedSteers.findIndex((row) => row.type === "compaction" || row.type === "move")
return (
promoted +
@@ -496,3 +484,12 @@ export const promote = Effect.fn("SessionInbox.promote")(function* (
}),
)
})
const pendingSteers = (db: DatabaseService, sessionID: SessionSchema.ID) =>
db
.select()
.from(SessionInboxTable)
.where(and(eq(SessionInboxTable.session_id, sessionID), eq(SessionInboxTable.delivery, "steer")))
.orderBy(asc(SessionInboxTable.enqueued_seq))
.all()
.pipe(Effect.orDie)
packages/core/src/session/message-updater.ts
+13 -13
View File
@@ -21,8 +21,18 @@ export interface Adapter {
readonly appendMessage: (message: SessionMessage.Info) => Effect.Effect<void, never, never>
}
type DraftAssistant = WritableDraft<SessionMessage.Assistant>
const projectTerminalSnapshot = (draft: DraftAssistant, event: SessionEvent.Step.Ended | SessionEvent.Step.Failed) => {
if (event.data.snapshot || event.data.files)
draft.snapshot = {
...draft.snapshot,
end: event.data.snapshot,
files: event.data.files ? Array.from(event.data.files) : undefined,
}
}
export function update(adapter: Adapter, event: SessionEvent.DurableEvent) {
type DraftAssistant = WritableDraft<SessionMessage.Assistant>
type DraftTool = WritableDraft<SessionMessage.AssistantTool>
type DraftText = WritableDraft<SessionMessage.AssistantText>
type DraftReasoning = WritableDraft<SessionMessage.AssistantReasoning>
@@ -229,12 +239,7 @@ export function update(adapter: Adapter, event: SessionEvent.DurableEvent) {
draft.providerState = castDraft(event.data.providerState)
draft.cost = event.data.cost
draft.tokens = event.data.tokens
if (event.data.snapshot || event.data.files)
draft.snapshot = {
...draft.snapshot,
end: event.data.snapshot,
files: event.data.files ? Array.from(event.data.files) : undefined,
}
projectTerminalSnapshot(draft, event)
})
},
"session.step.failed": (event) => {
@@ -249,12 +254,7 @@ export function update(adapter: Adapter, event: SessionEvent.DurableEvent) {
draft.cost = event.data.cost
draft.tokens = castDraft(event.data.tokens)
}
if (event.data.snapshot || event.data.files)
draft.snapshot = {
...draft.snapshot,
end: event.data.snapshot,
files: event.data.files ? Array.from(event.data.files) : undefined,
}
projectTerminalSnapshot(draft, event)
})
},
"session.text.started": (event) => {
packages/core/src/session/sql.ts
+2 -1
View File
@@ -16,7 +16,8 @@ import type { CompactionPayload, MovePayload, SyntheticPayload, UserPayload } fr
import type { RevertV1 } from "@opencode-ai/schema/session-revert"
import type { Schema } from "effect"
type SessionMessageData = Omit<(typeof SessionMessage.Info)["Encoded"], "type" | "id">
type DistributiveOmit<T, K extends PropertyKey> = T extends unknown ? Omit<T, K> : never
type SessionMessageData = DistributiveOmit<(typeof SessionMessage.Info)["Encoded"], "type" | "id">
export const SessionTable = sqliteTable(
"session_v2",
packages/core/src/session/store.ts
+2 -3
View File
@@ -1,7 +1,7 @@
export * as SessionStore from "./store.js"
import { and, eq, isNotNull, isNull, sql } from "drizzle-orm"
import { Context, Effect, Layer, Schema } from "effect"
import { Context, Effect, Layer } from "effect"
import { Database } from "../database/database.js"
import { makeGlobalNode } from "@opencode-ai/util/effect/app-node"
import { SessionHistory } from "./history.js"
@@ -51,7 +51,6 @@ const layer = Layer.effect(
Service,
Effect.gen(function* () {
const { db } = yield* Database.Service
const decodeMessage = Schema.decodeUnknownEffect(SessionMessage.Info)
return Service.of({
get: Effect.fnUntraced(function* (sessionID) {
@@ -69,7 +68,7 @@ const layer = Layer.effect(
return row
? {
sessionID: Session.ID.make(row.session_id),
message: yield* decodeMessage({ ...row.data, id: row.id, type: row.type }).pipe(Effect.orDie),
message: yield* SessionHistory.decodeMessageRow(row).pipe(Effect.orDie),
}
: undefined
}),
packages/core/test/session-projector.test.ts
+128 -1
View File
@@ -12,7 +12,7 @@ import { Model } from "@opencode-ai/core/model"
import { Project } from "@opencode-ai/core/project"
import { ProjectTable } from "@opencode-ai/core/project/sql"
import { Provider } from "@opencode-ai/core/provider"
import { AbsolutePath } from "@opencode-ai/core/schema"
import { AbsolutePath, RelativePath } from "@opencode-ai/core/schema"
import { Session } from "@opencode-ai/core/session"
import { SessionEvent } from "@opencode-ai/core/session/event"
import { SessionMessage } from "@opencode-ai/core/session/message"
@@ -298,6 +298,50 @@ describe("SessionProjector", () => {
}).pipe(Effect.provide(sessionsLayer)),
)
it.effect("maps malformed persisted rows consistently while single-message lookup defects", () =>
Effect.gen(function* () {
const { db } = yield* Database.Service
yield* db
.insert(ProjectTable)
.values({ id: Project.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] })
.run()
.pipe(Effect.orDie)
yield* db
.insert(SessionTable)
.values({
id: sessionID,
project_id: Project.ID.global,
slug: "test",
directory: "/project",
title: "test",
version: "test",
})
.run()
.pipe(Effect.orDie)
const messageID = SessionMessage.ID.make("msg_malformed")
yield* db
.insert(SessionMessageTable)
.values({
id: messageID,
session_id: sessionID,
type: "user",
seq: 0,
data: { text: "valid before corruption", time: { created: 0 } },
})
.run()
.pipe(Effect.orDie)
yield* db.run(sql`update session_message set data = '{"time":{"created":0}}' where id = ${messageID}`)
const sessions = yield* Session.Service
const expected = { _tag: "Session.MessageDecodeError", sessionID, messageID }
expect(yield* sessions.messages({ sessionID }).pipe(Effect.flip)).toMatchObject(expected)
expect(yield* sessions.context(sessionID).pipe(Effect.flip)).toMatchObject(expected)
expect(yield* sessions.message({ sessionID, messageID }).pipe(Effect.catchDefect(Effect.succeed))).toMatchObject(
expected,
)
}).pipe(Effect.provide(sessionsLayer)),
)
it.effect("consumes the pending row and projects the message at promotion", () =>
Effect.gen(function* () {
const { db } = yield* Database.Service
@@ -711,6 +755,89 @@ describe("SessionProjector", () => {
}),
)
it.effect("projects ended and failed step terminal state", () =>
Effect.gen(function* () {
const { db } = yield* Database.Service
yield* db
.insert(ProjectTable)
.values({ id: Project.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] })
.run()
.pipe(Effect.orDie)
yield* db
.insert(SessionTable)
.values({
id: sessionID,
project_id: Project.ID.global,
slug: "test",
directory: "/project",
title: "test",
version: "test",
})
.run()
.pipe(Effect.orDie)
const endedID = SessionMessage.ID.make("msg_ended")
const failedID = SessionMessage.ID.make("msg_failed")
yield* db
.insert(SessionMessageTable)
.values([assistantRow(endedID, 0), assistantRow(failedID, 1)])
.run()
.pipe(Effect.orDie)
const service = yield* Bus.Service
yield* service.publish(SessionEvent.Step.Ended, {
sessionID,
assistantMessageID: endedID,
finish: "stop",
rawFinish: "stop_sequence",
providerState: { response: "ended" },
cost: Money.USD.make(1),
tokens: { input: 2, output: 3, reasoning: 4, cache: { read: 5, write: 6 } },
snapshot: Snapshot.ID.make("snap_ended"),
files: [RelativePath.make("src/ended.ts")],
})
yield* service.publish(SessionEvent.Step.Failed, {
sessionID,
assistantMessageID: failedID,
finish: "content-filter",
rawFinish: "blocked",
providerState: { response: "failed" },
error: { type: "provider.invalid-request", message: "Failed" },
snapshot: Snapshot.ID.make("snap_failed"),
files: [RelativePath.make("src/failed.ts")],
})
const rows = yield* db
.select()
.from(SessionMessageTable)
.where(eq(SessionMessageTable.session_id, sessionID))
.orderBy(asc(SessionMessageTable.seq))
.all()
.pipe(Effect.orDie)
const messages = rows.map((row) =>
Schema.decodeUnknownSync(SessionMessage.Info)({ ...row.data, id: row.id, type: row.type }),
)
expect(messages[0]).toMatchObject({
type: "assistant",
finish: "stop",
rawFinish: "stop_sequence",
providerState: { response: "ended" },
cost: Money.USD.make(1),
tokens: { input: 2, output: 3, reasoning: 4, cache: { read: 5, write: 6 } },
snapshot: { end: "snap_ended", files: ["src/ended.ts"] },
time: { completed: created },
})
expect(messages[1]).toMatchObject({
type: "assistant",
finish: "content-filter",
rawFinish: "blocked",
providerState: { response: "failed" },
error: { type: "provider.invalid-request", message: "Failed" },
snapshot: { end: "snap_failed", files: ["src/failed.ts"] },
time: { completed: created },
})
}),
)
it.effect("does not revive a stale incomplete assistant projection", () =>
Effect.gen(function* () {
const { db } = yield* Database.Service
packages/core/test/session-sql.types.ts
+25
View File
@@ -0,0 +1,25 @@
import { SessionMessageTable } from "@opencode-ai/core/session/sql"
type MessageData = (typeof SessionMessageTable.$inferInsert)["data"]
const user = {
text: "Hello",
time: { created: 0 },
} satisfies MessageData
const assistant = {
agent: "build",
model: { id: "model", providerID: "provider" },
content: [],
time: { created: 0 },
} satisfies MessageData
const invalid = {
// @ts-expect-error Persisted message variants retain their field types.
text: 42,
time: { created: 0 },
} satisfies MessageData
void user
void assistant
void invalid