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

refactor(schema): apply session review decisions (#35793)

Browse source
This commit is contained in:
Kit Langton 2026-07-07 22:10:11 -04:00 • committed by GitHub
commit ed6ad272ec
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
142 changed files with 4239 additions and 3174 deletions
Download patch file Download diff file Expand all files Collapse all files

39
packages/core/src/session/compaction.ts
View file

@ -64,19 +64,21 @@ type Dependencies = {
export type AutoInput = {
readonly sessionID: SessionSchema.ID
readonly messages: readonly SessionMessage.Message[]
readonly messages: readonly SessionMessage.Info[]
readonly request: LLMRequest
}
type CompactInput = {
readonly sessionID: SessionSchema.ID
readonly messages: readonly SessionMessage.Message[]
readonly messages: readonly SessionMessage.Info[]
readonly model: Model
readonly inputID?: SessionMessage.ID
}
export type ManualInput = {
readonly session: SessionSchema.Info
readonly messages: readonly SessionMessage.Message[]
readonly messages: readonly SessionMessage.Info[]
readonly inputID: SessionMessage.ID
}
export interface Interface {
@ -99,7 +101,7 @@ export const serializeToolContent = (content: SessionMessage.ToolStateCompleted[
)
.join("\n")
const serialize = (message: SessionMessage.Message) => {
const serialize = (message: SessionMessage.Info) => {
if (message.type === "user") {
const files =
message.files?.map(
@ -128,7 +130,7 @@ const serialize = (message: SessionMessage.Message) => {
if (message.type === "system") return `[System update]: ${message.text}`
if (message.type === "synthetic") return `[Synthetic context]: ${message.text}`
if (message.type === "skill") return `[Skill activated: ${message.name}]\n${message.text}`
if (message.type === "shell") return `[Shell]: ${message.shell.command}\n${truncate(message.output?.output ?? "")}`
if (message.type === "shell") return `[Shell]: ${message.command}\n${truncate(message.output?.output ?? "")}`
return ""
}
@ -147,7 +149,7 @@ const settings = (documents: readonly Config.Entry[]) => {
}
const select = (
messages: readonly SessionMessage.Message[],
messages: readonly SessionMessage.Info[],
tokens: number,
): { readonly head: string; readonly recent: string } | undefined => {
const conversation = messages
@ -198,6 +200,7 @@ const make = (dependencies: Dependencies) => {
readonly context: readonly string[]
readonly recent: string
readonly output?: number
readonly inputID?: SessionMessage.ID
}) {
const context = input.model.route.defaults.limits?.context
if (context === undefined || context <= 0) return false
@ -208,6 +211,8 @@ const make = (dependencies: Dependencies) => {
yield* dependencies.events.publish(SessionEvent.Compaction.Started, {
sessionID: input.sessionID,
reason: input.reason,
recent: input.recent,
inputID: input.inputID,
})
const chunks: string[] = []
@ -235,9 +240,27 @@ const make = (dependencies: Dependencies) => {
}),
Effect.as(true),
Effect.catchTag("LLM.Error", () => Effect.succeed(false)),
Effect.onInterrupt(() =>
input.reason === "auto"
? dependencies.events.publish(SessionEvent.Compaction.Failed, {
sessionID: input.sessionID,
reason: input.reason,
error: { type: "compaction.interrupted", message: "Compaction was interrupted" },
inputID: input.inputID,
})
: Effect.void,
),
)
const summary = chunks.join("")
if (!summarized || failed || !summary.trim()) return false
if (!summarized || failed || !summary.trim()) {
yield* dependencies.events.publish(SessionEvent.Compaction.Failed, {
sessionID: input.sessionID,
reason: input.reason,
error: { type: "compaction.failed", message: "Compaction produced no summary" },
inputID: input.inputID,
})
return false
}
yield* dependencies.events.publish(SessionEvent.Compaction.Ended, {
sessionID: input.sessionID,
reason: input.reason,
@ -284,6 +307,7 @@ const make = (dependencies: Dependencies) => {
),
recent: forcedShortContext ? "" : selected.recent,
output: input.output,
inputID: input.inputID,
})
})
const compactManual = Effect.fn("SessionCompaction.compactManual")(function* (input: CompactInput) {
@ -327,6 +351,7 @@ export const layer = Layer.effect(
sessionID: input.session.id,
messages: input.messages,
model: resolved.model,
inputID: input.inputID,
})
}),
})

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

@ -8,7 +8,7 @@ import { InstructionCheckpointTable, SessionMessageTable } from "./sql"
type DatabaseService = Database.Interface["db"]
const decode = Schema.decodeUnknownEffect(SessionMessage.Message)
const decode = Schema.decodeUnknownEffect(SessionMessage.Info)
export const latestCompaction = Effect.fnUntraced(function* (db: DatabaseService, sessionID: SessionSchema.ID) {
return yield* db

11
packages/core/src/session/info.ts
View file

@ -1,4 +1,4 @@
import { DateTime } from "effect"
import { DateTime, Schema } from "effect"
import { AgentV2 } from "../agent"
import { Location } from "../location"
import { ModelV2 } from "../model"
@ -9,7 +9,10 @@ import { WorkspaceV2 } from "../workspace"
import { SessionSchema } from "./schema"
import { SessionTable } from "./sql"
import { SessionMessage } from "./message"
import { Snapshot } from "../snapshot"
import { PersistedRevert } from "@opencode-ai/schema/session-revert"
import { Money } from "@opencode-ai/schema/money"
const decodeRevert = Schema.decodeUnknownSync(PersistedRevert)
export function fromRow(row: typeof SessionTable.$inferSelect): SessionSchema.Info {
return SessionSchema.Info.make({
@ -31,7 +34,7 @@ export function fromRow(row: typeof SessionTable.$inferSelect): SessionSchema.In
variant: ModelV2.VariantID.make(row.model.variant ?? "default"),
}
: undefined,
cost: row.cost,
cost: Money.USD.make(row.cost),
tokens: {
input: row.tokens_input,
output: row.tokens_output,
@ -46,7 +49,7 @@ export function fromRow(row: typeof SessionTable.$inferSelect): SessionSchema.In
workspaceID: row.workspace_id ? WorkspaceV2.ID.make(row.workspace_id) : undefined,
}),
subpath: row.path ? RelativePath.make(row.path) : undefined,
revert: row.revert ? { ...row.revert, messageID: SessionMessage.ID.make(row.revert.messageID) } : undefined,
revert: row.revert ? decodeRevert(row.revert) : undefined,
time: {
created: DateTime.makeUnsafe(row.time_created),
updated: DateTime.makeUnsafe(row.time_updated),

6
packages/core/src/session/input.ts
View file

@ -2,7 +2,7 @@ export * as SessionInput from "./input"
import { and, asc, eq, isNull } from "drizzle-orm"
import { DateTime, Effect, Schema } from "effect"
import { Admitted, Compaction, Delivery, Entry, PromptEntry } from "@opencode-ai/schema/session-input"
import { Admitted, Compaction, Delivery, Info, PromptEntry } from "@opencode-ai/schema/session-input"
import type { Database } from "../database/database"
import type { EventV2 } from "../event"
import { KeyedMutex } from "../effect/keyed-mutex"
@ -14,7 +14,7 @@ import { SessionInputTable, SessionMessageTable } from "./sql"
type DatabaseService = Database.Interface["db"]
export { Admitted, Compaction, Delivery, Entry, PromptEntry }
export { Admitted, Compaction, Delivery, Info, PromptEntry }
const decodePrompt = Schema.decodeUnknownSync(Prompt)
const encodePrompt = Schema.encodeSync(Prompt)
@ -24,7 +24,7 @@ export class LifecycleConflict extends Schema.TaggedErrorClass<LifecycleConflict
id: SessionMessage.ID,
}) {}
const fromRow = (row: typeof SessionInputTable.$inferSelect): Entry => {
const fromRow = (row: typeof SessionInputTable.$inferSelect): Info => {
const base = {
admittedSeq: row.admitted_seq,
id: SessionMessage.ID.make(row.id),

71
packages/core/src/session/message-updater.ts
View file

@ -4,7 +4,7 @@ import { SessionEvent } from "./event"
import { SessionMessage } from "./message"
export type MemoryState = {
messages: SessionMessage.Message[]
messages: SessionMessage.Info[]
}
export interface Adapter {
@ -14,13 +14,13 @@ export interface Adapter {
messageID: SessionMessage.ID,
) => Effect.Effect<SessionMessage.Assistant | undefined, never, never>
readonly getShell: (
shellID: SessionMessage.Shell["shell"]["id"],
shellID: SessionMessage.Shell["shellID"],
) => Effect.Effect<SessionMessage.Shell | undefined, never, never>
readonly getCompaction: () => Effect.Effect<SessionMessage.Compaction | undefined, never, never>
readonly updateAssistant: (assistant: SessionMessage.Assistant) => Effect.Effect<void, never, never>
readonly updateShell: (shell: SessionMessage.Shell) => Effect.Effect<void, never, never>
readonly updateCompaction: (compaction: SessionMessage.Compaction) => Effect.Effect<void, never, never>
readonly appendMessage: (message: SessionMessage.Message) => Effect.Effect<void, never, never>
readonly appendMessage: (message: SessionMessage.Info) => Effect.Effect<void, never, never>
}
export function memory(state: MemoryState): Adapter {
@ -29,9 +29,7 @@ export function memory(state: MemoryState): Adapter {
const shellIndex = (messageID: SessionMessage.ID) =>
state.messages.findLastIndex((message) => message.id === messageID)
const compactionIndex = () =>
state.messages.findLastIndex(
(message) => message.type === "compaction" && (message.status === "queued" || message.status === "running"),
)
state.messages.findLastIndex((message) => message.type === "compaction" && message.status === "running")
// A newer step supersedes stale incomplete rows; never resume an older assistant projection.
const latestAssistantIndex = () => state.messages.findLastIndex((message) => message.type === "assistant")
@ -64,7 +62,7 @@ export function memory(state: MemoryState): Adapter {
getShell(shellID) {
return Effect.sync(() => {
return state.messages.find((message): message is SessionMessage.Shell => {
return message.type === "shell" && message.shell.id === shellID
return message.type === "shell" && message.shellID === shellID
})
})
},
@ -186,13 +184,13 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
id: SessionMessage.ID.fromEvent(event.id),
type: "system",
text: event.data.text,
metadata: event.metadata,
time: { created: event.created },
}),
),
"session.synthetic": (event) => {
return adapter.appendMessage(
SessionMessage.Synthetic.make({
sessionID: event.data.sessionID,
text: event.data.text,
description: event.data.description,
metadata: event.data.metadata,
@ -207,8 +205,10 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
SessionMessage.Skill.make({
id: SessionMessage.ID.fromEvent(event.id),
type: "skill",
skill: event.data.id,
name: event.data.name,
text: event.data.text,
metadata: event.metadata,
time: { created: event.created },
}),
)
@ -219,7 +219,9 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
id: SessionMessage.ID.fromEvent(event.id),
type: "shell",
metadata: event.metadata,
shell: event.data.shell,
shellID: event.data.shell.id,
command: event.data.shell.command,
status: event.data.shell.status,
time: { created: event.created },
}),
)
@ -230,7 +232,8 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
if (currentShell) {
yield* adapter.updateShell(
produce(currentShell, (draft) => {
draft.shell = castDraft(event.data.shell)
draft.status = event.data.shell.status
draft.exit = event.data.shell.exit
draft.output = event.data.output
draft.time.completed = event.created
}),
@ -270,6 +273,7 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
type: "assistant",
agent: event.data.agent,
model: event.data.model,
metadata: event.metadata,
time: { created: event.created },
content: [],
snapshot: event.data.snapshot ? { start: event.data.snapshot } : undefined,
@ -329,7 +333,7 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
id: event.data.callID,
name: event.data.name,
time: { created: event.created },
state: SessionMessage.ToolStatePending.make({ status: "pending", input: "" }),
state: SessionMessage.ToolStateStreaming.make({ status: "streaming", input: "" }),
}),
),
)
@ -339,7 +343,7 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
"session.tool.input.ended": (event) => {
return updateOwnedAssistant(event.data.assistantMessageID, (draft) => {
const match = latestTool(draft, event.data.callID)
if (match && match.state.status === "pending") match.state.input = event.data.text
if (match && match.state.status === "streaming") match.state.input = event.data.text
})
},
"session.tool.called": (event) => {
@ -382,7 +386,6 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
input: match.state.input,
structured: event.data.structured,
content: [...event.data.content],
outputPaths: event.data.outputPaths ? [...event.data.outputPaths] : [],
result: event.data.result,
}),
)
@ -392,7 +395,7 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
"session.tool.failed": (event) => {
return updateOwnedAssistant(event.data.assistantMessageID, (draft) => {
const match = latestTool(draft, event.data.callID)
if (match && (match.state.status === "pending" || match.state.status === "running")) {
if (match && (match.state.status === "streaming" || match.state.status === "running")) {
match.executed = event.data.executed || match.executed === true
match.providerResultState = event.data.resultState
match.time.completed = event.created
@ -448,31 +451,30 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
}
})
},
"session.compaction.admitted": (event) =>
"session.compaction.admitted": () => Effect.void,
"session.compaction.started": (event) =>
adapter.appendMessage(
SessionMessage.Compaction.make({
id: event.data.inputID,
SessionMessage.CompactionRunning.make({
id: event.data.inputID ?? SessionMessage.ID.fromEvent(event.id),
type: "compaction",
status: "queued",
status: "running",
metadata: event.metadata,
reason: "manual",
reason: event.data.reason,
summary: "",
recent: "",
recent: event.data.recent ?? "",
time: { created: event.created },
}),
),
"session.compaction.started": (event) =>
"session.compaction.delta": (event) =>
Effect.gen(function* () {
if (event.data.reason !== "manual") return
const current = yield* adapter.getCompaction()
if (!current) return
yield* adapter.updateCompaction({ ...current, status: "running" })
if (current?.status !== "running") return
yield* adapter.updateCompaction({ ...current, summary: current.summary + event.data.text })
}),
"session.compaction.delta": () => Effect.void,
"session.compaction.ended": (event) => {
return Effect.gen(function* () {
const current = event.data.reason === "manual" ? yield* adapter.getCompaction() : undefined
if (current) {
const current = yield* adapter.getCompaction()
if (current?.status === "running") {
yield* adapter.updateCompaction({
...current,
status: "completed",
@ -496,11 +498,20 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
)
})
},
"session.compaction.failed": () =>
"session.compaction.failed": (event) =>
Effect.gen(function* () {
const current = yield* adapter.getCompaction()
if (!current) return
yield* adapter.updateCompaction({ ...current, status: "failed" })
const failed = SessionMessage.CompactionFailed.make({
id: current?.id ?? event.data.inputID ?? SessionMessage.ID.fromEvent(event.id),
type: "compaction",
status: "failed",
metadata: current?.metadata ?? event.metadata,
reason: event.data.reason,
error: event.data.error,
time: current?.time ?? { created: event.created },
})
if (current?.status === "running") return yield* adapter.updateCompaction(failed)
yield* adapter.appendMessage(failed)
}),
"session.revert.staged": () => Effect.void,
"session.revert.cleared": () => Effect.void,

80
packages/core/src/session/projector.ts
View file

@ -24,15 +24,14 @@ import {
} from "./sql"
import type { DeepMutable } from "../schema"
import { Slug } from "../util/slug"
import { Money } from "@opencode-ai/schema/money"
type DatabaseService = Database.Interface["db"]
type MessageEvent = Exclude<
SessionEvent.DurableEvent,
typeof SessionEvent.Forked.Type | typeof SessionEvent.Deleted.Type
>
type CurrentDurableEvent = Extract<SessionEvent.Event, { readonly durable: object }>
type MessageEvent = Exclude<CurrentDurableEvent, typeof SessionEvent.Forked.Type | typeof SessionEvent.Deleted.Type>
const decodeMessage = Schema.decodeUnknownSync(SessionMessage.Message)
const encodeMessage = Schema.encodeSync(SessionMessage.Message)
const decodeMessage = Schema.decodeUnknownSync(SessionMessage.Info)
const encodeMessage = Schema.encodeSync(SessionMessage.Info)
export class SessionAlreadyProjected extends Error {}
@ -87,7 +86,14 @@ function sessionRow(info: SessionV1.SessionInfo): typeof SessionTable.$inferInse
tokens_reasoning: (info.tokens ?? { reasoning: 0 }).reasoning,
tokens_cache_read: (info.tokens ?? { cache: { read: 0 } }).cache.read,
tokens_cache_write: (info.tokens ?? { cache: { write: 0 } }).cache.write,
revert: info.revert ? { ...info.revert, messageID: SessionMessage.ID.make(info.revert.messageID) } : null,
revert: info.revert
? {
messageID: SessionMessage.ID.make(info.revert.messageID),
partID: info.revert.partID,
snapshot: info.revert.snapshot,
diff: info.revert.diff,
}
: null,
permission: info.permission ? [...info.permission] : undefined,
time_created: info.time.created,
time_updated: info.time.updated,
@ -151,7 +157,7 @@ const publishSessionUsage = Effect.fn("SessionProjector.publishUsage")(function*
if (!row) return
yield* events.publish(SessionEvent.UsageUpdated, {
sessionID,
cost: row.cost,
cost: Money.USD.make(row.cost),
tokens: {
input: row.input,
output: row.output,
@ -257,7 +263,7 @@ const projectFork = Effect.fn("SessionProjector.projectFork")(function* (
eq(SessionMessageTable.session_id, event.data.parentID),
gt(SessionMessageTable.seq, cursor),
lt(SessionMessageTable.seq, copiedSeq + 1),
sql`${SessionMessageTable.type} != 'compaction' or json_extract(${SessionMessageTable.data}, '$.status') not in ('queued', 'running')`,
sql`${SessionMessageTable.type} != 'compaction' or json_extract(${SessionMessageTable.data}, '$.status') != 'running'`,
),
)
.orderBy(asc(SessionMessageTable.seq))
@ -280,7 +286,7 @@ const projectFork = Effect.fn("SessionProjector.projectFork")(function* (
seq: row.seq,
time_created: row.time_created,
time_updated: row.time_updated,
data: row.type === "synthetic" ? { ...row.data, sessionID: event.data.sessionID } : row.data,
data: row.data,
}
}),
)
@ -336,7 +342,7 @@ function run(db: DatabaseService, event: MessageEvent) {
return Effect.gen(function* () {
const decodeRow = (row: typeof SessionMessageTable.$inferSelect) =>
decodeMessage({ ...row.data, id: row.id, type: row.type })
const updateMessage = (message: SessionMessage.Message) => {
const updateMessage = (message: SessionMessage.Info) => {
if (event.durable === undefined)
return Effect.die(new Error("Durable Session event is missing aggregate sequence"))
const encoded = encodeMessage(message)
@ -353,7 +359,7 @@ function run(db: DatabaseService, event: MessageEvent) {
.run()
.pipe(Effect.orDie)
}
const appendMessage = (message: SessionMessage.Message) => insertMessage(db, event, message)
const appendMessage = (message: SessionMessage.Info) => insertMessage(db, event, message)
const adapter: SessionMessageUpdater.Adapter = {
getModel() {
return db
@ -412,7 +418,7 @@ function run(db: DatabaseService, event: MessageEvent) {
and(
eq(SessionMessageTable.session_id, event.data.sessionID),
eq(SessionMessageTable.type, "shell"),
sql`json_extract(${SessionMessageTable.data}, '$.shell.id') = ${shellID}`,
sql`json_extract(${SessionMessageTable.data}, '$.shellID') = ${shellID}`,
),
)
.orderBy(desc(SessionMessageTable.seq))
@ -433,7 +439,7 @@ function run(db: DatabaseService, event: MessageEvent) {
and(
eq(SessionMessageTable.session_id, event.data.sessionID),
eq(SessionMessageTable.type, "compaction"),
sql`json_extract(${SessionMessageTable.data}, '$.status') in ('queued', 'running')`,
sql`json_extract(${SessionMessageTable.data}, '$.status') = 'running'`,
),
)
.orderBy(desc(SessionMessageTable.seq))
@ -454,7 +460,7 @@ function run(db: DatabaseService, event: MessageEvent) {
})
}
function insertMessage(db: DatabaseService, event: SessionEvent.DurableEvent, message: SessionMessage.Message) {
function insertMessage(db: DatabaseService, event: SessionEvent.DurableEvent, message: SessionMessage.Info) {
if (event.durable === undefined) return Effect.die(new Error("Durable Session event is missing aggregate sequence"))
const encoded = encodeMessage(message)
const { id, type, ...data } = encoded
@ -661,14 +667,12 @@ const layer = Layer.effectDiscard(
Effect.gen(function* () {
if (event.durable === undefined)
return yield* Effect.die(new Error("Durable Session event is missing aggregate sequence"))
const admitted = yield* SessionInput.projectCompactionAdmitted(db, {
yield* SessionInput.projectCompactionAdmitted(db, {
admittedSeq: event.durable.seq,
id: event.data.inputID,
sessionID: event.data.sessionID,
timeCreated: event.created,
})
if (admitted.id !== event.data.inputID) return
yield* run(db, event)
}),
)
yield* events.project(SessionEvent.Execution.Succeeded, (event) => run(db, event))
@ -676,15 +680,7 @@ const layer = Layer.effectDiscard(
yield* events.project(SessionEvent.Execution.Interrupted, (event) => run(db, event))
yield* events.project(SessionEvent.InstructionsUpdated, (event) => run(db, event))
yield* events.project(SessionEvent.Synthetic, (event) => run(db, event))
yield* events.project(SessionEvent.Skill.Activated, (event) =>
insertMessage(db, event, {
id: SessionMessage.ID.fromEvent(event.id),
type: "skill",
name: event.data.name,
text: event.data.text,
time: { created: event.created },
}),
)
yield* events.project(SessionEvent.Skill.Activated, (event) => run(db, event))
yield* events.project(SessionEvent.Shell.Started, (event) => run(db, event))
yield* events.project(SessionEvent.Shell.Ended, (event) => run(db, event))
yield* events.project(SessionEvent.Step.Started, (event) => run(db, event))
@ -730,22 +726,26 @@ const layer = Layer.effectDiscard(
yield* run(db, event)
if (event.durable === undefined)
return yield* Effect.die(new Error("Durable Session event is missing aggregate sequence"))
yield* SessionInput.settleCompaction(db, {
sessionID: event.data.sessionID,
handledSeq: event.durable.seq,
})
if (event.data.reason === "manual")
yield* SessionInput.settleCompaction(db, {
sessionID: event.data.sessionID,
handledSeq: event.durable.seq,
})
}),
)
yield* events.project(SessionEvent.RevertEvent.Staged, (event) =>
db
.update(SessionTable)
.set({
revert: { ...event.data.revert, files: event.data.revert.files ? [...event.data.revert.files] : undefined },
time_updated: DateTime.toEpochMillis(event.created),
})
.where(eq(SessionTable.id, event.data.sessionID))
.run()
.pipe(Effect.orDie, Effect.asVoid),
Effect.gen(function* () {
const revert = event.data.revert
yield* db
.update(SessionTable)
.set({
revert: { ...revert, files: revert.files ? [...revert.files] : undefined },
time_updated: DateTime.toEpochMillis(event.created),
})
.where(eq(SessionTable.id, event.data.sessionID))
.run()
.pipe(Effect.orDie)
}),
)
yield* events.project(SessionEvent.RevertEvent.Cleared, (event) =>
db

12
packages/core/src/session/revert.ts
View file

@ -1,7 +1,7 @@
export * as SessionRevert from "./revert"
import { and, asc, eq, gt } from "drizzle-orm"
import { DateTime, Effect, Schema } from "effect"
import { Effect, Schema } from "effect"
import { Database } from "../database/database"
import { EventV2 } from "../event"
import { RelativePath } from "../schema"
@ -46,7 +46,7 @@ const plan = Effect.fn("SessionRevert.plan")(function* (input: BoundaryInput) {
.orderBy(asc(SessionMessageTable.seq))
.all()
.pipe(Effect.orDie)
const decode = Schema.decodeUnknownEffect(SessionMessage.Message)
const decode = Schema.decodeUnknownEffect(SessionMessage.Info)
const files = new Map<RelativePath, Snapshot.ID>()
for (const row of rows) {
const message = yield* decode({ ...row.data, id: row.id, type: row.type }).pipe(Effect.orDie)
@ -70,7 +70,7 @@ export const stage = Effect.fn("SessionRevert.stage")(function* (input: {
const next = yield* plan({ sessionID: input.session.id, messageID: input.messageID })
const restore = new Map<RelativePath, Snapshot.ID>()
if (original) {
for (const file of input.session.revert?.files ?? []) restore.set(file.path, original)
for (const file of input.session.revert?.files ?? []) restore.set(RelativePath.make(file.file), original)
}
if (input.files !== false) for (const [file, tree] of next) restore.set(file, tree)
if (restore.size) yield* snapshot.restore({ files: restore })
@ -81,10 +81,6 @@ export const stage = Effect.fn("SessionRevert.stage")(function* (input: {
const revert = {
messageID: input.messageID,
snapshot: original,
diff: files
.map((file) => file.patch)
.join("")
.trim(),
files,
} satisfies SessionSchema.Info["revert"]
yield* events.publish(SessionEvent.RevertEvent.Staged, {
@ -100,7 +96,7 @@ export const clear = Effect.fn("SessionRevert.clear")(function* (session: Sessio
const original = session.revert.snapshot ? Snapshot.ID.make(session.revert.snapshot) : undefined
if (original)
yield* snapshot.restore({
files: new Map((session.revert.files ?? []).map((file) => [file.path, original])),
files: new Map((session.revert.files ?? []).map((file) => [RelativePath.make(file.file), original])),
})
const events = yield* EventV2.Service
yield* events.publish(SessionEvent.RevertEvent.Cleared, {

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

@ -11,6 +11,7 @@ import {
type ProviderErrorEvent,
} from "@opencode-ai/llm"
import { SessionError } from "@opencode-ai/schema/session-error"
import { Money } from "@opencode-ai/schema/money"
import { Cause, Effect, Exit, Fiber, FiberSet, Layer, Option, Semaphore, Stream } from "effect"
import { AgentV2 } from "../../agent"
import { Config } from "../../config"
@ -67,13 +68,13 @@ export function calculateCost(costs: ModelV2.Info["cost"], tokens: StepTokens) {
.filter((cost) => cost.tier?.type === "context" && context > cost.tier.size)
.toSorted((a, b) => (b.tier?.size ?? 0) - (a.tier?.size ?? 0))[0]
const cost = tier ?? costs.find((cost) => cost.tier === undefined)
if (!cost) return 0
return (
if (!cost) return Money.USD.zero
return Money.USD.make(
(tokens.input * cost.input +
(tokens.output + tokens.reasoning) * cost.output +
tokens.cache.read * cost.cache.read +
tokens.cache.write * cost.cache.write) /
1_000_000
1_000_000,
)
}
@ -165,7 +166,7 @@ const layer = Layer.effect(
for (const message of yield* store.context(sessionID)) {
if (message.type !== "assistant") continue
for (const tool of message.content) {
if (tool.type !== "tool" || (tool.state.status !== "pending" && tool.state.status !== "running")) continue
if (tool.type !== "tool" || (tool.state.status !== "streaming" && tool.state.status !== "running")) continue
yield* events.publish(SessionEvent.Tool.Failed, {
sessionID,
assistantMessageID: message.id,
@ -300,8 +301,7 @@ const layer = Layer.effect(
// Durable publishes are serialized so tool fibers and step settlement never interleave
// mid-event.
const serialized = <A, E, R>(effect: Effect.Effect<A, E, R>) => publication.withPermit(effect)
const publish = (event: LLMEvent, outputPaths: ReadonlyArray<string> = [], error?: SessionError.Error) =>
serialized(publisher.publish(event, outputPaths, error))
const publish = (event: LLMEvent, error?: SessionError.Error) => serialized(publisher.publish(event, error))
let overflowFailure: ProviderErrorEvent | undefined
const providerStream = llm.stream(hookedRequest).pipe(
Stream.runForEach((event) =>
@ -359,7 +359,6 @@ const layer = Layer.effect(
result: settlement.result,
output: settlement.output,
}),
settlement.outputPaths ?? [],
settlement.error,
).pipe(
Effect.andThen(
@ -599,12 +598,30 @@ const layer = Layer.effect(
return yield* compaction.compactManual({
session,
messages: yield* store.context(sessionID),
inputID: pending.id,
})
}),
).pipe(Effect.exit)
if (Exit.isSuccess(compacted) && compacted.value) return true
yield* events.publish(SessionEvent.Compaction.Failed, { sessionID })
if (Exit.isFailure(compacted)) return yield* Effect.failCause(compacted.cause)
if (Exit.isFailure(compacted)) {
const unsettled = yield* SessionInput.pendingCompaction(db, sessionID)
if (unsettled)
yield* events.publish(SessionEvent.Compaction.Failed, {
sessionID,
reason: "manual",
error: { type: "compaction.failed", message: Cause.pretty(compacted.cause) },
inputID: unsettled.id,
})
return yield* Effect.failCause(compacted.cause)
}
const unsettled = yield* SessionInput.pendingCompaction(db, sessionID)
if (unsettled)
yield* events.publish(SessionEvent.Compaction.Failed, {
sessionID,
reason: "manual",
error: { type: "compaction.failed", message: "Compaction could not start" },
inputID: unsettled.id,
})
return true
}),
)

16
packages/core/src/session/runner/publish-llm-event.ts
View file

@ -6,13 +6,16 @@ import { SessionEvent } from "../event"
import { SessionMessage } from "../message"
import { SessionSchema } from "../schema"
import { SessionError } from "@opencode-ai/schema/session-error"
import { Money } from "@opencode-ai/schema/money"
import { AgentV2 } from "../../agent"
import { Snapshot } from "../../snapshot"
type Input = {
readonly sessionID: SessionSchema.ID
readonly agent: string
readonly agent: AgentV2.ID
readonly model: ModelV2.Ref
readonly provider: string
readonly snapshot?: string
readonly snapshot?: Snapshot.ID
readonly assistantMessageID?: SessionMessage.ID
}
@ -226,7 +229,7 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
})
const publishStepFailure = Effect.fnUntraced(function* (usage?: {
readonly cost: number
readonly cost: Money.USD
readonly tokens: ReturnType<typeof tokens>
}) {
if (stepFailed || stepFailure === undefined) return
@ -265,11 +268,7 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
return tool ? Effect.succeed(tool.assistantMessageID) : Effect.die(new Error(`Unknown tool call: ${callID}`))
}
const publish = Effect.fn("SessionRunner.publishLLMEvent")(function* (
event: LLMEvent,
outputPaths: ReadonlyArray<string> = [],
error?: SessionError.Error,
) {
const publish = Effect.fn("SessionRunner.publishLLMEvent")(function* (event: LLMEvent, error?: SessionError.Error) {
switch (event.type) {
case "step-start":
yield* startAssistant()
@ -395,7 +394,6 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
assistantMessageID: tool.assistantMessageID,
callID: event.id,
...result,
outputPaths,
...(executed ? { result: event.result } : {}),
executed,
resultState,

8
packages/core/src/session/runner/to-llm-message.ts
View file

@ -69,7 +69,7 @@ const providerMetadata = (
): ProviderMetadata | undefined => (state === undefined ? undefined : { [provider]: state })
const toolInput = (tool: SessionMessage.AssistantTool) =>
tool.state.status === "pending"
tool.state.status === "streaming"
? Option.getOrElse(decodeToolInput(tool.state.input), () => tool.state.input)
: tool.state.input
@ -168,7 +168,7 @@ const assistant = (message: SessionMessage.Assistant, model: ModelV2.Ref) => {
]
}
function toLLMMessage(message: SessionMessage.Message, model: ModelV2.Ref): Message[] {
function toLLMMessage(message: SessionMessage.Info, model: ModelV2.Ref): Message[] {
switch (message.type) {
case "agent-switched":
case "model-switched":
@ -202,7 +202,7 @@ function toLLMMessage(message: SessionMessage.Message, model: ModelV2.Ref): Mess
Message.make({
id: message.id,
role: "user",
content: `Shell command: ${message.shell.command}\n\n${message.output?.output ?? ""}`,
content: `Shell command: ${message.command}\n\n${message.output?.output ?? ""}`,
metadata: message.metadata,
}),
]
@ -232,5 +232,5 @@ ${message.recent}
}
/** Translate projected V2 Session history into canonical @opencode-ai/llm context. */
export const toLLMMessages = (messages: readonly SessionMessage.Message[], model: ModelV2.Ref) =>
export const toLLMMessages = (messages: readonly SessionMessage.Info[], model: ModelV2.Ref) =>
messages.flatMap((message) => toLLMMessage(message, model))

13
packages/core/src/session/sql.ts
View file

@ -5,7 +5,7 @@ import { ProjectTable } from "../project/sql"
import type { SessionMessage } from "./message"
import type { Prompt } from "@opencode-ai/schema/prompt"
import type { SessionInput } from "./input"
import type { Snapshot } from "../snapshot"
import type { FileDiff } from "@opencode-ai/schema/file-diff"
import { PermissionV1 } from "../v1/permission"
import { ProjectV2 } from "../project"
import type { SessionSchema } from "./schema"
@ -13,10 +13,11 @@ import type { MessageID, PartID, SessionV1 } from "../v1/session"
import { WorkspaceV2 } from "../workspace"
import { Timestamps } from "../database/schema.sql"
import type { Instructions } from "../instructions/index"
import type { Revert } from "@opencode-ai/schema/revert"
import type { Session } from "@opencode-ai/schema/session"
import type { RevertV1 } from "@opencode-ai/schema/session-revert"
import type { Schema } from "effect"
type SessionMessageData = Omit<(typeof SessionMessage.Message)["Encoded"], "type" | "id">
type SessionMessageData = Omit<(typeof SessionMessage.Info)["Encoded"], "type" | "id">
type V1MessageData = Omit<SessionV1.Info, "id" | "sessionID">
type V1PartData = Omit<SessionV1.Part, "id" | "sessionID" | "messageID">
@ -41,7 +42,7 @@ export const SessionTable = sqliteTable(
summary_additions: integer(),
summary_deletions: integer(),
summary_files: integer(),
summary_diffs: text({ mode: "json" }).$type<Snapshot.LegacyFileDiff[]>(),
summary_diffs: text({ mode: "json" }).$type<FileDiff.LegacyInfo[]>(),
metadata: text({ mode: "json" }).$type<Record<string, unknown>>(),
cost: real().notNull().default(0),
tokens_input: integer().notNull().default(0),
@ -49,7 +50,7 @@ export const SessionTable = sqliteTable(
tokens_reasoning: integer().notNull().default(0),
tokens_cache_read: integer().notNull().default(0),
tokens_cache_write: integer().notNull().default(0),
revert: text({ mode: "json" }).$type<Revert.State>(),
revert: text({ mode: "json" }).$type<Session.Revert | RevertV1>(),
permission: text({ mode: "json" }).$type<PermissionV1.Ruleset>(),
agent: text(),
model: text({ mode: "json" }).$type<{
@ -148,7 +149,7 @@ export const SessionInputTable = sqliteTable(
.$type<SessionSchema.ID>()
.notNull()
.references(() => SessionTable.id, { onDelete: "cascade" }),
type: text().$type<SessionInput.Entry["type"]>().notNull(),
type: text().$type<SessionInput.Info["type"]>().notNull(),
prompt: text({ mode: "json" }).$type<Prompt>(),
delivery: text().$type<SessionInput.Delivery>(),
admitted_seq: integer().notNull(),

6
packages/core/src/session/store.ts
View file

@ -13,10 +13,10 @@ import { fromRow } from "./info"
export interface Interface {
readonly get: (sessionID: Session.ID) => Effect.Effect<Session.Info | undefined>
readonly context: (sessionID: Session.ID) => Effect.Effect<SessionMessage.Message[], MessageDecodeError>
readonly context: (sessionID: Session.ID) => Effect.Effect<SessionMessage.Info[], MessageDecodeError>
readonly message: (
messageID: SessionMessage.ID,
) => Effect.Effect<{ readonly sessionID: Session.ID; readonly message: SessionMessage.Message } | undefined>
) => Effect.Effect<{ readonly sessionID: Session.ID; readonly message: SessionMessage.Info } | undefined>
}
export class Service extends Context.Service<Service, Interface>()("@opencode/v2/SessionStore") {}
@ -25,7 +25,7 @@ const layer = Layer.effect(
Service,
Effect.gen(function* () {
const { db } = yield* Database.Service
const decodeMessage = Schema.decodeUnknownEffect(SessionMessage.Message)
const decodeMessage = Schema.decodeUnknownEffect(SessionMessage.Info)
return Service.of({
get: Effect.fn("SessionStore.get")(function* (sessionID) {