Use SyncEvent service at event call sites (#26782)
This commit is contained in:
Kit Langton
@@ -21,10 +21,11 @@ import { Question } from "@/question"
|
||||
import { errorMessage } from "@/util/error"
|
||||
import * as Log from "@opencode-ai/core/util/log"
|
||||
import { isRecord } from "@/util/record"
|
||||
import { EventV2 } from "@/v2/event"
|
||||
import { SyncEvent } from "@/sync"
|
||||
import { SessionEvent } from "@/v2/session-event"
|
||||
import { Modelv2 } from "@/v2/model"
|
||||
import * as DateTime from "effect/DateTime"
|
||||
import { Flag } from "@opencode-ai/core/flag/flag"
|
||||
|
||||
const DOOM_LOOP_THRESHOLD = 3
|
||||
const log = Log.create({ service: "session.processor" })
|
||||
@@ -96,6 +97,7 @@ export const layer: Layer.Layer<
|
||||
| Image.Service
|
||||
| SessionSummary.Service
|
||||
| SessionStatus.Service
|
||||
| SyncEvent.Service
|
||||
> = Layer.effect(
|
||||
Service,
|
||||
Effect.gen(function* () {
|
||||
@@ -111,6 +113,7 @@ export const layer: Layer.Layer<
|
||||
const scope = yield* Scope.Scope
|
||||
const status = yield* SessionStatus.Service
|
||||
const image = yield* Image.Service
|
||||
const sync = yield* SyncEvent.Service
|
||||
|
||||
const create = Effect.fn("SessionProcessor.create")(function* (input: Input) {
|
||||
// Pre-capture snapshot before the LLM stream starts. The AI SDK
|
||||
@@ -229,11 +232,13 @@ export const layer: Layer.Layer<
|
||||
case "reasoning-start":
|
||||
if (value.id in ctx.reasoningMap) return
|
||||
// TODO(v2): Temporary dual-write while migrating session messages to v2 events.
|
||||
EventV2.run(SessionEvent.Reasoning.Started.Sync, {
|
||||
sessionID: ctx.sessionID,
|
||||
reasoningID: value.id,
|
||||
timestamp: DateTime.makeUnsafe(Date.now()),
|
||||
})
|
||||
if (Flag.OPENCODE_EXPERIMENTAL_EVENT_SYSTEM) {
|
||||
yield* sync.run(SessionEvent.Reasoning.Started.Sync, {
|
||||
sessionID: ctx.sessionID,
|
||||
reasoningID: value.id,
|
||||
timestamp: DateTime.makeUnsafe(Date.now()),
|
||||
})
|
||||
}
|
||||
ctx.reasoningMap[value.id] = {
|
||||
id: PartID.ascending(),
|
||||
messageID: ctx.assistantMessage.id,
|
||||
@@ -262,12 +267,14 @@ export const layer: Layer.Layer<
|
||||
case "reasoning-end":
|
||||
if (!(value.id in ctx.reasoningMap)) return
|
||||
// TODO(v2): Temporary dual-write while migrating session messages to v2 events.
|
||||
EventV2.run(SessionEvent.Reasoning.Ended.Sync, {
|
||||
sessionID: ctx.sessionID,
|
||||
reasoningID: value.id,
|
||||
text: ctx.reasoningMap[value.id].text,
|
||||
timestamp: DateTime.makeUnsafe(Date.now()),
|
||||
})
|
||||
if (Flag.OPENCODE_EXPERIMENTAL_EVENT_SYSTEM) {
|
||||
yield* sync.run(SessionEvent.Reasoning.Ended.Sync, {
|
||||
sessionID: ctx.sessionID,
|
||||
reasoningID: value.id,
|
||||
text: ctx.reasoningMap[value.id].text,
|
||||
timestamp: DateTime.makeUnsafe(Date.now()),
|
||||
})
|
||||
}
|
||||
// oxlint-disable-next-line no-self-assign -- reactivity trigger
|
||||
ctx.reasoningMap[value.id].text = ctx.reasoningMap[value.id].text
|
||||
ctx.reasoningMap[value.id].time = { ...ctx.reasoningMap[value.id].time, end: Date.now() }
|
||||
@@ -281,12 +288,14 @@ export const layer: Layer.Layer<
|
||||
throw new Error(`Tool call not allowed while generating summary: ${value.toolName}`)
|
||||
}
|
||||
// TODO(v2): Temporary dual-write while migrating session messages to v2 events.
|
||||
EventV2.run(SessionEvent.Tool.Input.Started.Sync, {
|
||||
sessionID: ctx.sessionID,
|
||||
callID: value.id,
|
||||
name: value.toolName,
|
||||
timestamp: DateTime.makeUnsafe(Date.now()),
|
||||
})
|
||||
if (Flag.OPENCODE_EXPERIMENTAL_EVENT_SYSTEM) {
|
||||
yield* sync.run(SessionEvent.Tool.Input.Started.Sync, {
|
||||
sessionID: ctx.sessionID,
|
||||
callID: value.id,
|
||||
name: value.toolName,
|
||||
timestamp: DateTime.makeUnsafe(Date.now()),
|
||||
})
|
||||
}
|
||||
const part = yield* session.updatePart({
|
||||
id: ctx.toolcalls[value.id]?.partID ?? PartID.ascending(),
|
||||
messageID: ctx.assistantMessage.id,
|
||||
@@ -310,12 +319,14 @@ export const layer: Layer.Layer<
|
||||
|
||||
case "tool-input-end": {
|
||||
// TODO(v2): Temporary dual-write while migrating session messages to v2 events.
|
||||
EventV2.run(SessionEvent.Tool.Input.Ended.Sync, {
|
||||
sessionID: ctx.sessionID,
|
||||
callID: value.id,
|
||||
text: "",
|
||||
timestamp: DateTime.makeUnsafe(Date.now()),
|
||||
})
|
||||
if (Flag.OPENCODE_EXPERIMENTAL_EVENT_SYSTEM) {
|
||||
yield* sync.run(SessionEvent.Tool.Input.Ended.Sync, {
|
||||
sessionID: ctx.sessionID,
|
||||
callID: value.id,
|
||||
text: "",
|
||||
timestamp: DateTime.makeUnsafe(Date.now()),
|
||||
})
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
@@ -325,17 +336,19 @@ export const layer: Layer.Layer<
|
||||
}
|
||||
const toolCall = yield* readToolCall(value.toolCallId)
|
||||
// TODO(v2): Temporary dual-write while migrating session messages to v2 events.
|
||||
EventV2.run(SessionEvent.Tool.Called.Sync, {
|
||||
sessionID: ctx.sessionID,
|
||||
callID: value.toolCallId,
|
||||
tool: value.toolName,
|
||||
input: value.input,
|
||||
provider: {
|
||||
executed: toolCall?.part.metadata?.providerExecuted === true,
|
||||
...(value.providerMetadata ? { metadata: value.providerMetadata } : {}),
|
||||
},
|
||||
timestamp: DateTime.makeUnsafe(Date.now()),
|
||||
})
|
||||
if (Flag.OPENCODE_EXPERIMENTAL_EVENT_SYSTEM) {
|
||||
yield* sync.run(SessionEvent.Tool.Called.Sync, {
|
||||
sessionID: ctx.sessionID,
|
||||
callID: value.toolCallId,
|
||||
tool: value.toolName,
|
||||
input: value.input,
|
||||
provider: {
|
||||
executed: toolCall?.part.metadata?.providerExecuted === true,
|
||||
...(value.providerMetadata ? { metadata: value.providerMetadata } : {}),
|
||||
},
|
||||
timestamp: DateTime.makeUnsafe(Date.now()),
|
||||
})
|
||||
}
|
||||
yield* updateToolCall(value.toolCallId, (match) => ({
|
||||
...match,
|
||||
tool: value.toolName,
|
||||
@@ -405,27 +418,29 @@ export const layer: Layer.Layer<
|
||||
attachments: attachments?.length ? attachments : undefined,
|
||||
}
|
||||
// TODO(v2): Temporary dual-write while migrating session messages to v2 events.
|
||||
EventV2.run(SessionEvent.Tool.Success.Sync, {
|
||||
sessionID: ctx.sessionID,
|
||||
callID: value.toolCallId,
|
||||
structured: output.metadata,
|
||||
content: [
|
||||
{
|
||||
type: "text",
|
||||
text: output.output,
|
||||
if (Flag.OPENCODE_EXPERIMENTAL_EVENT_SYSTEM) {
|
||||
yield* sync.run(SessionEvent.Tool.Success.Sync, {
|
||||
sessionID: ctx.sessionID,
|
||||
callID: value.toolCallId,
|
||||
structured: output.metadata,
|
||||
content: [
|
||||
{
|
||||
type: "text",
|
||||
text: output.output,
|
||||
},
|
||||
...(output.attachments?.map((item: MessageV2.FilePart) => ({
|
||||
type: "file",
|
||||
uri: item.url,
|
||||
mime: item.mime,
|
||||
name: item.filename,
|
||||
})) ?? []),
|
||||
],
|
||||
provider: {
|
||||
executed: toolCall?.part.metadata?.providerExecuted === true,
|
||||
},
|
||||
...(output.attachments?.map((item: MessageV2.FilePart) => ({
|
||||
type: "file",
|
||||
uri: item.url,
|
||||
mime: item.mime,
|
||||
name: item.filename,
|
||||
})) ?? []),
|
||||
],
|
||||
provider: {
|
||||
executed: toolCall?.part.metadata?.providerExecuted === true,
|
||||
},
|
||||
timestamp: DateTime.makeUnsafe(Date.now()),
|
||||
})
|
||||
timestamp: DateTime.makeUnsafe(Date.now()),
|
||||
})
|
||||
}
|
||||
yield* completeToolCall(value.toolCallId, output)
|
||||
return
|
||||
}
|
||||
@@ -433,18 +448,20 @@ export const layer: Layer.Layer<
|
||||
case "tool-error": {
|
||||
const toolCall = yield* readToolCall(value.toolCallId)
|
||||
// TODO(v2): Temporary dual-write while migrating session messages to v2 events.
|
||||
EventV2.run(SessionEvent.Tool.Failed.Sync, {
|
||||
sessionID: ctx.sessionID,
|
||||
callID: value.toolCallId,
|
||||
error: {
|
||||
type: "unknown",
|
||||
message: errorMessage(value.error),
|
||||
},
|
||||
provider: {
|
||||
executed: toolCall?.part.metadata?.providerExecuted === true,
|
||||
},
|
||||
timestamp: DateTime.makeUnsafe(Date.now()),
|
||||
})
|
||||
if (Flag.OPENCODE_EXPERIMENTAL_EVENT_SYSTEM) {
|
||||
yield* sync.run(SessionEvent.Tool.Failed.Sync, {
|
||||
sessionID: ctx.sessionID,
|
||||
callID: value.toolCallId,
|
||||
error: {
|
||||
type: "unknown",
|
||||
message: errorMessage(value.error),
|
||||
},
|
||||
provider: {
|
||||
executed: toolCall?.part.metadata?.providerExecuted === true,
|
||||
},
|
||||
timestamp: DateTime.makeUnsafe(Date.now()),
|
||||
})
|
||||
}
|
||||
yield* failToolCall(value.toolCallId, value.error)
|
||||
return
|
||||
}
|
||||
@@ -456,17 +473,19 @@ export const layer: Layer.Layer<
|
||||
if (!ctx.snapshot) ctx.snapshot = yield* snapshot.track()
|
||||
if (!ctx.assistantMessage.summary) {
|
||||
// TODO(v2): Temporary dual-write while migrating session messages to v2 events.
|
||||
EventV2.run(SessionEvent.Step.Started.Sync, {
|
||||
sessionID: ctx.sessionID,
|
||||
agent: input.assistantMessage.agent,
|
||||
model: {
|
||||
id: Modelv2.ID.make(ctx.model.id),
|
||||
providerID: Modelv2.ProviderID.make(ctx.model.providerID),
|
||||
variant: Modelv2.VariantID.make(input.assistantMessage.variant ?? "default"),
|
||||
},
|
||||
snapshot: ctx.snapshot,
|
||||
timestamp: DateTime.makeUnsafe(Date.now()),
|
||||
})
|
||||
if (Flag.OPENCODE_EXPERIMENTAL_EVENT_SYSTEM) {
|
||||
yield* sync.run(SessionEvent.Step.Started.Sync, {
|
||||
sessionID: ctx.sessionID,
|
||||
agent: input.assistantMessage.agent,
|
||||
model: {
|
||||
id: Modelv2.ID.make(ctx.model.id),
|
||||
providerID: Modelv2.ProviderID.make(ctx.model.providerID),
|
||||
variant: Modelv2.VariantID.make(input.assistantMessage.variant ?? "default"),
|
||||
},
|
||||
snapshot: ctx.snapshot,
|
||||
timestamp: DateTime.makeUnsafe(Date.now()),
|
||||
})
|
||||
}
|
||||
}
|
||||
yield* session.updatePart({
|
||||
id: PartID.ascending(),
|
||||
@@ -486,14 +505,16 @@ export const layer: Layer.Layer<
|
||||
})
|
||||
if (!ctx.assistantMessage.summary) {
|
||||
// TODO(v2): Temporary dual-write while migrating session messages to v2 events.
|
||||
EventV2.run(SessionEvent.Step.Ended.Sync, {
|
||||
sessionID: ctx.sessionID,
|
||||
finish: value.finishReason,
|
||||
cost: usage.cost,
|
||||
tokens: usage.tokens,
|
||||
snapshot: completedSnapshot,
|
||||
timestamp: DateTime.makeUnsafe(Date.now()),
|
||||
})
|
||||
if (Flag.OPENCODE_EXPERIMENTAL_EVENT_SYSTEM) {
|
||||
yield* sync.run(SessionEvent.Step.Ended.Sync, {
|
||||
sessionID: ctx.sessionID,
|
||||
finish: value.finishReason,
|
||||
cost: usage.cost,
|
||||
tokens: usage.tokens,
|
||||
snapshot: completedSnapshot,
|
||||
timestamp: DateTime.makeUnsafe(Date.now()),
|
||||
})
|
||||
}
|
||||
}
|
||||
ctx.assistantMessage.finish = value.finishReason
|
||||
ctx.assistantMessage.cost += usage.cost
|
||||
@@ -541,10 +562,12 @@ export const layer: Layer.Layer<
|
||||
case "text-start":
|
||||
if (!ctx.assistantMessage.summary) {
|
||||
// TODO(v2): Temporary dual-write while migrating session messages to v2 events.
|
||||
EventV2.run(SessionEvent.Text.Started.Sync, {
|
||||
sessionID: ctx.sessionID,
|
||||
timestamp: DateTime.makeUnsafe(Date.now()),
|
||||
})
|
||||
if (Flag.OPENCODE_EXPERIMENTAL_EVENT_SYSTEM) {
|
||||
yield* sync.run(SessionEvent.Text.Started.Sync, {
|
||||
sessionID: ctx.sessionID,
|
||||
timestamp: DateTime.makeUnsafe(Date.now()),
|
||||
})
|
||||
}
|
||||
}
|
||||
ctx.currentText = {
|
||||
id: PartID.ascending(),
|
||||
@@ -586,11 +609,13 @@ export const layer: Layer.Layer<
|
||||
)).text
|
||||
if (!ctx.assistantMessage.summary) {
|
||||
// TODO(v2): Temporary dual-write while migrating session messages to v2 events.
|
||||
EventV2.run(SessionEvent.Text.Ended.Sync, {
|
||||
sessionID: ctx.sessionID,
|
||||
text: ctx.currentText.text,
|
||||
timestamp: DateTime.makeUnsafe(Date.now()),
|
||||
})
|
||||
if (Flag.OPENCODE_EXPERIMENTAL_EVENT_SYSTEM) {
|
||||
yield* sync.run(SessionEvent.Text.Ended.Sync, {
|
||||
sessionID: ctx.sessionID,
|
||||
text: ctx.currentText.text,
|
||||
timestamp: DateTime.makeUnsafe(Date.now()),
|
||||
})
|
||||
}
|
||||
}
|
||||
{
|
||||
const end = Date.now()
|
||||
@@ -680,14 +705,16 @@ export const layer: Layer.Layer<
|
||||
}
|
||||
if (!ctx.assistantMessage.summary) {
|
||||
// TODO(v2): Temporary dual-write while migrating session messages to v2 events.
|
||||
EventV2.run(SessionEvent.Step.Failed.Sync, {
|
||||
sessionID: ctx.sessionID,
|
||||
error: {
|
||||
type: "unknown",
|
||||
message: errorMessage(e),
|
||||
},
|
||||
timestamp: DateTime.makeUnsafe(Date.now()),
|
||||
})
|
||||
if (Flag.OPENCODE_EXPERIMENTAL_EVENT_SYSTEM) {
|
||||
yield* sync.run(SessionEvent.Step.Failed.Sync, {
|
||||
sessionID: ctx.sessionID,
|
||||
error: {
|
||||
type: "unknown",
|
||||
message: errorMessage(e),
|
||||
},
|
||||
timestamp: DateTime.makeUnsafe(Date.now()),
|
||||
})
|
||||
}
|
||||
}
|
||||
ctx.assistantMessage.error = error
|
||||
yield* bus.publish(Session.Event.Error, {
|
||||
@@ -732,22 +759,28 @@ export const layer: Layer.Layer<
|
||||
parse,
|
||||
set: (info) => {
|
||||
// TODO(v2): Temporary dual-write while migrating session messages to v2 events.
|
||||
EventV2.run(SessionEvent.Retried.Sync, {
|
||||
sessionID: ctx.sessionID,
|
||||
attempt: info.attempt,
|
||||
error: {
|
||||
message: info.message,
|
||||
isRetryable: true,
|
||||
},
|
||||
timestamp: DateTime.makeUnsafe(Date.now()),
|
||||
})
|
||||
return status.set(ctx.sessionID, {
|
||||
type: "retry",
|
||||
attempt: info.attempt,
|
||||
message: info.message,
|
||||
action: info.action,
|
||||
next: info.next,
|
||||
})
|
||||
const event = Flag.OPENCODE_EXPERIMENTAL_EVENT_SYSTEM
|
||||
? sync.run(SessionEvent.Retried.Sync, {
|
||||
sessionID: ctx.sessionID,
|
||||
attempt: info.attempt,
|
||||
error: {
|
||||
message: info.message,
|
||||
isRetryable: true,
|
||||
},
|
||||
timestamp: DateTime.makeUnsafe(Date.now()),
|
||||
})
|
||||
: Effect.void
|
||||
return event.pipe(
|
||||
Effect.andThen(
|
||||
status.set(ctx.sessionID, {
|
||||
type: "retry",
|
||||
attempt: info.attempt,
|
||||
message: info.message,
|
||||
action: info.action,
|
||||
next: info.next,
|
||||
}),
|
||||
),
|
||||
)
|
||||
},
|
||||
}),
|
||||
),
|
||||
@@ -788,6 +821,7 @@ export const defaultLayer = Layer.suspend(() =>
|
||||
Layer.provide(Image.defaultLayer),
|
||||
Layer.provide(Bus.layer),
|
||||
Layer.provide(Config.defaultLayer),
|
||||
Layer.provide(SyncEvent.defaultLayer),
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user