refactor(core): extract session context loading (#37203)
This commit is contained in:
Kit Langton
GitHub
parent
b4be28d4c3
commit
dc7e0f6d2e
3 changed files with 417 additions and 68 deletions
118
packages/core/src/session/context.ts
Normal file
118
packages/core/src/session/context.ts
Normal file
|
|
@ -0,0 +1,118 @@
|
|||
export * as SessionContext from "./context"
|
||||
|
||||
import { Context, Effect, Layer } from "effect"
|
||||
import { AgentV2 } from "../agent"
|
||||
import { Database } from "../database/database"
|
||||
import { makeLocationNode } from "../effect/app-node"
|
||||
import { InstructionDiscovery } from "../instruction-discovery"
|
||||
import { Instructions } from "../instructions/index"
|
||||
import { InstructionBuiltIns } from "../instructions/builtins"
|
||||
import { Location } from "../location"
|
||||
import { McpGuidance } from "../mcp/guidance"
|
||||
import { PluginSupervisor } from "../plugin/supervisor"
|
||||
import { ReferenceGuidance } from "../reference/guidance"
|
||||
import { SkillGuidance } from "../skill/guidance"
|
||||
import { AgentNotFoundError } from "./error"
|
||||
import { SessionHistory } from "./history"
|
||||
import { InstructionEntry } from "./instruction-entry"
|
||||
import { SessionMessage } from "./message"
|
||||
import { SessionRunnerModel } from "./runner/model"
|
||||
import { SessionSchema } from "./schema"
|
||||
import { SessionStore } from "./store"
|
||||
|
||||
export interface Selection {
|
||||
readonly session: SessionSchema.Info
|
||||
readonly agent: AgentV2.Selection & { readonly info: AgentV2.Info }
|
||||
readonly instructions: Instructions.Instructions
|
||||
}
|
||||
|
||||
export interface Loaded {
|
||||
readonly session: SessionSchema.Info
|
||||
readonly agent: AgentV2.Selection & { readonly info: AgentV2.Info }
|
||||
readonly model: SessionRunnerModel.Resolved
|
||||
readonly initial: string
|
||||
readonly messages: ReadonlyArray<SessionMessage.Info>
|
||||
}
|
||||
|
||||
export interface Interface {
|
||||
/** Resolves the Session, selected agent, and current instruction sources before durable Step preparation. */
|
||||
readonly select: (sessionID: SessionSchema.ID) => Effect.Effect<Selection, AgentNotFoundError>
|
||||
/** Loads the selected model and active history after instruction sync and pending-input promotion. */
|
||||
readonly load: (selection: Selection) => Effect.Effect<Loaded, SessionRunnerModel.Error>
|
||||
}
|
||||
|
||||
export class Service extends Context.Service<Service, Interface>()("@opencode/v2/SessionContext") {}
|
||||
|
||||
const layer = Layer.effect(
|
||||
Service,
|
||||
Effect.gen(function* () {
|
||||
const agents = yield* AgentV2.Service
|
||||
const builtins = yield* InstructionBuiltIns.Service
|
||||
const db = (yield* Database.Service).db
|
||||
const discovery = yield* InstructionDiscovery.Service
|
||||
const entries = yield* InstructionEntry.Service
|
||||
const location = yield* Location.Service
|
||||
const mcpGuidance = yield* McpGuidance.Service
|
||||
const models = yield* SessionRunnerModel.Service
|
||||
const plugins = yield* PluginSupervisor.Service
|
||||
const referenceGuidance = yield* ReferenceGuidance.Service
|
||||
const skillGuidance = yield* SkillGuidance.Service
|
||||
const store = yield* SessionStore.Service
|
||||
|
||||
const select = Effect.fn("SessionContext.select")(function* (sessionID: SessionSchema.ID) {
|
||||
const session = yield* store.get(sessionID)
|
||||
if (!session) return yield* Effect.die(new Error(`Session not found: ${sessionID}`))
|
||||
if (session.location.directory !== location.directory || session.location.workspaceID !== location.workspaceID)
|
||||
return yield* Effect.interrupt
|
||||
|
||||
yield* plugins.flush
|
||||
const agent = yield* agents.select(session.agent)
|
||||
if (!agent.info) return yield* new AgentNotFoundError({ sessionID: session.id, agent: session.agent ?? agent.id })
|
||||
const instructions = yield* Effect.all(
|
||||
[
|
||||
builtins.load(sessionID),
|
||||
discovery.load(),
|
||||
skillGuidance.load(agent),
|
||||
referenceGuidance.load(),
|
||||
mcpGuidance.load(agent),
|
||||
entries.load(sessionID),
|
||||
],
|
||||
{ concurrency: "unbounded" },
|
||||
).pipe(Effect.map(Instructions.combine))
|
||||
return { session, agent: { ...agent, info: agent.info }, instructions }
|
||||
})
|
||||
|
||||
const load = Effect.fn("SessionContext.load")(function* (selection: Selection) {
|
||||
const model = yield* models.resolve(selection.session)
|
||||
const history = yield* SessionHistory.entriesForRunner(db, selection.session.id, selection.instructions)
|
||||
return {
|
||||
session: selection.session,
|
||||
agent: selection.agent,
|
||||
model,
|
||||
initial: history.initial,
|
||||
messages: history.entries.map((entry) => entry.message),
|
||||
}
|
||||
})
|
||||
|
||||
return Service.of({ select, load })
|
||||
}),
|
||||
)
|
||||
|
||||
export const node = makeLocationNode({
|
||||
service: Service,
|
||||
layer,
|
||||
deps: [
|
||||
AgentV2.node,
|
||||
Database.node,
|
||||
InstructionBuiltIns.node,
|
||||
InstructionDiscovery.node,
|
||||
InstructionEntry.node,
|
||||
Location.node,
|
||||
McpGuidance.node,
|
||||
PluginSupervisor.node,
|
||||
ReferenceGuidance.node,
|
||||
SessionRunnerModel.node,
|
||||
SessionStore.node,
|
||||
SkillGuidance.node,
|
||||
],
|
||||
})
|
||||
|
|
@ -14,33 +14,23 @@ import type { SessionHooks } from "@opencode-ai/plugin/v2/effect/session"
|
|||
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 { Database } from "../../database/database"
|
||||
import { EventV2 } from "../../event"
|
||||
import { Location } from "../../location"
|
||||
import { ModelV2 } from "../../model"
|
||||
import { PermissionV2 } from "../../permission"
|
||||
import { Instructions } from "../../instructions/index"
|
||||
import { InstructionBuiltIns } from "../../instructions/builtins"
|
||||
import { InstructionDiscovery } from "../../instruction-discovery"
|
||||
import { SkillGuidance } from "../../skill/guidance"
|
||||
import { ReferenceGuidance } from "../../reference/guidance"
|
||||
import { McpGuidance } from "../../mcp/guidance"
|
||||
import { InstructionEntry } from "../instruction-entry"
|
||||
import { QuestionTool } from "../../tool/question"
|
||||
import { ToolRegistry } from "../../tool/registry"
|
||||
import { ToolOutputStore } from "../../tool-output-store"
|
||||
import { InstructionState } from "../instruction-state"
|
||||
import { SessionCompaction } from "../compaction"
|
||||
import { SessionContext } from "../context"
|
||||
import { SessionEvent } from "../event"
|
||||
import { SessionHistory } from "../history"
|
||||
import { SessionPending } from "../pending"
|
||||
import { SessionMessage } from "../message"
|
||||
import { SessionSchema } from "../schema"
|
||||
import { SessionStore } from "../store"
|
||||
import { SessionTitle } from "../title"
|
||||
import { Service } from "./index"
|
||||
import { SessionRunnerModel } from "./model"
|
||||
import { createLLMEventPublisher } from "./publish-llm-event"
|
||||
import { toLLMMessages } from "./to-llm-message"
|
||||
import { MAX_STEPS_PROMPT } from "./max-steps"
|
||||
|
|
@ -48,10 +38,9 @@ import PROMPT_DEFAULT from "./prompt/base.txt"
|
|||
import { Snapshot } from "../../snapshot"
|
||||
import { makeLocationNode } from "../../effect/app-node"
|
||||
import { llmClient } from "../../effect/app-node-platform"
|
||||
import { AgentNotFoundError, StepFailedError } from "../error"
|
||||
import { StepFailedError } from "../error"
|
||||
import { toSessionError } from "../to-session-error"
|
||||
import { SessionRunnerRetry } from "./retry"
|
||||
import { PluginSupervisor } from "../../plugin/supervisor"
|
||||
import { PluginHooks } from "../../plugin/hooks"
|
||||
import { SessionModelHeaders } from "../model-headers"
|
||||
|
||||
|
|
@ -89,23 +78,14 @@ const layer = Layer.effect(
|
|||
Effect.gen(function* () {
|
||||
const events = yield* EventV2.Service
|
||||
const llm = yield* LLMClient.Service
|
||||
const agents = yield* AgentV2.Service
|
||||
const tools = yield* ToolRegistry.Service
|
||||
const hooks = yield* PluginHooks.Service
|
||||
const models = yield* SessionRunnerModel.Service
|
||||
const store = yield* SessionStore.Service
|
||||
const location = yield* Location.Service
|
||||
const builtins = yield* InstructionBuiltIns.Service
|
||||
const discovery = yield* InstructionDiscovery.Service
|
||||
const skillGuidance = yield* SkillGuidance.Service
|
||||
const referenceGuidance = yield* ReferenceGuidance.Service
|
||||
const mcpGuidance = yield* McpGuidance.Service
|
||||
const entries = yield* InstructionEntry.Service
|
||||
const context = yield* SessionContext.Service
|
||||
const snapshots = yield* Snapshot.Service
|
||||
const db = (yield* Database.Service).db
|
||||
const compaction = yield* SessionCompaction.Service
|
||||
const title = yield* SessionTitle.Service
|
||||
const plugins = yield* PluginSupervisor.Service
|
||||
// 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.
|
||||
|
|
@ -116,9 +96,6 @@ const layer = Layer.effect(
|
|||
if (!session) return yield* Effect.die(new Error(`Session not found: ${sessionID}`))
|
||||
return session
|
||||
})
|
||||
const isCurrentLocation = (session: SessionSchema.Info) =>
|
||||
session.location.directory === location.directory && session.location.workspaceID === location.workspaceID
|
||||
|
||||
const failInterruptedTools = Effect.fn("SessionRunner.failInterruptedTools")(function* (
|
||||
sessionID: SessionSchema.ID,
|
||||
) {
|
||||
|
|
@ -145,19 +122,6 @@ const layer = Layer.effect(
|
|||
(reason.defect instanceof PermissionV2.DeclinedError || reason.defect instanceof QuestionTool.CancelledError),
|
||||
)
|
||||
|
||||
const loadInstructions = (agent: AgentV2.Selection, sessionID: SessionSchema.ID) =>
|
||||
Effect.all(
|
||||
[
|
||||
builtins.load(sessionID),
|
||||
discovery.load(),
|
||||
skillGuidance.load(agent),
|
||||
referenceGuidance.load(),
|
||||
mcpGuidance.load(agent),
|
||||
entries.load(sessionID),
|
||||
],
|
||||
{ concurrency: "unbounded" },
|
||||
).pipe(Effect.map(Instructions.combine))
|
||||
|
||||
const attemptStep = Effect.fn("SessionRunner.attemptStep")(function* (
|
||||
sessionID: SessionSchema.ID,
|
||||
promotion: SessionPending.Delivery | undefined,
|
||||
|
|
@ -165,32 +129,28 @@ const layer = Layer.effect(
|
|||
recoverOverflow?: typeof compaction.compact,
|
||||
assistantMessageID?: SessionMessage.ID,
|
||||
) {
|
||||
const session = yield* getSession(sessionID)
|
||||
if (!isCurrentLocation(session)) return yield* Effect.interrupt
|
||||
yield* plugins.flush
|
||||
const agent = yield* agents.select(session.agent)
|
||||
const agentInfo = agent.info
|
||||
if (!agentInfo) return yield* new AgentNotFoundError({ sessionID: session.id, agent: session.agent ?? agent.id })
|
||||
const selected = yield* context.select(sessionID)
|
||||
// Establish what the model knows before admitting what the user said, so
|
||||
// a blocked first step leaves pending inputs untouched.
|
||||
const instructions = yield* loadInstructions(agent, session.id)
|
||||
yield* InstructionState.prepare(db, events, instructions, session.id)
|
||||
yield* InstructionState.prepare(db, events, selected.instructions, selected.session.id)
|
||||
let currentStep = step
|
||||
if (promotion) {
|
||||
let promoted = 0
|
||||
if (promotion === "steer") promoted = yield* SessionPending.promoteSteers(db, events, session.id)
|
||||
if (promotion === "steer") promoted = yield* SessionPending.promoteSteers(db, events, selected.session.id)
|
||||
if (promotion === "queue") {
|
||||
promoted += Number(yield* SessionPending.promoteNextQueued(db, events, session.id))
|
||||
promoted += yield* SessionPending.promoteSteers(db, events, session.id)
|
||||
promoted += Number(yield* SessionPending.promoteNextQueued(db, events, selected.session.id))
|
||||
promoted += yield* SessionPending.promoteSteers(db, events, selected.session.id)
|
||||
}
|
||||
if (promoted > 0) currentStep = 1
|
||||
}
|
||||
const resolved = yield* models.resolve(session)
|
||||
const loaded = yield* context.load(selected)
|
||||
const session = loaded.session
|
||||
const agent = loaded.agent
|
||||
const agentInfo = agent.info
|
||||
const resolved = loaded.model
|
||||
const model = resolved.model
|
||||
const providerMetadataKey = model.route.providerMetadataKey ?? model.provider
|
||||
const history = yield* SessionHistory.entriesForRunner(db, session.id, instructions)
|
||||
const context = history.entries.map((entry) => entry.message)
|
||||
const compactionInput = { session, messages: context, model }
|
||||
const compactionInput = { session, messages: loaded.messages, model }
|
||||
if (compaction.required(compactionInput) && !(yield* SessionPending.compaction(db, session.id))) {
|
||||
const compacted = yield* compaction.compact(compactionInput)
|
||||
if (compacted.status === "completed") return { _tag: "RestartAfterCompaction", step: currentStep } as const
|
||||
|
|
@ -205,11 +165,11 @@ const layer = Layer.effect(
|
|||
headers: SessionModelHeaders.make(session),
|
||||
},
|
||||
providerOptions: { openai: { promptCacheKey } },
|
||||
system: [agentInfo.system ? agentInfo.system : PROMPT_DEFAULT, history.initial]
|
||||
system: [agentInfo.system ? agentInfo.system : PROMPT_DEFAULT, loaded.initial]
|
||||
.filter((part): part is string => part !== undefined && part.length > 0)
|
||||
.map(SystemPart.make),
|
||||
messages: [
|
||||
...toLLMMessages(context, resolved.ref, providerMetadataKey),
|
||||
...toLLMMessages(loaded.messages, resolved.ref, providerMetadataKey),
|
||||
...(isLastStep ? [Message.assistant(MAX_STEPS_PROMPT)] : []),
|
||||
],
|
||||
tools: toolMaterialization?.definitions ?? [],
|
||||
|
|
@ -366,8 +326,7 @@ const layer = Layer.effect(
|
|||
recoverOverflow &&
|
||||
!publisher.hasRetryEvidence() &&
|
||||
isContextOverflowFailure(overflowFailure ?? streamFailure) &&
|
||||
(yield* restore(recoverOverflow({ session, messages: context, model }))).status ===
|
||||
"completed"
|
||||
(yield* restore(recoverOverflow({ session, messages: loaded.messages, model }))).status === "completed"
|
||||
)
|
||||
return { _tag: "RestartAfterOverflowCompaction", step: currentStep } as const
|
||||
|
||||
|
|
@ -612,22 +571,13 @@ export const node = makeLocationNode({
|
|||
deps: [
|
||||
EventV2.node,
|
||||
llmClient,
|
||||
AgentV2.node,
|
||||
ToolRegistry.node,
|
||||
PluginHooks.node,
|
||||
SessionRunnerModel.node,
|
||||
SessionContext.node,
|
||||
SessionStore.node,
|
||||
Location.node,
|
||||
InstructionBuiltIns.node,
|
||||
InstructionDiscovery.node,
|
||||
SkillGuidance.node,
|
||||
ReferenceGuidance.node,
|
||||
McpGuidance.node,
|
||||
InstructionEntry.node,
|
||||
SessionCompaction.node,
|
||||
SessionTitle.node,
|
||||
Snapshot.node,
|
||||
Database.node,
|
||||
PluginSupervisor.node,
|
||||
],
|
||||
})
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue