core: extract session entry stepping logic into dedicated module
Move the step function from session-entry.ts to session-entry-stepper.ts and remove immer dependency. Add static fromEvent factory methods to Synthetic, Assistant, and Compaction classes for cleaner event-to-entry conversion.
This commit is contained in:
Dax Raad
253
packages/opencode/src/v2/session-entry-stepper.ts
Normal file
253
packages/opencode/src/v2/session-entry-stepper.ts
Normal file
@@ -0,0 +1,253 @@
|
||||
import { castDraft, produce, type WritableDraft } from "immer"
|
||||
import { SessionEvent } from "./session-event"
|
||||
import { SessionEntry } from "./session-entry"
|
||||
|
||||
export type MemoryState = {
|
||||
entries: SessionEntry.Entry[]
|
||||
pending: SessionEntry.Entry[]
|
||||
}
|
||||
|
||||
export interface Adapter<Result> {
|
||||
readonly getCurrentAssistant: () => SessionEntry.Assistant | undefined
|
||||
readonly updateAssistant: (assistant: SessionEntry.Assistant) => void
|
||||
readonly appendEntry: (entry: SessionEntry.Entry) => void
|
||||
readonly appendPending: (entry: SessionEntry.Entry) => void
|
||||
readonly finish: () => Result
|
||||
}
|
||||
|
||||
export function memory(state: MemoryState): Adapter<MemoryState> {
|
||||
const activeAssistantIndex = () =>
|
||||
state.entries.findLastIndex((entry) => entry.type === "assistant" && !entry.time.completed)
|
||||
|
||||
return {
|
||||
getCurrentAssistant() {
|
||||
const index = activeAssistantIndex()
|
||||
if (index < 0) return
|
||||
const assistant = state.entries[index]
|
||||
return assistant?.type === "assistant" ? assistant : undefined
|
||||
},
|
||||
updateAssistant(assistant) {
|
||||
const index = activeAssistantIndex()
|
||||
if (index < 0) return
|
||||
const current = state.entries[index]
|
||||
if (current?.type !== "assistant") return
|
||||
state.entries[index] = assistant
|
||||
},
|
||||
appendEntry(entry) {
|
||||
state.entries.push(entry)
|
||||
},
|
||||
appendPending(entry) {
|
||||
state.pending.push(entry)
|
||||
},
|
||||
finish() {
|
||||
return state
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
export function stepWith<Result>(adapter: Adapter<Result>, event: SessionEvent.Event): Result {
|
||||
const currentAssistant = adapter.getCurrentAssistant()
|
||||
type DraftAssistant = WritableDraft<SessionEntry.Assistant>
|
||||
type DraftTool = WritableDraft<SessionEntry.AssistantTool>
|
||||
type DraftText = WritableDraft<SessionEntry.AssistantText>
|
||||
type DraftReasoning = WritableDraft<SessionEntry.AssistantReasoning>
|
||||
|
||||
const latestTool = (assistant: DraftAssistant | undefined, callID?: string) =>
|
||||
assistant?.content.findLast(
|
||||
(item): item is DraftTool => item.type === "tool" && (callID === undefined || item.callID === callID),
|
||||
)
|
||||
|
||||
const latestText = (assistant: DraftAssistant | undefined) =>
|
||||
assistant?.content.findLast((item): item is DraftText => item.type === "text")
|
||||
|
||||
const latestReasoning = (assistant: DraftAssistant | undefined) =>
|
||||
assistant?.content.findLast((item): item is DraftReasoning => item.type === "reasoning")
|
||||
|
||||
SessionEvent.Event.match(event, {
|
||||
prompt: (event) => {
|
||||
const entry = SessionEntry.User.fromEvent(event)
|
||||
if (currentAssistant) {
|
||||
adapter.appendPending(entry)
|
||||
return
|
||||
}
|
||||
adapter.appendEntry(entry)
|
||||
},
|
||||
synthetic: (event) => {
|
||||
adapter.appendEntry(SessionEntry.Synthetic.fromEvent(event))
|
||||
},
|
||||
"step.started": (event) => {
|
||||
if (currentAssistant) {
|
||||
adapter.updateAssistant(
|
||||
produce(currentAssistant, (draft) => {
|
||||
draft.time.completed = event.timestamp
|
||||
}),
|
||||
)
|
||||
}
|
||||
adapter.appendEntry(SessionEntry.Assistant.fromEvent(event))
|
||||
},
|
||||
"step.ended": (event) => {
|
||||
if (currentAssistant) {
|
||||
adapter.updateAssistant(
|
||||
produce(currentAssistant, (draft) => {
|
||||
draft.time.completed = event.timestamp
|
||||
draft.cost = event.cost
|
||||
draft.tokens = event.tokens
|
||||
}),
|
||||
)
|
||||
}
|
||||
},
|
||||
"text.started": () => {
|
||||
if (currentAssistant) {
|
||||
adapter.updateAssistant(
|
||||
produce(currentAssistant, (draft) => {
|
||||
draft.content.push({
|
||||
type: "text",
|
||||
text: "",
|
||||
})
|
||||
}),
|
||||
)
|
||||
}
|
||||
},
|
||||
"text.delta": (event) => {
|
||||
if (currentAssistant) {
|
||||
adapter.updateAssistant(
|
||||
produce(currentAssistant, (draft) => {
|
||||
const match = latestText(draft)
|
||||
if (match) match.text += event.delta
|
||||
}),
|
||||
)
|
||||
}
|
||||
},
|
||||
"text.ended": () => {},
|
||||
"tool.input.started": (event) => {
|
||||
if (currentAssistant) {
|
||||
adapter.updateAssistant(
|
||||
produce(currentAssistant, (draft) => {
|
||||
draft.content.push({
|
||||
type: "tool",
|
||||
callID: event.callID,
|
||||
name: event.name,
|
||||
time: {
|
||||
created: event.timestamp,
|
||||
},
|
||||
state: {
|
||||
status: "pending",
|
||||
input: "",
|
||||
},
|
||||
})
|
||||
}),
|
||||
)
|
||||
}
|
||||
},
|
||||
"tool.input.delta": (event) => {
|
||||
if (currentAssistant) {
|
||||
adapter.updateAssistant(
|
||||
produce(currentAssistant, (draft) => {
|
||||
const match = latestTool(draft, event.callID)
|
||||
// oxlint-disable-next-line no-base-to-string -- event.delta is a Schema.String (runtime string)
|
||||
if (match && match.state.status === "pending") match.state.input += event.delta
|
||||
}),
|
||||
)
|
||||
}
|
||||
},
|
||||
"tool.input.ended": () => {},
|
||||
"tool.called": (event) => {
|
||||
if (currentAssistant) {
|
||||
adapter.updateAssistant(
|
||||
produce(currentAssistant, (draft) => {
|
||||
const match = latestTool(draft, event.callID)
|
||||
if (match) {
|
||||
match.time.ran = event.timestamp
|
||||
match.state = {
|
||||
status: "running",
|
||||
input: event.input,
|
||||
}
|
||||
}
|
||||
}),
|
||||
)
|
||||
}
|
||||
},
|
||||
"tool.success": (event) => {
|
||||
if (currentAssistant) {
|
||||
adapter.updateAssistant(
|
||||
produce(currentAssistant, (draft) => {
|
||||
const match = latestTool(draft, event.callID)
|
||||
if (match && match.state.status === "running") {
|
||||
match.state = {
|
||||
status: "completed",
|
||||
input: match.state.input,
|
||||
output: event.output ?? "",
|
||||
title: event.title,
|
||||
metadata: event.metadata ?? {},
|
||||
attachments: [...(event.attachments ?? [])],
|
||||
}
|
||||
}
|
||||
}),
|
||||
)
|
||||
}
|
||||
},
|
||||
"tool.error": (event) => {
|
||||
if (currentAssistant) {
|
||||
adapter.updateAssistant(
|
||||
produce(currentAssistant, (draft) => {
|
||||
const match = latestTool(draft, event.callID)
|
||||
if (match && match.state.status === "running") {
|
||||
match.state = {
|
||||
status: "error",
|
||||
error: event.error,
|
||||
input: match.state.input,
|
||||
metadata: event.metadata ?? {},
|
||||
}
|
||||
}
|
||||
}),
|
||||
)
|
||||
}
|
||||
},
|
||||
"reasoning.started": () => {
|
||||
if (currentAssistant) {
|
||||
adapter.updateAssistant(
|
||||
produce(currentAssistant, (draft) => {
|
||||
draft.content.push({
|
||||
type: "reasoning",
|
||||
text: "",
|
||||
})
|
||||
}),
|
||||
)
|
||||
}
|
||||
},
|
||||
"reasoning.delta": (event) => {
|
||||
if (currentAssistant) {
|
||||
adapter.updateAssistant(
|
||||
produce(currentAssistant, (draft) => {
|
||||
const match = latestReasoning(draft)
|
||||
if (match) match.text += event.delta
|
||||
}),
|
||||
)
|
||||
}
|
||||
},
|
||||
"reasoning.ended": (event) => {
|
||||
if (currentAssistant) {
|
||||
adapter.updateAssistant(
|
||||
produce(currentAssistant, (draft) => {
|
||||
const match = latestReasoning(draft)
|
||||
if (match) match.text = event.text
|
||||
}),
|
||||
)
|
||||
}
|
||||
},
|
||||
retried: () => {},
|
||||
compacted: (event) => {
|
||||
adapter.appendEntry(SessionEntry.Compaction.fromEvent(event))
|
||||
},
|
||||
})
|
||||
|
||||
return adapter.finish()
|
||||
}
|
||||
|
||||
export function step(old: MemoryState, event: SessionEvent.Event): MemoryState {
|
||||
return produce(old, (draft) => {
|
||||
stepWith(memory(draft as MemoryState), event)
|
||||
})
|
||||
}
|
||||
|
||||
export * as SessionEntryStepper from "./session-entry-stepper"
|
||||
@@ -1,6 +1,5 @@
|
||||
import { Schema } from "effect"
|
||||
import { SessionEvent } from "./session-event"
|
||||
import { castDraft, produce } from "immer"
|
||||
|
||||
export const ID = SessionEvent.ID
|
||||
export type ID = Schema.Schema.Type<typeof ID>
|
||||
@@ -40,7 +39,14 @@ export class Synthetic extends Schema.Class<Synthetic>("Session.Entry.Synthetic"
|
||||
...SessionEvent.Synthetic.fields,
|
||||
...Base,
|
||||
type: Schema.Literal("synthetic"),
|
||||
}) {}
|
||||
}) {
|
||||
static fromEvent(event: SessionEvent.Synthetic) {
|
||||
return new Synthetic({
|
||||
...event,
|
||||
time: { created: event.timestamp },
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
export class ToolStatePending extends Schema.Class<ToolStatePending>("Session.Entry.ToolState.Pending")({
|
||||
status: Schema.Literal("pending"),
|
||||
@@ -122,13 +128,32 @@ export class Assistant extends Schema.Class<Assistant>("Session.Entry.Assistant"
|
||||
created: Schema.DateTimeUtc,
|
||||
completed: Schema.DateTimeUtc.pipe(Schema.optional),
|
||||
}),
|
||||
}) {}
|
||||
}) {
|
||||
static fromEvent(event: SessionEvent.Step.Started) {
|
||||
return new Assistant({
|
||||
id: event.id,
|
||||
type: "assistant",
|
||||
time: {
|
||||
created: event.timestamp,
|
||||
},
|
||||
content: [],
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
export class Compaction extends Schema.Class<Compaction>("Session.Entry.Compaction")({
|
||||
...SessionEvent.Compacted.fields,
|
||||
type: Schema.Literal("compaction"),
|
||||
...Base,
|
||||
}) {}
|
||||
}) {
|
||||
static fromEvent(event: SessionEvent.Compacted) {
|
||||
return new Compaction({
|
||||
...event,
|
||||
type: "compaction",
|
||||
time: { created: event.timestamp },
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
export const Entry = Schema.Union([User, Synthetic, Assistant, Compaction]).pipe(Schema.toTaggedUnion("type"))
|
||||
|
||||
@@ -136,151 +161,6 @@ export type Entry = Schema.Schema.Type<typeof Entry>
|
||||
|
||||
export type Type = Entry["type"]
|
||||
|
||||
export type History = {
|
||||
entries: Entry[]
|
||||
pending: Entry[]
|
||||
}
|
||||
|
||||
export function step(old: History, event: SessionEvent.Event): History {
|
||||
return produce(old, (draft) => {
|
||||
const lastAssistant = draft.entries.findLast((x) => x.type === "assistant")
|
||||
const pendingAssistant = lastAssistant && !lastAssistant.time.completed ? lastAssistant : undefined
|
||||
type DraftContent = NonNullable<typeof pendingAssistant>["content"][number]
|
||||
type DraftTool = Extract<DraftContent, { type: "tool" }>
|
||||
|
||||
const latestTool = (callID?: string) =>
|
||||
pendingAssistant?.content.findLast(
|
||||
(item): item is DraftTool => item.type === "tool" && (callID === undefined || item.callID === callID),
|
||||
)
|
||||
const latestText = () => pendingAssistant?.content.findLast((item) => item.type === "text")
|
||||
const latestReasoning = () => pendingAssistant?.content.findLast((item) => item.type === "reasoning")
|
||||
|
||||
SessionEvent.Event.match(event, {
|
||||
prompt: (event) => {
|
||||
const entry = User.fromEvent(event)
|
||||
if (pendingAssistant) {
|
||||
draft.pending.push(castDraft(entry))
|
||||
return
|
||||
}
|
||||
draft.entries.push(castDraft(entry))
|
||||
},
|
||||
synthetic: (event) => {
|
||||
draft.entries.push(new Synthetic({ ...event, time: { created: event.timestamp } }))
|
||||
},
|
||||
"step.started": (event) => {
|
||||
if (pendingAssistant) pendingAssistant.time.completed = event.timestamp
|
||||
draft.entries.push({
|
||||
id: event.id,
|
||||
type: "assistant",
|
||||
time: {
|
||||
created: event.timestamp,
|
||||
},
|
||||
content: [],
|
||||
})
|
||||
},
|
||||
"step.ended": (event) => {
|
||||
if (!pendingAssistant) return
|
||||
pendingAssistant.time.completed = event.timestamp
|
||||
pendingAssistant.cost = event.cost
|
||||
pendingAssistant.tokens = event.tokens
|
||||
},
|
||||
"text.started": () => {
|
||||
if (!pendingAssistant) return
|
||||
pendingAssistant.content.push({
|
||||
type: "text",
|
||||
text: "",
|
||||
})
|
||||
},
|
||||
"text.delta": (event) => {
|
||||
if (!pendingAssistant) return
|
||||
const match = latestText()
|
||||
if (match) match.text += event.delta
|
||||
},
|
||||
"text.ended": () => {},
|
||||
"tool.input.started": (event) => {
|
||||
if (!pendingAssistant) return
|
||||
pendingAssistant.content.push({
|
||||
type: "tool",
|
||||
callID: event.callID,
|
||||
name: event.name,
|
||||
time: {
|
||||
created: event.timestamp,
|
||||
},
|
||||
state: {
|
||||
status: "pending",
|
||||
input: "",
|
||||
},
|
||||
})
|
||||
},
|
||||
"tool.input.delta": (event) => {
|
||||
if (!pendingAssistant) return
|
||||
const match = latestTool(event.callID)
|
||||
// oxlint-disable-next-line no-base-to-string -- event.delta is a Schema.String (runtime string)
|
||||
if (match) match.state.input += event.delta
|
||||
},
|
||||
"tool.input.ended": () => {},
|
||||
"tool.called": (event) => {
|
||||
if (!pendingAssistant) return
|
||||
const match = latestTool(event.callID)
|
||||
if (match) {
|
||||
match.time.ran = event.timestamp
|
||||
match.state = {
|
||||
status: "running",
|
||||
input: event.input,
|
||||
}
|
||||
}
|
||||
},
|
||||
"tool.success": (event) => {
|
||||
if (!pendingAssistant) return
|
||||
const match = latestTool(event.callID)
|
||||
if (match && match.state.status === "running") {
|
||||
match.state = {
|
||||
status: "completed",
|
||||
input: match.state.input,
|
||||
output: event.output ?? "",
|
||||
title: event.title,
|
||||
metadata: event.metadata ?? {},
|
||||
attachments: [...(event.attachments ?? [])],
|
||||
}
|
||||
}
|
||||
},
|
||||
"tool.error": (event) => {
|
||||
if (!pendingAssistant) return
|
||||
const match = latestTool(event.callID)
|
||||
if (match && match.state.status === "running") {
|
||||
match.state = {
|
||||
status: "error",
|
||||
error: event.error,
|
||||
input: match.state.input,
|
||||
metadata: event.metadata ?? {},
|
||||
}
|
||||
}
|
||||
},
|
||||
"reasoning.started": () => {
|
||||
if (!pendingAssistant) return
|
||||
pendingAssistant.content.push({
|
||||
type: "reasoning",
|
||||
text: "",
|
||||
})
|
||||
},
|
||||
"reasoning.delta": (event) => {
|
||||
if (!pendingAssistant) return
|
||||
const match = latestReasoning()
|
||||
if (match) match.text += event.delta
|
||||
},
|
||||
"reasoning.ended": (event) => {
|
||||
if (!pendingAssistant) return
|
||||
const match = latestReasoning()
|
||||
if (match) match.text = event.text
|
||||
},
|
||||
retried: () => {},
|
||||
compacted: (event) => {
|
||||
draft.entries.push(new Compaction({ ...event, type: "compaction", time: { created: event.timestamp } }))
|
||||
},
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
/*
|
||||
export interface Interface {
|
||||
readonly decode: (row: typeof SessionEntryTable.$inferSelect) => Entry
|
||||
|
||||
Reference in New Issue
Block a user