refactor(core): align runner naming with step vocabulary (#35227)
This commit is contained in:
Kit Langton
GitHub
parent
e66cbf36e9
commit
5d14c7a185
14 changed files with 47 additions and 50 deletions
|
|
@ -18,7 +18,7 @@ const decodeApplied = Schema.decodeUnknownOption(SystemContext.Applied)
|
|||
* Loads or creates the session's durable context checkpoint, narrating any
|
||||
* drift since the model was last told as a chronological update. Completed
|
||||
* compaction rebaselines; nothing else rewrites the baseline. Runs before
|
||||
* input promotion so a blocked first turn leaves pending inputs untouched.
|
||||
* input promotion so a blocked first step leaves pending inputs untouched.
|
||||
*/
|
||||
export const prepare = Effect.fn("SessionContextCheckpoint.prepare")(function* (
|
||||
db: DatabaseService,
|
||||
|
|
|
|||
|
|
@ -20,7 +20,7 @@ const layer = Layer.effect(
|
|||
drain: Effect.fnUntraced(function* (sessionID: SessionSchema.ID, force) {
|
||||
const session = yield* store.get(sessionID)
|
||||
if (!session) return yield* Effect.die(new Error(`Session not found: ${sessionID}`))
|
||||
return yield* SessionRunner.Service.use((runner) => runner.run({ sessionID, force })).pipe(
|
||||
return yield* SessionRunner.Service.use((runner) => runner.drain({ sessionID, force })).pipe(
|
||||
Effect.provide(locations.get(session.location)),
|
||||
Effect.tapCause((cause) =>
|
||||
Cause.hasInterruptsOnly(cause)
|
||||
|
|
|
|||
|
|
@ -34,7 +34,7 @@ const messageRows = Effect.fnUntraced(function* (
|
|||
and(
|
||||
eq(SessionMessageTable.session_id, sessionID),
|
||||
// Keep system updates visible in the gap between a completed compaction
|
||||
// and the next prepared turn's rebaseline, when their content is not yet
|
||||
// and the next prepared step's rebaseline, when their content is not yet
|
||||
// folded into a new baseline.
|
||||
compaction
|
||||
? or(
|
||||
|
|
|
|||
|
|
@ -36,9 +36,9 @@ const layer = Layer.effect(
|
|||
// absolute paths, but the human-facing description shows paths relative to the project
|
||||
// root so opening a subdirectory still describes paths from the project root.
|
||||
const root = yield* fs.resolve(location.project.directory)
|
||||
// Same-turn parallel reads settle concurrently, so an in-memory claim guards each
|
||||
// Same-step parallel reads settle concurrently, so an in-memory claim guards each
|
||||
// Session/path pair before any filesystem work. The durable history check below covers
|
||||
// paths injected in earlier turns after this Location layer was reopened.
|
||||
// paths injected in earlier steps after this Location layer was reopened.
|
||||
const injected = yield* Ref.make<Map<SessionSchema.ID, Set<string>>>(new Map())
|
||||
|
||||
const load = Effect.fn("SessionInstructions.load")(function* (input: {
|
||||
|
|
|
|||
|
|
@ -19,7 +19,7 @@ export interface Adapter {
|
|||
export function memory(state: MemoryState): Adapter {
|
||||
const assistantIndex = (messageID: SessionMessage.ID) =>
|
||||
state.messages.findLastIndex((message) => message.id === messageID)
|
||||
// A newer turn supersedes stale incomplete rows; never resume an older assistant projection.
|
||||
// A newer step supersedes stale incomplete rows; never resume an older assistant projection.
|
||||
const latestAssistantIndex = () => state.messages.findLastIndex((message) => message.type === "assistant")
|
||||
const activeShellIndex = (callID: string) =>
|
||||
state.messages.findLastIndex((message) => message.type === "shell" && message.callID === callID)
|
||||
|
|
|
|||
|
|
@ -168,7 +168,8 @@ const projectFork = Effect.fn("SessionProjector.projectFork")(function* (
|
|||
.get()
|
||||
.pipe(Effect.orDie)
|
||||
: undefined
|
||||
if (event.data.from && !boundary) return yield* Effect.die(new Error(`Fork boundary message not found: ${event.data.from}`))
|
||||
if (event.data.from && !boundary)
|
||||
return yield* Effect.die(new Error(`Fork boundary message not found: ${event.data.from}`))
|
||||
const copied = yield* db
|
||||
.select({ seq: SessionMessageTable.seq })
|
||||
.from(SessionMessageTable)
|
||||
|
|
@ -357,7 +358,7 @@ function run(db: DatabaseService, event: MessageEvent) {
|
|||
const adapter: SessionMessageUpdater.Adapter = {
|
||||
getCurrentAssistant() {
|
||||
return Effect.gen(function* () {
|
||||
// A newer turn supersedes stale incomplete rows; never resume an older assistant projection.
|
||||
// A newer step supersedes stale incomplete rows; never resume an older assistant projection.
|
||||
const row = yield* db
|
||||
.select()
|
||||
.from(SessionMessageTable)
|
||||
|
|
|
|||
|
|
@ -9,16 +9,12 @@ import type { SystemContext } from "../../system-context/index"
|
|||
import type { ToolOutputStore } from "../../tool-output-store"
|
||||
|
||||
export type RunError =
|
||||
| LLMError
|
||||
| SessionRunnerModel.Error
|
||||
| MessageDecodeError
|
||||
| SystemContext.InitializationBlocked
|
||||
| ToolOutputStore.Error
|
||||
LLMError | SessionRunnerModel.Error | MessageDecodeError | SystemContext.InitializationBlocked | ToolOutputStore.Error
|
||||
|
||||
/** Runs one local continuation from already-recorded Session history. */
|
||||
export interface Interface {
|
||||
/** Drains eligible durable work. Explicit runs perform one provider attempt even when no work is eligible. */
|
||||
readonly run: (input: {
|
||||
/** Drains eligible durable work. Explicit runs perform one physical attempt even when no work is eligible. */
|
||||
readonly drain: (input: {
|
||||
readonly sessionID: SessionSchema.ID
|
||||
readonly force: boolean
|
||||
}) => Effect.Effect<void, RunError>
|
||||
|
|
|
|||
|
|
@ -61,11 +61,11 @@ import { llmClient } from "../../effect/app-node-platform"
|
|||
* - Runtime context assembly
|
||||
* - Track V1 runtime-context parity canonically in `specs/v2/session.md`.
|
||||
*
|
||||
* - One provider turn
|
||||
* - One step
|
||||
* - [x] Translate every projected V2 Session message variant into canonical
|
||||
* `@opencode-ai/llm` messages.
|
||||
* - [ ] Resolve policy-filtered built-in, MCP, plugin, and structured-output tool definitions.
|
||||
* - [x] Stream exactly one `llm.stream(request)` provider turn.
|
||||
* - [x] Stream exactly one `llm.stream(request)` physical attempt.
|
||||
* - [x] Persist assistant text and usage events incrementally as they arrive.
|
||||
* - [ ] Persist snapshots, patches, and retry notices incrementally as they arrive.
|
||||
* - [x] Persist reasoning, provider errors, and tool-call events incrementally as they arrive.
|
||||
|
|
@ -77,8 +77,8 @@ import { llmClient } from "../../effect/app-node-platform"
|
|||
* - [x] Start each recorded local call eagerly and await all settlements before continuation.
|
||||
* - [ ] Add scoped runtime context, progress updates, attachment normalization,
|
||||
* plugins, and cancellation settlement.
|
||||
* - [x] Reload projected history and start the next explicit provider turn after local tool results.
|
||||
* - [x] Continue for durable user steering accepted during an active provider turn.
|
||||
* - [x] Reload projected history and start the next explicit step after local tool results.
|
||||
* - [x] Continue for durable user steering accepted during an active step.
|
||||
* - [ ] Continue for compaction or another continuation condition when required.
|
||||
*
|
||||
* - Post-run maintenance
|
||||
|
|
@ -86,12 +86,12 @@ import { llmClient } from "../../effect/app-node-platform"
|
|||
* - [ ] Coalesce streamed deltas and add covering projected-history indexes.
|
||||
* - [ ] Update title, summaries, compaction state, and cleanup in bounded background work.
|
||||
*
|
||||
* Use `llm.stream(request)` for each provider turn. Keep tool execution and continuation here.
|
||||
* Use `llm.stream(request)` for each physical attempt. Keep tool execution and continuation here.
|
||||
* Durable continuation recovery remains a separate future slice with an explicit retry policy.
|
||||
*
|
||||
* The current slice loads V2 history, translates it, resolves a model through a core service, and persists one
|
||||
* provider turn. Registry definitions are advertised, local tool calls are settled durably, and an
|
||||
* explicit loop starts the next provider turn after local settlement. Configured agent step limits bound the loop.
|
||||
* step. Registry definitions are advertised, local tool calls are settled durably, and an
|
||||
* explicit loop starts the next step after local settlement. Configured agent step limits bound the loop.
|
||||
*/
|
||||
|
||||
const layer = Layer.effect(
|
||||
|
|
@ -114,7 +114,7 @@ const layer = Layer.effect(
|
|||
const db = (yield* Database.Service).db
|
||||
const compaction = yield* SessionCompaction.Service
|
||||
const title = yield* SessionTitle.Service
|
||||
// Title generation is a side effect of the first turn; it must not delay turn continuation.
|
||||
// Title generation is a side effect of the first step; it must not delay step continuation.
|
||||
// Tracked per process so repeated wakes before the second user message arrives don't
|
||||
// re-fire a redundant LLM call; `SessionTitle` itself is idempotent based on durable history.
|
||||
const titleAttempted = new Set<SessionSchema.ID>()
|
||||
|
|
@ -166,7 +166,7 @@ const layer = Layer.effect(
|
|||
{ concurrency: "unbounded" },
|
||||
).pipe(Effect.map(SystemContext.combine))
|
||||
|
||||
const runTurnAttempt = Effect.fn("SessionRunner.runTurnAttempt")(function* (
|
||||
const attemptStep = Effect.fn("SessionRunner.attemptStep")(function* (
|
||||
sessionID: SessionSchema.ID,
|
||||
promotion: SessionInput.Delivery | undefined,
|
||||
step: number,
|
||||
|
|
@ -177,7 +177,7 @@ const layer = Layer.effect(
|
|||
return yield* Effect.interrupt
|
||||
const agent = yield* agents.select(session.agent)
|
||||
// Establish what the model knows before admitting what the user said, so
|
||||
// a blocked first turn leaves pending inputs untouched.
|
||||
// a blocked first step leaves pending inputs untouched.
|
||||
const checkpoint = yield* SessionContextCheckpoint.prepare(
|
||||
db,
|
||||
events,
|
||||
|
|
@ -231,7 +231,7 @@ const layer = Layer.effect(
|
|||
snapshot: startSnapshot,
|
||||
})
|
||||
const publication = Semaphore.makeUnsafe(1)
|
||||
// Durable publishes are serialized so tool fibers and turn settlement never interleave
|
||||
// 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> = []) =>
|
||||
|
|
@ -282,7 +282,7 @@ const layer = Layer.effect(
|
|||
Effect.ensuring(serialized(publisher.flush())),
|
||||
)
|
||||
|
||||
// Captures the end snapshot, diffs it against the turn's start, and durably ends the
|
||||
// Captures the end snapshot, diffs it against the step's start, and durably ends the
|
||||
// assistant step.
|
||||
const publishStepEnd = (settlement: NonNullable<ReturnType<typeof publisher.stepSettlement>>) =>
|
||||
Effect.gen(function* () {
|
||||
|
|
@ -316,7 +316,7 @@ const layer = Layer.effect(
|
|||
const streamInterrupted = stream._tag === "Failure" && Cause.hasInterrupts(stream.cause)
|
||||
|
||||
// A context overflow before any assistant output is recoverable: compact and
|
||||
// restart the turn instead of surfacing the provider error.
|
||||
// restart the step instead of surfacing the provider error.
|
||||
if (
|
||||
recoverOverflow &&
|
||||
!publisher.hasAssistantStarted() &&
|
||||
|
|
@ -325,7 +325,7 @@ const layer = Layer.effect(
|
|||
)
|
||||
return { _tag: "RestartAfterOverflowCompaction", step: currentStep } as const
|
||||
|
||||
// An unrecovered held-back overflow becomes the turn's durable provider error. A
|
||||
// An unrecovered held-back overflow becomes the step's durable provider error. A
|
||||
// thrown LLM failure fails hosted tool calls and the assistant unless a provider
|
||||
// error was already recorded from the stream.
|
||||
if (overflowFailure) yield* publish(overflowFailure)
|
||||
|
|
@ -346,12 +346,12 @@ const layer = Layer.effect(
|
|||
if (questionDismissed || streamInterrupted || toolsInterrupted) {
|
||||
yield* FiberSet.clear(toolFibers)
|
||||
yield* serialized(publisher.failUnsettledTools("Tool execution interrupted"))
|
||||
yield* serialized(publisher.failAssistant("Provider turn interrupted"))
|
||||
yield* serialized(publisher.failAssistant("Step interrupted"))
|
||||
// Match V1: dismissing a question halts the loop like an interruption.
|
||||
if (questionDismissed) return yield* Effect.interrupt
|
||||
}
|
||||
// A settled tool fiber failure is one of two things. A defect from a tool
|
||||
// implementation becomes a failed tool call the model can read, and the turn still
|
||||
// implementation becomes a failed tool call the model can read, and the step still
|
||||
// settles so the model may recover. A typed infrastructure failure (tool output
|
||||
// could not be persisted) also fails the assistant and then fails the drain.
|
||||
const settledFailure = settled._tag === "Failure" && !toolsInterrupted ? settled.cause : undefined
|
||||
|
|
@ -387,7 +387,7 @@ const layer = Layer.effect(
|
|||
)
|
||||
}, Effect.scoped)
|
||||
|
||||
const runTurn = Effect.fnUntraced(function* (
|
||||
const runStep = Effect.fnUntraced(function* (
|
||||
sessionID: SessionSchema.ID,
|
||||
promotion: SessionInput.Delivery | undefined,
|
||||
step: number,
|
||||
|
|
@ -399,7 +399,7 @@ const layer = Layer.effect(
|
|||
let currentPromotion = promotion
|
||||
let currentStep = step
|
||||
while (true) {
|
||||
const attempt = yield* runTurnAttempt(sessionID, currentPromotion, currentStep, recoverOverflow)
|
||||
const attempt = yield* attemptStep(sessionID, currentPromotion, currentStep, recoverOverflow)
|
||||
if (attempt._tag === "Completed") return { needsContinuation: attempt.needsContinuation, step: attempt.step }
|
||||
if (attempt._tag === "RestartAfterOverflowCompaction") recoverOverflow = undefined
|
||||
yield* Effect.yieldNow
|
||||
|
|
@ -410,7 +410,7 @@ const layer = Layer.effect(
|
|||
|
||||
// ExecutionSettled is published per execution (busy period) by SessionExecution, not per
|
||||
// drain here.
|
||||
const run = Effect.fn("SessionRunner.run")(function* (input: {
|
||||
const drain = Effect.fn("SessionRunner.drain")(function* (input: {
|
||||
readonly sessionID: SessionSchema.ID
|
||||
readonly force: boolean
|
||||
}) {
|
||||
|
|
@ -428,8 +428,8 @@ const layer = Layer.effect(
|
|||
// a provider error suppresses it. Pending steers also continue the loop so
|
||||
// interjections are answered before the session goes idle.
|
||||
while (needsContinuation) {
|
||||
const result = yield* runTurn(input.sessionID, promotion, step)
|
||||
// Steer/queue promotion inside runTurn has already made the pending input a visible
|
||||
const result = yield* runStep(input.sessionID, promotion, step)
|
||||
// Steer/queue promotion inside runStep has already made the pending input a visible
|
||||
// user message by this point, so the first-user-message check below is reliable.
|
||||
if (!titleAttempted.has(input.sessionID)) {
|
||||
titleAttempted.add(input.sessionID)
|
||||
|
|
@ -445,7 +445,7 @@ const layer = Layer.effect(
|
|||
}
|
||||
})
|
||||
|
||||
return Service.of({ run })
|
||||
return Service.of({ drain })
|
||||
}),
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -50,7 +50,7 @@ const settledOutput = (value: ToolOutput | undefined, result: ToolResultValue):
|
|||
return { structured: record(settled.structured), content: settled.content }
|
||||
}
|
||||
|
||||
/** Persist one provider turn without executing tools or starting a continuation turn. */
|
||||
/** Persist one step without executing tools or starting a continuation step. */
|
||||
export const createLLMEventPublisher = (events: EventV2.Interface, input: Input) => {
|
||||
const tools = new Map<
|
||||
string,
|
||||
|
|
|
|||
|
|
@ -12,7 +12,7 @@ import { Effect, Option, Schema } from "effect"
|
|||
* The durable `Applied` record tracks what the model was last told, per source:
|
||||
* it is the model's current belief. Interpreters uphold one invariant —
|
||||
* `reconcile` never rewrites the baseline; it only narrates drift as update
|
||||
* text. Only `rebaseline` (compaction) and `initialize` (first turn) produce
|
||||
* text. Only `rebaseline` (compaction) and `initialize` (first step) produce
|
||||
* baseline text.
|
||||
*
|
||||
* Returning `unavailable` means observation failed temporarily. It differs from
|
||||
|
|
|
|||
|
|
@ -354,7 +354,7 @@ Recent work
|
|||
state: SessionMessage.ToolStateError.make({
|
||||
status: "error",
|
||||
input: { query: "Effect" },
|
||||
error: { type: "unknown", message: "Provider turn interrupted" },
|
||||
error: { type: "unknown", message: "Step interrupted" },
|
||||
content: [],
|
||||
structured: {},
|
||||
}),
|
||||
|
|
@ -362,7 +362,7 @@ Recent work
|
|||
}),
|
||||
],
|
||||
finish: "error",
|
||||
error: { type: "unknown", message: "Provider turn interrupted" },
|
||||
error: { type: "unknown", message: "Step interrupted" },
|
||||
time: { created, completed: created },
|
||||
}),
|
||||
],
|
||||
|
|
@ -386,7 +386,7 @@ Recent work
|
|||
result: {
|
||||
type: "error",
|
||||
value: {
|
||||
error: { type: "unknown", message: "Provider turn interrupted" },
|
||||
error: { type: "unknown", message: "Step interrupted" },
|
||||
content: [],
|
||||
structured: {},
|
||||
},
|
||||
|
|
|
|||
|
|
@ -98,7 +98,7 @@ const execution = Layer.effect(
|
|||
Effect.gen(function* () {
|
||||
const sessionRunner = yield* SessionRunner.Service
|
||||
const coordinator = yield* SessionRunCoordinator.make<SessionV2.ID, SessionRunner.RunError>({
|
||||
drain: (sessionID, force) => sessionRunner.run({ sessionID, force }),
|
||||
drain: (sessionID, force) => sessionRunner.drain({ sessionID, force }),
|
||||
})
|
||||
return SessionExecution.Service.of({
|
||||
active: coordinator.active,
|
||||
|
|
|
|||
|
|
@ -260,7 +260,7 @@ const execution = Layer.effect(
|
|||
Effect.gen(function* () {
|
||||
const sessionRunner = yield* SessionRunner.Service
|
||||
const coordinator = yield* SessionRunCoordinator.make<SessionV2.ID, SessionRunner.RunError>({
|
||||
drain: (sessionID, force) => sessionRunner.run({ sessionID, force }),
|
||||
drain: (sessionID, force) => sessionRunner.drain({ sessionID, force }),
|
||||
})
|
||||
return SessionExecution.Service.of({
|
||||
active: coordinator.active,
|
||||
|
|
@ -575,7 +575,7 @@ const verifyPartialFlushOnInterruption = (kind: FragmentKind) =>
|
|||
)
|
||||
|
||||
const runner = yield* SessionRunner.Service
|
||||
const fiber = yield* runner.run({ sessionID, force: true }).pipe(Effect.forkChild)
|
||||
const fiber = yield* runner.drain({ sessionID, force: true }).pipe(Effect.forkChild)
|
||||
yield* Deferred.await(streamed)
|
||||
yield* Fiber.interrupt(fiber)
|
||||
expect(yield* session.context(sessionID)).toMatchObject([
|
||||
|
|
@ -583,7 +583,7 @@ const verifyPartialFlushOnInterruption = (kind: FragmentKind) =>
|
|||
{
|
||||
type: "assistant",
|
||||
finish: "error",
|
||||
error: { type: "unknown", message: "Provider turn interrupted" },
|
||||
error: { type: "unknown", message: "Step interrupted" },
|
||||
content: [
|
||||
kind === "tool input"
|
||||
? { type: "tool", id: fragmentID(kind, "interrupted"), state: { status: "error" } }
|
||||
|
|
@ -2983,7 +2983,7 @@ describe("SessionRunnerLLM", () => {
|
|||
expect(requests).toHaveLength(1)
|
||||
expect(yield* session.context(sessionID)).toMatchObject([
|
||||
{ type: "user", text: "Interrupt provider" },
|
||||
{ type: "assistant", finish: "error", error: { type: "unknown", message: "Provider turn interrupted" } },
|
||||
{ type: "assistant", finish: "error", error: { type: "unknown", message: "Step interrupted" } },
|
||||
])
|
||||
expect(yield* recordedEventTypes(sessionID)).toContain("step.failed.1")
|
||||
yield* session.interrupt(sessionID)
|
||||
|
|
@ -3007,7 +3007,7 @@ describe("SessionRunnerLLM", () => {
|
|||
]
|
||||
|
||||
const runner = yield* SessionRunner.Service
|
||||
const run = yield* runner.run({ sessionID, force: true }).pipe(Effect.forkChild)
|
||||
const run = yield* runner.drain({ sessionID, force: true }).pipe(Effect.forkChild)
|
||||
yield* Deferred.await(toolExecutionsStarted)
|
||||
yield* Fiber.interrupt(run)
|
||||
toolExecutionGate = undefined
|
||||
|
|
@ -3018,7 +3018,7 @@ describe("SessionRunnerLLM", () => {
|
|||
{
|
||||
type: "assistant",
|
||||
finish: "error",
|
||||
error: { type: "unknown", message: "Provider turn interrupted" },
|
||||
error: { type: "unknown", message: "Step interrupted" },
|
||||
content: [
|
||||
{
|
||||
type: "tool",
|
||||
|
|
|
|||
|
|
@ -42,7 +42,7 @@ Execution routing starts from only the Session ID:
|
|||
SessionExecution.resume(sessionID)
|
||||
-> SessionStore.get(sessionID)
|
||||
-> LocationServiceMap.get(session.location)
|
||||
-> SessionRunner.run({ sessionID, force? })
|
||||
-> SessionRunner.drain({ sessionID, force? })
|
||||
```
|
||||
|
||||
`SessionExecution` and the read-side `SessionStore` are process-global. `SessionRunner`, catalog, model resolver, tool registry, permission state, and filesystem are cached per Location. No layer takes a Session ID. An omitted `Location.workspaceID` means implicit-local placement; explicit workspace identity remains reserved for future placement semantics.
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue