Compare commits

...

1 Commits

Author SHA1 Message Date
Kit Langton aa5742330b refactor(core): simplify session runner bookkeeping 2026-07-09 23:17:54 -04:00
4 changed files with 11 additions and 28 deletions
+1 -1
View File
@@ -581,7 +581,7 @@ const layer = Layer.effect(
yield* shellLocks.withLock(input.sessionID)(
Effect.gen(function* () {
activeShells.add(input.sessionID)
if ((yield* execution.active).has(input.sessionID)) yield* execution.awaitIdle(input.sessionID)
yield* execution.awaitIdle(input.sessionID)
const started = yield* Effect.gen(function* () {
const shell = yield* Shell.Service
return yield* shell.create({ command: input.command, cwd: session.location.directory, timeout: 0 })
+2 -5
View File
@@ -14,7 +14,6 @@ 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"
import { Database } from "../../database/database"
import { EventV2 } from "../../event"
import { Location } from "../../location"
@@ -209,11 +208,10 @@ const layer = Layer.effect(
const toolFibers = yield* FiberSet.make<void, ToolOutputStore.Error>()
const ownedToolFibers: Array<Fiber.Fiber<void, ToolOutputStore.Error>> = []
let needsContinuation = false
const hookedRequest = request
// Automatic compaction completed; rebuild the request from compacted history.
if (
!(yield* SessionPending.compaction(db, session.id)) &&
(yield* compaction.compactIfNeeded({ sessionID: session.id, messages: context, request: hookedRequest }))
(yield* compaction.compactIfNeeded({ sessionID: session.id, messages: context, request }))
)
return { _tag: "RestartAfterCompaction", step: currentStep } as const
const startSnapshot = yield* snapshots.capture()
@@ -233,7 +231,7 @@ const layer = Layer.effect(
const serialized = <A, E, R>(effect: Effect.Effect<A, E, R>) => publication.withPermit(effect)
const publish = (event: LLMEvent, error?: SessionError.Error) => serialized(publisher.publish(event, error))
let overflowFailure: ProviderErrorEvent | undefined
const providerStream = llm.stream(hookedRequest).pipe(
const providerStream = llm.stream(request).pipe(
Stream.runForEach((event) =>
Effect.gen(function* () {
if (overflowFailure || publisher.hasProviderError()) return
@@ -594,7 +592,6 @@ export const node = makeLocationNode({
InstructionEntry.node,
SessionCompaction.node,
SessionTitle.node,
Config.node,
Snapshot.node,
Database.node,
PluginSupervisor.node,
@@ -57,13 +57,12 @@ const settledOutput = (value: ToolOutput | undefined, result: ToolResultValue):
}
/** Persist one step without executing tools or starting a continuation step. */
export const createLLMEventPublisher = (events: EventV2.Interface, input: Input) => {
export const createLLMEventPublisher = (events: Pick<EventV2.Interface, "publish">, input: Input) => {
const tools = new Map<
string,
{
readonly assistantMessageID: SessionMessage.ID
readonly name: string
inputEnded: boolean
called: boolean
settled: boolean
providerExecuted: boolean
@@ -140,7 +139,7 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
const flush = Effect.fnUntraced(function* () {
for (const id of chunks.keys()) yield* end(id)
})
return { start, append, end, flush }
return { start, append, end, flush, has: (id: string) => chunks.has(id) }
}
const text = fragments(
@@ -180,7 +179,6 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
callID,
text: value,
})
tool.inputEnded = true
}),
)
@@ -196,7 +194,6 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
tools.set(event.id, {
assistantMessageID,
name: event.name,
inputEnded: false,
called: false,
settled: false,
providerExecuted: false,
@@ -215,7 +212,7 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
if (!tool) return yield* Effect.die(new Error(`Tool input end before start: ${event.id}`))
if (tool.name !== event.name)
return yield* Effect.die(new Error(`Tool input name changed for ${event.id}: ${tool.name} -> ${event.name}`))
if (tool.inputEnded) return yield* Effect.die(new Error(`Duplicate tool input end: ${event.id}`))
if (!toolInput.has(event.id)) return yield* Effect.die(new Error(`Duplicate tool input end: ${event.id}`))
yield* toolInput.end(event.id)
})
@@ -330,7 +327,7 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
if (!tool) return yield* Effect.die(new Error(`Tool input delta before start: ${event.id}`))
if (tool.name !== event.name)
return yield* Effect.die(new Error(`Tool input name changed for ${event.id}: ${tool.name} -> ${event.name}`))
if (tool.inputEnded) return yield* Effect.die(new Error(`Tool input delta after end: ${event.id}`))
if (!toolInput.has(event.id)) return yield* Effect.die(new Error(`Tool input delta after end: ${event.id}`))
yield* toolInput.append(event.id, event.text)
yield* events.publish(SessionEvent.Tool.Input.Delta, {
sessionID: input.sessionID,
@@ -347,7 +344,7 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
retryEvidence = true
if (!tools.has(event.id)) yield* startToolInput(event)
const tool = tools.get(event.id)!
if (!tool.inputEnded) yield* endToolInput(event)
if (toolInput.has(event.id)) yield* endToolInput(event)
if (tool.name !== event.name)
return yield* Effect.die(new Error(`Tool call name changed for ${event.id}: ${tool.name} -> ${event.name}`))
if (tool.called) return yield* Effect.die(new Error(`Duplicate tool call: ${event.id}`))
@@ -364,7 +361,6 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
return
}
case "tool-result": {
retryEvidence = true
const tool = tools.get(event.id)
if (!tool?.called) return yield* Effect.die(new Error(`Tool result before call: ${event.id}`))
if (tool.name !== event.name)
@@ -401,7 +397,6 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
return
}
case "tool-error": {
retryEvidence = true
const tool = tools.get(event.id)
if (!tool?.called) return yield* Effect.die(new Error(`Tool error before call: ${event.id}`))
if (tool.name !== event.name)
@@ -1,5 +1,5 @@
import { expect, test } from "bun:test"
import { Effect, Schema, Stream } from "effect"
import { Effect, Schema } from "effect"
import { LLMEvent } from "@opencode-ai/llm"
import { Money } from "@opencode-ai/schema/money"
import { EventV2 } from "@opencode-ai/core/event"
@@ -16,7 +16,7 @@ const base64 = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAAB"
const capture = (providerMetadataKey = "anthropic") => {
const published: Array<{ readonly type: string; readonly data: unknown }> = []
const events = EventV2.Service.of({
const events: Pick<EventV2.Interface, "publish"> = {
publish: (definition, data) =>
Effect.sync(() => {
const event = { id: EventV2.ID.create(), type: definition.type, data } as EventV2.Payload<typeof definition>
@@ -28,16 +28,7 @@ const capture = (providerMetadataKey = "anthropic") => {
})
return event
}),
subscribe: () => Stream.empty,
log: () => Stream.empty,
sequences: () => Effect.succeed(new Map()),
listen: () => Effect.succeed(Effect.void),
project: () => Effect.void,
replay: () => Effect.void,
replayAll: () => Effect.succeed(undefined),
remove: () => Effect.void,
claim: () => Effect.void,
})
}
return {
published,
publisher: createLLMEventPublisher(events, {