refactor(v2): tag session unions and exhaustively match events (#23201)
This commit is contained in:
@@ -1,6 +1,6 @@
|
|||||||
import { Schema } from "effect"
|
import { Schema } from "effect"
|
||||||
import { SessionEvent } from "./session-event"
|
import { SessionEvent } from "./session-event"
|
||||||
import { produce } from "immer"
|
import { castDraft, produce } from "immer"
|
||||||
|
|
||||||
export const ID = SessionEvent.ID
|
export const ID = SessionEvent.ID
|
||||||
export type ID = Schema.Schema.Type<typeof ID>
|
export type ID = Schema.Schema.Type<typeof ID>
|
||||||
@@ -70,7 +70,9 @@ export class ToolStateError extends Schema.Class<ToolStateError>("Session.Entry.
|
|||||||
metadata: Schema.Record(Schema.String, Schema.Unknown).pipe(Schema.optional),
|
metadata: Schema.Record(Schema.String, Schema.Unknown).pipe(Schema.optional),
|
||||||
}) {}
|
}) {}
|
||||||
|
|
||||||
export const ToolState = Schema.Union([ToolStatePending, ToolStateRunning, ToolStateCompleted, ToolStateError])
|
export const ToolState = Schema.Union([ToolStatePending, ToolStateRunning, ToolStateCompleted, ToolStateError]).pipe(
|
||||||
|
Schema.toTaggedUnion("status"),
|
||||||
|
)
|
||||||
export type ToolState = Schema.Schema.Type<typeof ToolState>
|
export type ToolState = Schema.Schema.Type<typeof ToolState>
|
||||||
|
|
||||||
export class AssistantTool extends Schema.Class<AssistantTool>("Session.Entry.Assistant.Tool")({
|
export class AssistantTool extends Schema.Class<AssistantTool>("Session.Entry.Assistant.Tool")({
|
||||||
@@ -96,7 +98,9 @@ export class AssistantReasoning extends Schema.Class<AssistantReasoning>("Sessio
|
|||||||
text: Schema.String,
|
text: Schema.String,
|
||||||
}) {}
|
}) {}
|
||||||
|
|
||||||
export const AssistantContent = Schema.Union([AssistantText, AssistantReasoning, AssistantTool])
|
export const AssistantContent = Schema.Union([AssistantText, AssistantReasoning, AssistantTool]).pipe(
|
||||||
|
Schema.toTaggedUnion("type"),
|
||||||
|
)
|
||||||
export type AssistantContent = Schema.Schema.Type<typeof AssistantContent>
|
export type AssistantContent = Schema.Schema.Type<typeof AssistantContent>
|
||||||
|
|
||||||
export class Assistant extends Schema.Class<Assistant>("Session.Entry.Assistant")({
|
export class Assistant extends Schema.Class<Assistant>("Session.Entry.Assistant")({
|
||||||
@@ -126,7 +130,7 @@ export class Compaction extends Schema.Class<Compaction>("Session.Entry.Compacti
|
|||||||
...Base,
|
...Base,
|
||||||
}) {}
|
}) {}
|
||||||
|
|
||||||
export const Entry = Schema.Union([User, Synthetic, Assistant, Compaction])
|
export const Entry = Schema.Union([User, Synthetic, Assistant, Compaction]).pipe(Schema.toTaggedUnion("type"))
|
||||||
|
|
||||||
export type Entry = Schema.Schema.Type<typeof Entry>
|
export type Entry = Schema.Schema.Type<typeof Entry>
|
||||||
|
|
||||||
@@ -141,19 +145,29 @@ export function step(old: History, event: SessionEvent.Event): History {
|
|||||||
return produce(old, (draft) => {
|
return produce(old, (draft) => {
|
||||||
const lastAssistant = draft.entries.findLast((x) => x.type === "assistant")
|
const lastAssistant = draft.entries.findLast((x) => x.type === "assistant")
|
||||||
const pendingAssistant = lastAssistant && !lastAssistant.time.completed ? lastAssistant : undefined
|
const pendingAssistant = lastAssistant && !lastAssistant.time.completed ? lastAssistant : undefined
|
||||||
|
type DraftContent = NonNullable<typeof pendingAssistant>["content"][number]
|
||||||
|
type DraftTool = Extract<DraftContent, { type: "tool" }>
|
||||||
|
|
||||||
switch (event.type) {
|
const latestTool = (callID?: string) =>
|
||||||
case "prompt": {
|
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) {
|
if (pendingAssistant) {
|
||||||
// @ts-expect-error
|
draft.pending.push(castDraft(entry))
|
||||||
draft.pending.push(User.fromEvent(event))
|
return
|
||||||
break
|
|
||||||
}
|
}
|
||||||
// @ts-expect-error
|
draft.entries.push(castDraft(entry))
|
||||||
draft.entries.push(User.fromEvent(event))
|
},
|
||||||
break
|
synthetic: (event) => {
|
||||||
}
|
draft.entries.push(new Synthetic({ ...event, time: { created: event.timestamp } }))
|
||||||
case "step.started": {
|
},
|
||||||
|
"step.started": (event) => {
|
||||||
if (pendingAssistant) pendingAssistant.time.completed = event.timestamp
|
if (pendingAssistant) pendingAssistant.time.completed = event.timestamp
|
||||||
draft.entries.push({
|
draft.entries.push({
|
||||||
id: event.id,
|
id: event.id,
|
||||||
@@ -163,27 +177,28 @@ export function step(old: History, event: SessionEvent.Event): History {
|
|||||||
},
|
},
|
||||||
content: [],
|
content: [],
|
||||||
})
|
})
|
||||||
break
|
},
|
||||||
}
|
"step.ended": (event) => {
|
||||||
case "text.started": {
|
if (!pendingAssistant) return
|
||||||
if (!pendingAssistant) break
|
pendingAssistant.time.completed = event.timestamp
|
||||||
|
pendingAssistant.cost = event.cost
|
||||||
|
pendingAssistant.tokens = event.tokens
|
||||||
|
},
|
||||||
|
"text.started": () => {
|
||||||
|
if (!pendingAssistant) return
|
||||||
pendingAssistant.content.push({
|
pendingAssistant.content.push({
|
||||||
type: "text",
|
type: "text",
|
||||||
text: "",
|
text: "",
|
||||||
})
|
})
|
||||||
break
|
},
|
||||||
}
|
"text.delta": (event) => {
|
||||||
case "text.delta": {
|
if (!pendingAssistant) return
|
||||||
if (!pendingAssistant) break
|
const match = latestText()
|
||||||
const match = pendingAssistant.content.findLast((x) => x.type === "text")
|
|
||||||
if (match) match.text += event.delta
|
if (match) match.text += event.delta
|
||||||
break
|
},
|
||||||
}
|
"text.ended": () => {},
|
||||||
case "text.ended": {
|
"tool.input.started": (event) => {
|
||||||
break
|
if (!pendingAssistant) return
|
||||||
}
|
|
||||||
case "tool.input.started": {
|
|
||||||
if (!pendingAssistant) break
|
|
||||||
pendingAssistant.content.push({
|
pendingAssistant.content.push({
|
||||||
type: "tool",
|
type: "tool",
|
||||||
callID: event.callID,
|
callID: event.callID,
|
||||||
@@ -196,21 +211,17 @@ export function step(old: History, event: SessionEvent.Event): History {
|
|||||||
input: "",
|
input: "",
|
||||||
},
|
},
|
||||||
})
|
})
|
||||||
break
|
},
|
||||||
}
|
"tool.input.delta": (event) => {
|
||||||
case "tool.input.delta": {
|
if (!pendingAssistant) return
|
||||||
if (!pendingAssistant) break
|
const match = latestTool(event.callID)
|
||||||
const match = pendingAssistant.content.findLast((x) => x.type === "tool")
|
|
||||||
// oxlint-disable-next-line no-base-to-string -- event.delta is a Schema.String (runtime string)
|
// oxlint-disable-next-line no-base-to-string -- event.delta is a Schema.String (runtime string)
|
||||||
if (match) match.state.input += event.delta
|
if (match) match.state.input += event.delta
|
||||||
break
|
},
|
||||||
}
|
"tool.input.ended": () => {},
|
||||||
case "tool.input.ended": {
|
"tool.called": (event) => {
|
||||||
break
|
if (!pendingAssistant) return
|
||||||
}
|
const match = latestTool(event.callID)
|
||||||
case "tool.called": {
|
|
||||||
if (!pendingAssistant) break
|
|
||||||
const match = pendingAssistant.content.findLast((x) => x.type === "tool")
|
|
||||||
if (match) {
|
if (match) {
|
||||||
match.time.ran = event.timestamp
|
match.time.ran = event.timestamp
|
||||||
match.state = {
|
match.state = {
|
||||||
@@ -218,11 +229,10 @@ export function step(old: History, event: SessionEvent.Event): History {
|
|||||||
input: event.input,
|
input: event.input,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
break
|
},
|
||||||
}
|
"tool.success": (event) => {
|
||||||
case "tool.success": {
|
if (!pendingAssistant) return
|
||||||
if (!pendingAssistant) break
|
const match = latestTool(event.callID)
|
||||||
const match = pendingAssistant.content.findLast((x) => x.type === "tool")
|
|
||||||
if (match && match.state.status === "running") {
|
if (match && match.state.status === "running") {
|
||||||
match.state = {
|
match.state = {
|
||||||
status: "completed",
|
status: "completed",
|
||||||
@@ -230,15 +240,13 @@ export function step(old: History, event: SessionEvent.Event): History {
|
|||||||
output: event.output ?? "",
|
output: event.output ?? "",
|
||||||
title: event.title,
|
title: event.title,
|
||||||
metadata: event.metadata ?? {},
|
metadata: event.metadata ?? {},
|
||||||
// @ts-expect-error
|
attachments: [...(event.attachments ?? [])],
|
||||||
attachments: event.attachments ?? [],
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
break
|
},
|
||||||
}
|
"tool.error": (event) => {
|
||||||
case "tool.error": {
|
if (!pendingAssistant) return
|
||||||
if (!pendingAssistant) break
|
const match = latestTool(event.callID)
|
||||||
const match = pendingAssistant.content.findLast((x) => x.type === "tool")
|
|
||||||
if (match && match.state.status === "running") {
|
if (match && match.state.status === "running") {
|
||||||
match.state = {
|
match.state = {
|
||||||
status: "error",
|
status: "error",
|
||||||
@@ -247,36 +255,29 @@ export function step(old: History, event: SessionEvent.Event): History {
|
|||||||
metadata: event.metadata ?? {},
|
metadata: event.metadata ?? {},
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
break
|
},
|
||||||
}
|
"reasoning.started": () => {
|
||||||
case "reasoning.started": {
|
if (!pendingAssistant) return
|
||||||
if (!pendingAssistant) break
|
|
||||||
pendingAssistant.content.push({
|
pendingAssistant.content.push({
|
||||||
type: "reasoning",
|
type: "reasoning",
|
||||||
text: "",
|
text: "",
|
||||||
})
|
})
|
||||||
break
|
},
|
||||||
}
|
"reasoning.delta": (event) => {
|
||||||
case "reasoning.delta": {
|
if (!pendingAssistant) return
|
||||||
if (!pendingAssistant) break
|
const match = latestReasoning()
|
||||||
const match = pendingAssistant.content.findLast((x) => x.type === "reasoning")
|
|
||||||
if (match) match.text += event.delta
|
if (match) match.text += event.delta
|
||||||
break
|
},
|
||||||
}
|
"reasoning.ended": (event) => {
|
||||||
case "reasoning.ended": {
|
if (!pendingAssistant) return
|
||||||
if (!pendingAssistant) break
|
const match = latestReasoning()
|
||||||
const match = pendingAssistant.content.findLast((x) => x.type === "reasoning")
|
|
||||||
if (match) match.text = event.text
|
if (match) match.text = event.text
|
||||||
break
|
},
|
||||||
}
|
retried: () => {},
|
||||||
case "step.ended": {
|
compacted: (event) => {
|
||||||
if (!pendingAssistant) break
|
draft.entries.push(new Compaction({ ...event, type: "compaction", time: { created: event.timestamp } }))
|
||||||
pendingAssistant.time.completed = event.timestamp
|
},
|
||||||
pendingAssistant.cost = event.cost
|
})
|
||||||
pendingAssistant.tokens = event.tokens
|
|
||||||
break
|
|
||||||
}
|
|
||||||
}
|
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -441,7 +441,7 @@ export namespace SessionEvent {
|
|||||||
{
|
{
|
||||||
mode: "oneOf",
|
mode: "oneOf",
|
||||||
},
|
},
|
||||||
)
|
).pipe(Schema.toTaggedUnion("type"))
|
||||||
export type Event = Schema.Schema.Type<typeof Event>
|
export type Event = Schema.Schema.Type<typeof Event>
|
||||||
export type Type = Event["type"]
|
export type Type = Event["type"]
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -591,7 +591,64 @@ describe("session-entry step", () => {
|
|||||||
)
|
)
|
||||||
})
|
})
|
||||||
|
|
||||||
test.failing("records synthetic events", () => {
|
test("routes tool events by callID when tool streams interleave", () => {
|
||||||
|
FastCheck.assert(
|
||||||
|
FastCheck.property(dict, dict, word, word, text, text, (a, b, titleA, titleB, deltaA, deltaB) => {
|
||||||
|
const next = run(
|
||||||
|
[
|
||||||
|
SessionEvent.Tool.Input.Started.create({ callID: "a", name: "bash", timestamp: time(1) }),
|
||||||
|
SessionEvent.Tool.Input.Started.create({ callID: "b", name: "grep", timestamp: time(2) }),
|
||||||
|
SessionEvent.Tool.Input.Delta.create({ callID: "a", delta: deltaA, timestamp: time(3) }),
|
||||||
|
SessionEvent.Tool.Input.Delta.create({ callID: "b", delta: deltaB, timestamp: time(4) }),
|
||||||
|
SessionEvent.Tool.Called.create({
|
||||||
|
callID: "a",
|
||||||
|
tool: "bash",
|
||||||
|
input: a,
|
||||||
|
provider: { executed: true },
|
||||||
|
timestamp: time(5),
|
||||||
|
}),
|
||||||
|
SessionEvent.Tool.Called.create({
|
||||||
|
callID: "b",
|
||||||
|
tool: "grep",
|
||||||
|
input: b,
|
||||||
|
provider: { executed: true },
|
||||||
|
timestamp: time(6),
|
||||||
|
}),
|
||||||
|
SessionEvent.Tool.Success.create({
|
||||||
|
callID: "a",
|
||||||
|
title: titleA,
|
||||||
|
output: "done-a",
|
||||||
|
provider: { executed: true },
|
||||||
|
timestamp: time(7),
|
||||||
|
}),
|
||||||
|
SessionEvent.Tool.Success.create({
|
||||||
|
callID: "b",
|
||||||
|
title: titleB,
|
||||||
|
output: "done-b",
|
||||||
|
provider: { executed: true },
|
||||||
|
timestamp: time(8),
|
||||||
|
}),
|
||||||
|
],
|
||||||
|
active(),
|
||||||
|
)
|
||||||
|
|
||||||
|
const first = tool(next, "a")
|
||||||
|
const second = tool(next, "b")
|
||||||
|
|
||||||
|
expect(first?.state.status).toBe("completed")
|
||||||
|
expect(second?.state.status).toBe("completed")
|
||||||
|
if (first?.state.status !== "completed" || second?.state.status !== "completed") return
|
||||||
|
|
||||||
|
expect(first.state.input).toEqual(a)
|
||||||
|
expect(second.state.input).toEqual(b)
|
||||||
|
expect(first.state.title).toBe(titleA)
|
||||||
|
expect(second.state.title).toBe(titleB)
|
||||||
|
}),
|
||||||
|
{ numRuns: 50 },
|
||||||
|
)
|
||||||
|
})
|
||||||
|
|
||||||
|
test("records synthetic events", () => {
|
||||||
FastCheck.assert(
|
FastCheck.assert(
|
||||||
FastCheck.property(word, (body) => {
|
FastCheck.property(word, (body) => {
|
||||||
const next = SessionEntry.step(history(), SessionEvent.Synthetic.create({ text: body, timestamp: time(1) }))
|
const next = SessionEntry.step(history(), SessionEvent.Synthetic.create({ text: body, timestamp: time(1) }))
|
||||||
@@ -604,7 +661,7 @@ describe("session-entry step", () => {
|
|||||||
)
|
)
|
||||||
})
|
})
|
||||||
|
|
||||||
test.failing("records compaction events", () => {
|
test("records compaction events", () => {
|
||||||
FastCheck.assert(
|
FastCheck.assert(
|
||||||
FastCheck.property(FastCheck.boolean(), maybe(FastCheck.boolean()), (auto, overflow) => {
|
FastCheck.property(FastCheck.boolean(), maybe(FastCheck.boolean()), (auto, overflow) => {
|
||||||
const next = SessionEntry.step(
|
const next = SessionEntry.step(
|
||||||
|
|||||||
Reference in New Issue
Block a user