feat(opencode): cut plugin system and remote behavior feeds; add SessionContext assembly and Runner
- delete plugin machinery (loader, install, meta, hooks) and all trigger sites - fold first-party provider auth into static registry (provider/hooks.ts) - remove remote instruction URL fetch, skills remote puller, TUI plugin host - add session/context.ts: single provenance-tagged context assembly point - add session/runner.ts: admit/context/stream/tools loop with compaction policy - relocate audit artifacts to audit/ (FORK-AUDIT, instruction docs as evidence)
This commit is contained in:
@@ -0,0 +1,266 @@
|
||||
export * as Runner from "./runner"
|
||||
|
||||
import { LayerNode } from "@opencode-ai/core/effect/layer-node"
|
||||
import { Context, Effect, Layer } from "effect"
|
||||
import { SessionV1 } from "@opencode-ai/core/v1/session"
|
||||
import { MessageID, PartID, SessionID } from "./schema"
|
||||
import { type Agent, Agent as Agents } from "@/agent/agent"
|
||||
import { Provider } from "@/provider/provider"
|
||||
import { Permission } from "@/permission"
|
||||
import { RuntimeFlags } from "@/effect/runtime-flags"
|
||||
import type { TaskPromptOps } from "@/tool/task"
|
||||
import { NotFoundError } from "@/storage/storage"
|
||||
import { Session } from "./session"
|
||||
import { SessionProcessor } from "./processor"
|
||||
import { SessionCompaction } from "./compaction"
|
||||
import { SessionTools } from "./tools"
|
||||
import { Truncate } from "@/tool/truncate"
|
||||
import { ToolRegistry } from "@/tool/registry"
|
||||
import { MCP } from "@/mcp"
|
||||
import { SessionContext } from "./context"
|
||||
import { MessageV2 } from "./message-v2"
|
||||
import { InstanceState } from "@/effect/instance-state"
|
||||
|
||||
/**
|
||||
* The agent runtime.
|
||||
*
|
||||
* Admit a durable user message, then loop: assemble context through
|
||||
* SessionContext (the only door text takes to a provider), stream one explicit
|
||||
* provider turn, persist parts, repeat until the model stops requesting tools.
|
||||
*
|
||||
* There is no other orchestration. Anything that needs to influence what the
|
||||
* model sees belongs in SessionContext or it does not happen.
|
||||
*/
|
||||
|
||||
// Hard backstop. A turn that still requests tools after this many steps is a
|
||||
// runaway, and runaways die loudly rather than quietly billing.
|
||||
const MAX_STEPS = 32
|
||||
|
||||
type PartInput =
|
||||
| SessionV1.TextPartInput
|
||||
| SessionV1.FilePartInput
|
||||
| SessionV1.AgentPartInput
|
||||
| SessionV1.SubtaskPartInput
|
||||
|
||||
type ModelRef = SessionV1.User["model"]
|
||||
|
||||
export interface Interface {
|
||||
/** Persist a user message with its parts, then run turns until done. */
|
||||
readonly prompt: (input: {
|
||||
sessionID: SessionID
|
||||
parts: readonly PartInput[]
|
||||
agent?: string
|
||||
model?: ModelRef
|
||||
}) => Effect.Effect<SessionV1.WithParts[], NotFoundError>
|
||||
/** Continue a session from its current transcript state. */
|
||||
readonly resume: (input: { sessionID: SessionID }) => Effect.Effect<SessionV1.WithParts[], NotFoundError>
|
||||
}
|
||||
|
||||
export class Service extends Context.Service<Service, Interface>()("@neuron/Runner") {}
|
||||
|
||||
const layer = Layer.effect(
|
||||
Service,
|
||||
Effect.gen(function* () {
|
||||
const sessions = yield* Session.Service
|
||||
const processor = yield* SessionProcessor.Service
|
||||
const permission = yield* Permission.Service
|
||||
const agents = yield* Agents.Service
|
||||
const provider = yield* Provider.Service
|
||||
const compaction = yield* SessionCompaction.Service
|
||||
const registry = yield* ToolRegistry.Service
|
||||
const mcp = yield* MCP.Service
|
||||
const truncate = yield* Truncate.Service
|
||||
const flags = yield* RuntimeFlags.Service
|
||||
const ctx = yield* InstanceState.context
|
||||
const sessionContext = yield* SessionContext.Service
|
||||
|
||||
// Subagent prompts route back through this same runner into a CHILD
|
||||
// session, so the task tool's transcript isolation holds. Command
|
||||
// expansion is intentionally absent: subagents get literal prompts.
|
||||
const ops = (parentSessionID: SessionID): TaskPromptOps => ({
|
||||
cancel: (sessionID) => sessions.touch(sessionID).pipe(Effect.asVoid),
|
||||
resolvePromptParts: (template) => Effect.succeed([{ type: "text", text: template }]),
|
||||
prompt: (input) =>
|
||||
Effect.flatMap(sessions.create({ parentID: parentSessionID }), (child) =>
|
||||
Effect.flatMap(prompt({ ...input, sessionID: child.id }), (msgs) =>
|
||||
Effect.sync(() => {
|
||||
const last = msgs.at(-1)
|
||||
if (!last) throw new Error("child session produced no messages")
|
||||
return last
|
||||
}),
|
||||
).pipe(Effect.catch((error) => Effect.die(new Error(`subagent run failed: ${String(error)}`)))),
|
||||
),
|
||||
})
|
||||
|
||||
function materialize(input: PartInput, messageID: ReturnType<typeof MessageID.ascending>, sessionID: SessionID) {
|
||||
const id = input.id ?? PartID.ascending()
|
||||
if (input.type === "text") {
|
||||
const part: SessionV1.TextPart = { ...input, id, sessionID, messageID }
|
||||
return part
|
||||
}
|
||||
if (input.type === "file") {
|
||||
const part: SessionV1.FilePart = { ...input, id, sessionID, messageID }
|
||||
return part
|
||||
}
|
||||
if (input.type === "agent") {
|
||||
const part: SessionV1.AgentPart = { ...input, id, sessionID, messageID }
|
||||
return part
|
||||
}
|
||||
const part: SessionV1.SubtaskPart = { ...input, id, sessionID, messageID }
|
||||
return part
|
||||
}
|
||||
|
||||
const prompt = Effect.fn("Runner.prompt")(function* (input: {
|
||||
sessionID: SessionID
|
||||
parts: readonly PartInput[]
|
||||
agent?: string
|
||||
model?: ModelRef
|
||||
}) {
|
||||
const agentName = input.agent ?? (yield* agents.defaultAgent().pipe(Effect.orDie))
|
||||
const modelRef = input.model ?? (yield* provider.defaultModel().pipe(Effect.orDie))
|
||||
const info: SessionV1.User = {
|
||||
id: MessageID.ascending(),
|
||||
sessionID: input.sessionID,
|
||||
time: { created: Date.now() },
|
||||
role: "user",
|
||||
agent: agentName,
|
||||
model: modelRef,
|
||||
}
|
||||
yield* sessions.updateMessage(info)
|
||||
for (const part of input.parts) {
|
||||
yield* sessions.updatePart(materialize(part, info.id, input.sessionID))
|
||||
}
|
||||
return yield* turns(input.sessionID)
|
||||
})
|
||||
|
||||
const turns = Effect.fn("Runner.turns")(function* (sessionID: SessionID) {
|
||||
const session = yield* sessions.get(sessionID).pipe(Effect.orDie)
|
||||
|
||||
for (let step = 1; ; step++) {
|
||||
const msgs = yield* sessions.messages({ sessionID })
|
||||
const latest = MessageV2.latest(msgs)
|
||||
if (!latest.user) throw new Error("no user message in session")
|
||||
|
||||
// Stop when the last assistant turn finished without pending tool calls.
|
||||
const lastAssistant = msgs.findLast(
|
||||
(msg): msg is SessionV1.WithParts & { info: SessionV1.Assistant } =>
|
||||
msg.info.role === "assistant" && msg.info.id === latest.assistant?.id,
|
||||
)
|
||||
const pendingTools =
|
||||
lastAssistant?.parts.some((part) => part.type === "tool" && !part.metadata?.providerExecuted) ?? false
|
||||
const finished =
|
||||
latest.assistant?.finish !== undefined &&
|
||||
!["tool-calls", "unknown"].includes(latest.assistant.finish) &&
|
||||
latest.assistant.parentID === latest.user.id
|
||||
if ((finished && !pendingTools) || step > MAX_STEPS) {
|
||||
if (step > MAX_STEPS) throw new Error(`runaway turn loop: exceeded ${MAX_STEPS} steps`)
|
||||
return yield* sessions.messages({ sessionID })
|
||||
}
|
||||
|
||||
const agentName = latest.user.agent
|
||||
const agent = yield* agents.get(agentName).pipe(Effect.orDie)
|
||||
|
||||
const modelRef = latest.user.model
|
||||
const model = yield* provider.getModel(modelRef.providerID, modelRef.modelID).pipe(Effect.orDie)
|
||||
|
||||
const msg: SessionV1.Assistant = {
|
||||
id: MessageID.ascending(),
|
||||
parentID: latest.user.id,
|
||||
role: "assistant",
|
||||
mode: agent.name,
|
||||
agent: agent.name,
|
||||
variant: modelRef.variant,
|
||||
path: { cwd: ctx.directory, root: ctx.worktree },
|
||||
cost: 0,
|
||||
tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } },
|
||||
modelID: model.id,
|
||||
providerID: model.providerID,
|
||||
time: { created: Date.now() },
|
||||
sessionID,
|
||||
}
|
||||
yield* sessions.updateMessage(msg)
|
||||
|
||||
const handle = yield* processor.create({ assistantMessage: msg, sessionID, model })
|
||||
|
||||
// Provenance tags ride on every block: the model sees where each piece
|
||||
// of its context came from. Unattributed text cannot exist here.
|
||||
const [blocks, messages, tools] = yield* Effect.all([
|
||||
sessionContext.system({ agent, modelID: model.api.id }),
|
||||
sessionContext.messages({ sessionID, model }),
|
||||
SessionTools.resolve({
|
||||
agent,
|
||||
session,
|
||||
model,
|
||||
processor: handle,
|
||||
bypassAgentCheck: false,
|
||||
messages: msgs,
|
||||
promptOps: ops(sessionID),
|
||||
}).pipe(
|
||||
Effect.provideService(Permission.Service, permission),
|
||||
Effect.provideService(ToolRegistry.Service, registry),
|
||||
Effect.provideService(MCP.Service, mcp),
|
||||
Effect.provideService(Truncate.Service, truncate),
|
||||
Effect.provideService(RuntimeFlags.Service, flags),
|
||||
),
|
||||
])
|
||||
|
||||
const result = yield* handle.process({
|
||||
user: latest.user,
|
||||
sessionID,
|
||||
agent,
|
||||
system: blocks.map((block) => `[${block.source}${block.origin ? `:${block.origin}` : ""}]\n${block.text}`),
|
||||
messages,
|
||||
tools,
|
||||
model,
|
||||
})
|
||||
|
||||
// Surface refusals as errors instead of silent empty turns.
|
||||
if (handle.message.finish === "content-filter" && !handle.message.error) {
|
||||
handle.message.error = new SessionV1.ContentFilterError({
|
||||
message: "The response was blocked by the provider's content filter",
|
||||
}).toObject()
|
||||
yield* sessions.updateMessage(handle.message)
|
||||
return yield* sessions.messages({ sessionID })
|
||||
}
|
||||
|
||||
if (result === "stop") return yield* sessions.messages({ sessionID })
|
||||
|
||||
// Overflow is a runner policy: hand the session to compaction and let
|
||||
// the loop re-evaluate from the compacted transcript.
|
||||
if (result === "compact") {
|
||||
yield* compaction.create({
|
||||
sessionID,
|
||||
agent: latest.user.agent,
|
||||
model: { providerID: model.providerID, modelID: model.id },
|
||||
auto: true,
|
||||
overflow: !handle.message.finish,
|
||||
})
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
const resume: Interface["resume"] = Effect.fn("Runner.resume")(function* (input) {
|
||||
return yield* turns(input.sessionID)
|
||||
})
|
||||
|
||||
return Service.of({ prompt, resume })
|
||||
}),
|
||||
)
|
||||
|
||||
export const node = LayerNode.make({
|
||||
service: Service,
|
||||
layer,
|
||||
deps: [
|
||||
Session.node,
|
||||
SessionProcessor.node,
|
||||
SessionCompaction.node,
|
||||
Permission.node,
|
||||
Agents.node,
|
||||
Provider.node,
|
||||
ToolRegistry.node,
|
||||
MCP.node,
|
||||
Truncate.node,
|
||||
RuntimeFlags.node,
|
||||
SessionContext.node,
|
||||
],
|
||||
})
|
||||
Reference in New Issue
Block a user